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

Timestamp:

04/05/11 11:21:27 (14 years ago)

Author:

tbretz

Message:

- Added enum ConnectionStatus_t
- change fConnectionStatus type to ConnectionStatus_t
- removed obsolete fReadBuffer
- added const qualifiers to const_buffer_1 in AsyncWrite
- added Out() member function
- (temporarily?) removed some PostMessage member functions
- removed host and port from constructors for simplification
- improved how connections are established
- improved output of host name during connection attempts

File:

• trunk/FACT++/src/Connection.cc (modified) (20 diffs)

trunk/FACT++/src/Connection.cc

@@ -31 +31 @@
 }
 
-void Connection::AsyncRead(ba::mutable_buffers_1 buffers, int type)
+void Connection::AsyncRead(const ba::mutable_buffers_1 buffers, int type)
 {
     ba::async_read(*this, buffers,
@@ -38 +38 @@
 }
 
-void Connection::AsyncWrite(ba::mutable_buffers_1 buffers)
+void Connection::AsyncWrite(const ba::const_buffers_1 &buffers)
 {
     ba::async_write(*this, buffers,
@@ -63 +63 @@
 void Connection::AsyncConnect(tcp::resolver::iterator iterator)
 {
-    //cout << "Start async connect(...)" << endl;
-    fConnectionStatus = 1;
-
     tcp::endpoint endpoint = *iterator;
 
@@ -72 +69 @@
                   boost::bind(&Connection::ConnectImp,
                               this, ba::placeholders::error,
-                              ++iterator));
+                              iterator));
 
     // We will get a "Connection timeout anyway"
@@ -85 +82 @@
     {
         stringstream str;
-        str << "Connection closed to " << URL() << ".";// << endl;
+        str << "Connection closed to " << URL() << ".";
         Message(str);
     }
@@ -96 +93 @@
 
     // Reset the connection status
-    fConnectionStatus = 0;
+    fConnectionStatus = kDisconnected;
 
     // Stop deadline counters
@@ -102 +99 @@
     fOutTimeout.cancel();
 
+    // Empty output queue
+    fOutQueue.clear();
+
     if (!restart || IsConnecting())
         return;
@@ -111 +111 @@
     // Start trying to reconnect
     fMsgConnect = "";
+    fErrConnect = "";
     StartConnect();
 }
@@ -122 +123 @@
 void Connection::HandleWriteTimeout(const bs::error_code &error)
 {
-    // 125: Operation canceled
-    if (error && error!=bs::error_code(125, bs::system_category))
+    // 125: Operation canceled (bs::error_code(125, bs::system_category))
+    if (error && error!=ba::error::basic_errors::operation_aborted)
     {
         stringstream str;
         str << "Write timeout of " << URL() << ": " << error.message() << " (" << error << ")";// << endl;
         Error(str);
+
+        CloseImp();
+        return;
     }
 
@@ -149 +153 @@
 }
 
-void Connection::HandleSentData(const bs::error_code& error, size_t)
+void Connection::HandleSentData(const bs::error_code& error, size_t n)
 {
     if (error)
@@ -161 +165 @@
     }
 
-    Message("Data successfully sent to "+URL());
+    stringstream msg;
+    msg << n << " bytes successfully sent to " << URL();
+    Message(msg);
 
     // This is "thread" safe because SendMessage and HandleSentMessage
@@ -177 +183 @@
 
     // AsyncWrite + Deadline
-    AsyncWrite(ba::buffer(fOutQueue.front())/*, &Connection::HandleSentData*/);
+    AsyncWrite(ba::const_buffers_1(&fOutQueue.front()[0],fOutQueue.front().size())/*, &Connection::HandleSentData*/);
     AsyncWait(fOutTimeout, 5000, &Connection::HandleWriteTimeout);
 }
 
-void Connection::SendMessageImp(const vector<char> &msg)
+// It is important that when SendMessageImp is called, or to be more
+// precise boost::bind is called, teh data is copied!
+void Connection::SendMessageImp(const vector<char> msg)
 {
     /*
@@ -202 +210 @@
 
     // AsyncWrite + Deadline
-    AsyncWrite(ba::buffer(fOutQueue.front())/*, &Connection::HandleSentData*/);
+    AsyncWrite(ba::const_buffers_1(&fOutQueue.front()[0],fOutQueue.front().size())/*, &Connection::HandleSentData*/);
     AsyncWait(fOutTimeout, 5000, &Connection::HandleWriteTimeout);
-}
-
-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));
 }
 
@@ -239 +229 @@
     vector <char>msg(max);
 
