Changeset 10268

Timestamp:

03/30/11 15:39:31 (14 years ago)

Author:

tbretz

Message:

Some improvements to the connection handling, added some PostMessage functions which automatically convert to network byte order.

Location:

trunk/FACT++/src

Files:

• Connection.cc (modified) (11 diffs)
• Connection.h (modified) (4 diffs)

trunk/FACT++/src/Connection.cc

@@ -31 +31 @@
 }
 
-void Connection::AsyncRead(ba::mutable_buffers_1 buffers)
+void Connection::AsyncRead(ba::mutable_buffers_1 buffers, int type)
 {
     ba::async_read(*this, buffers,
                    boost::bind(&Connection::HandleReceivedData, this,
-                               dummy::error, dummy::bytes_transferred));
+                               dummy::error, dummy::bytes_transferred, type));
 }
 
@@ -89 +89 @@
     }
 
-    // Close the connection
+    // Stop any pending connection attempt
+    fConnectionTimer.cancel();
+
+    // Close possible open connections
+    close();
+
+    // Reset the connection status
     fConnectionStatus = 0;
-    close();
 
     // Stop deadline counters
     fInTimeout.cancel();
     fOutTimeout.cancel();
-    //fConnectTimeout.cancel();
 
     if (!restart || IsConnecting())
@@ -106 +110 @@
 
     // Start trying to reconnect
-    AsyncConnect();
+    fMsgConnect = "";
+    StartConnect();
 }
 
@@ -201 +206 @@
 }
 
+void Connection::PostMessage(vector<uint16_t> inp)
+{
+    // Convert to network byte order
+    for_each(inp.begin(), inp.end(), htons);
+
+    // FIXME FIXME
+
+    PostMessage((char*)&inp.front(), sizeof(uint16_t)*inp.size());
+
+//    const vector<char> msg((char*)&inp.front(), (char*)&inp.last()+1);
+//    get_io_service().post(boost::bind(&Connection::SendMessageImp, this, msg));
+}
+
 void Connection::PostMessage(const vector<char> &msg)
 {
+    get_io_service().post(boost::bind(&Connection::SendMessageImp, this, msg));
+}
+
+void Connection::PostMessage(const void *ptr, size_t max)
+{
+    const vector<char> msg(reinterpret_cast<const char*>(ptr),
+                           reinterpret_cast<const char*>(ptr)+max);
+
     get_io_service().post(boost::bind(&Connection::SendMessageImp, this, msg));
 }
@@ -244 +270 @@
     // to run.
     if (fConnectionTimer.expires_at() < ba::deadline_timer::traits_type::now())
-        AsyncConnect();
+        StartConnect();
 }
 
@@ -272 +298 @@
 
         // fSocket.get_io_service()/*fEventQueue*/.stop();
+
+        ConnectionEstablished();
         return;
     }
@@ -287 +315 @@
     {
         stringstream msg;
-        msg << "Connecting to " << URL() << ": " << error.message() << " (" << error << ")";
+        if (URL()!=":")
+            msg << "Connecting to " << URL() << ": " << error.message() << " (" << error << ")";
 
         if (fErrConnect!=msg.str())
@@ -315 +344 @@
 
 // FIXME: Async connect should get address and port as an argument
-void Connection::AsyncConnect()
+void Connection::StartConnect()
 {
     fConnectionStatus = 1;
@@ -321 +350 @@
     tcp::resolver resolver(get_io_service());
 
+    boost::system::error_code ec;
+
     tcp::resolver::query query(fAddress, fPort);
-    tcp::resolver::iterator iterator = resolver.resolve(query);
+    tcp::resolver::iterator iterator = resolver.resolve(query, ec);
 
     stringstream msg;
-    msg << "Trying to connect to " << fAddress << ":" << fPort << "...";
-
+    if (!fAddress.empty() || !fPort.empty() || ec)
+        msg << "Trying to connect to " << URL() << "...";
+
+    if (ec)
+        msg << " " << ec.message() << " (" << ec << ")";
+
+    // Only output message if it has changed
     if (fMsgConnect!=msg.str())
     {
         fMsgConnect = msg.str();
-        Message(msg);
-    }
-
-    // Start connection attempts (will also reset deadline counter)
-    AsyncConnect(iterator);
-}
-
-void Connection::SetEndpoint(const char *addr, int port)
+        ec ? Error(msg) : Message(msg);
+    }
+
+    if (ec)
+        AsyncWait(fConnectionTimer, 250, &Connection::HandleConnectionTimer);
+    else
+        // Start connection attempts (will also reset deadline counter)
+        AsyncConnect(iterator);
+}
+
+void Connection::SetEndpoint(const string &addr, int port)
 {
     if (fConnectionStatus>=1)
@@ -346 +385 @@
 }
 
