Changeset 10268 for trunk/FACT++/src/Connection.cc

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.

File:

• trunk/FACT++/src/Connection.cc (modified) (11 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 @@
 {
 }
-
+*/