-    for (unsigned int i=0; i<max; i++)
-        msg[i] = 0;
-
-    for (unsigned int i=0; i<min(cmd.length()+1, max); i++)
-        msg[i] = cmd[i];
-
-    get_io_service().post(boost::bind(&Connection::SendMessageImp, this, msg));
+    copy(cmd.begin(), cmd.begin()+min(cmd.length()+1, max), msg.begin());
+
+    PostMessage(msg);
 }
 
@@ -251 +237 @@
 {
     // 125: Operation canceled
-    if (error && error!=bs::error_code(125, bs::system_category))
+    if (error && error!=ba::error::basic_errors::operation_aborted)
     {
         stringstream str;
@@ -274 +260 @@
 
 void Connection::ConnectImp(const bs::error_code& error,
-                            tcp::resolver::iterator endpoint_iterator)
-{
+                            tcp::resolver::iterator iterator)
+{
+    tcp::endpoint endpoint = *iterator;
+
+    const string host = endpoint.port()==0 ? "" :
+        endpoint.address().to_string()+":"+lexical_cast<string>(endpoint.port());
+
     // Connection established
     if (!error)
     {
-        Message("Connection established to "+URL()+"...");
-
-        // Initialize hardware...
-        // Set state to : connected/undefined until we get the first message
-        //                (send request for configuration) ?
-        //                (send configuration?)
-        fMsgConnect = "";
-        fErrConnect = "";
-
-        fConnectionStatus = 2;
-        //StartAsyncRead();
-
-        // We can create the buffer here and delete it in
-        // Handle Read. However, the read buffer is not filled again
-        // before it was not read within HandleReceivedData -- so a single
-        // buffer should be good enough. Before HandleReceivedData
-        // returns the data must be copied to a "safe" place.
-
-        // fSocket.get_io_service()/*fEventQueue*/.stop();
+        Message("Connection established to "+host+"...");
+
+        fConnectionStatus = kConnected;
 
         ConnectionEstablished();
@@ -309 +284 @@
     // the endpoint again.
     CloseImp(false);
-    //fSocket.close();
-
-    // 111: Connection refused
-    if (1/*error!=bs::error_code(111, bs::system_category)*/)
-    {
-        stringstream msg;
-        if (URL()!=":")
-            msg << "Connecting to " << URL() << ": " << error.message() << " (" << error << ")";
-
-        if (fErrConnect!=msg.str())
-        {
+
+    stringstream msg;
+    if (!host.empty())
+        msg << "Connecting to " << host << ": " << error.message() << " (" << error << ")";
+
+    if (fErrConnect!=msg.str())
+    {
+        if (error!=ba::error::basic_errors::connection_refused)
             fMsgConnect = "";
-            fErrConnect = msg.str();
-            Warn(fErrConnect);
-        }
+        fErrConnect = msg.str();
+        Warn(fErrConnect);
     }
 
     // Go on with the next
-    if (endpoint_iterator != tcp::resolver::iterator())
-    {
-        AsyncConnect(endpoint_iterator);
+    if (++iterator != tcp::resolver::iterator())
+    {
+        AsyncConnect(iterator);
         return;
     }
@@ -346 +317 @@
 void Connection::StartConnect()
 {
-    fConnectionStatus = 1;
+    fConnectionStatus = kConnecting;
 
     tcp::resolver resolver(get_io_service());
@@ -401 +372 @@
     if (p0==string::npos || p0!=p1)
     {
-        Error("Connection::SetEndPoint - Wrong format of argument.");
+        Error("Connection::SetEndPoint - Wrong format of argument ('host:port' expected)");
         return;
     }
@@ -411 +382 @@
 
 Connection::Connection(ba::io_service& ioservice, ostream &out) :
-MessageImp(out), tcp::socket(ioservice),
-fLog(0), //fAddress("localhost"), fPort("5000"),
+MessageImp(out), tcp::socket(ioservice), fLog(0),
 fInTimeout(ioservice), fOutTimeout(ioservice), fConnectionTimer(ioservice),
-fConnectionStatus(0)
-{
-}
-/*
-Connection::Connection(ba::io_service& ioservice, const string &addr, int port) :
-tcp::socket(ioservice),
-fLog(0), fAddress(addr), fPort(lexical_cast<string>(port)),
-fInTimeout(ioservice), fOutTimeout(ioservice), fConnectionTimer(ioservice),
-fConnectionStatus(0)
-{
-}
-
-Connection::Connection(ba::io_service& ioservice, const string &addr, const string &port) :
-tcp::socket(ioservice), fLog(0), fAddress(addr), fPort(port),
-fInTimeout(ioservice), fOutTimeout(ioservice), fConnectionTimer(ioservice),
-fConnectionStatus(0)
-{
-}
-*/
+fConnectionStatus(kDisconnected)
+{
+}