+void Connection::SetEndpoint(const string &addr, const string &port)
+{
+    if (fConnectionStatus>=1 && URL()!=":")
+        Warn("Connection or connection attempt in progress. New endpoint only valid for next connection.");
+
+    fAddress = addr;
+    fPort    = port;
+}
+
+void Connection::SetEndpoint(const string &addr)
+{
+    const size_t p0 = addr.find_first_of(':');
+    const size_t p1 = addr.find_last_of(':');
+
+    if (p0==string::npos || p0!=p1)
+    {
+        Error("Connection::SetEndPoint - Wrong format of argument.");
+        return;
+    }
+
+    SetEndpoint(addr.substr(0, p0), addr.substr(p0+1));
+}
+
+
 
 Connection::Connection(ba::io_service& ioservice, ostream &out) :
 MessageImp(out), tcp::socket(ioservice),
-fLog(0), fAddress("localhost"), fPort("5000"),
+fLog(0), //fAddress("localhost"), fPort("5000"),
 fInTimeout(ioservice), fOutTimeout(ioservice), fConnectionTimer(ioservice),
 fConnectionStatus(0)
 {
 }
-
+/*
 Connection::Connection(ba::io_service& ioservice, const string &addr, int port) :
 tcp::socket(ioservice),
@@ -369 +432 @@
 {
 }
-
+*/

trunk/FACT++/src/Connection.h

@@ -5 +5 @@
 #include <string>
 #include <boost/asio.hpp>
+#include <boost/function.hpp>
 #include <boost/asio/deadline_timer.hpp>
 
@@ -40 +41 @@
     int Write(const Time &t, const char *txt, int qos=kInfo);
 
-    void AsyncRead(boost::asio::mutable_buffers_1 buffers/*,
-                   void (Connection::*handler)(const boost::system::error_code&, size_t)*/);
-    void AsyncWrite(boost::asio::mutable_buffers_1 buffers/*,
-                    void (Connection::*handler)(const boost::system::error_code&)*/);
+    void AsyncRead(boost::asio::mutable_buffers_1 buffers, int type=0);
+    void AsyncWrite(boost::asio::mutable_buffers_1 buffers);
     void AsyncWait(boost::asio::deadline_timer &timer, int millisec,
                    void (Connection::*handler)(const boost::system::error_code&));
@@ -50 +49 @@
     std::string URL() const { return fAddress + ":" + fPort; }
 
+    void CloseImp(bool restart=true);
+
+    void ConnectImp(const boost::system::error_code& error,
+                    boost::asio::ip::tcp::resolver::iterator endpoint_iterator);
+
 public:
 
     // ------------------------ close --------------------------
     // close from another thread
-    void CloseImp(bool restart=true);
     void PostClose(bool restart=true);
 
@@ -62 +65 @@
     void HandleSentData(const boost::system::error_code& error, size_t);
     void SendMessageImp(const std::vector<char> &msg);
+    void PostMessage(std::vector<uint16_t> msg);
     void PostMessage(const std::vector<char> &msg);
+    void PostMessage(const void *msg, size_t s=0);
     void PostMessage(const std::string &cmd, size_t s=-1);
+
+    template<std::size_t N>
+        void PostMessage(boost::array<uint16_t, N> arr)
+    {
+        // Convert to network byte order
+        for_each(arr.begin(), arr.end(), htons);
+        PostMessage(std::vector<uint16_t>(arr.begin(), arr.end()));
+    }
 
     // ------------------------ connect --------------------------
 
-    virtual void ConnectImp(const boost::system::error_code& error,
-                            boost::asio::ip::tcp::resolver::iterator endpoint_iterator);
+    virtual void ConnectionEstablished() { }
 
-    void AsyncConnect();
+    void StartConnect();
 
     Connection(boost::asio::io_service& io_service, std::ostream &out);
-    Connection(boost::asio::io_service& io_service, const std::string &addr="localhost", int port=5000);
-    Connection(boost::asio::io_service& io_service, const std::string &addr, const std::string &port);
+//    Connection(boost::asio::io_service& io_service, const std::string &addr="localhost", int port=5000);
+//    Connection(boost::asio::io_service& io_service, const std::string &addr, const std::string &port);
 
-    void SetEndpoint(const char *addr, int port);
+    void SetEndpoint(const std::string &addr, int port);
+    void SetEndpoint(const std::string &addr, const std::string &port);
+    void SetEndpoint(const std::string &addr);
 
     // ------------------------ others --------------------------
 
-    virtual void HandleReceivedData(const boost::system::error_code&, size_t) { }
+    virtual void HandleReceivedData(const boost::system::error_code&, size_t, int = 0) { }
     virtual void HandleReadTimeout(const boost::system::error_code&) { }