Changeset 17932 for trunk

Timestamp:

07/23/14 15:39:06 (10 years ago)

Author:

tbretz

Message:

Revived the valid state; make sure the first message (which might still be junk from a non empty buffer) is properly treated; make sure that an automatic re-connect properly works

File:

• trunk/FACT++/src/sqmctrl.cc (modified) (11 diffs)

trunk/FACT++/src/sqmctrl.cc

@@ -30 +30 @@
 private:
     bool     fIsVerbose;
+    bool     fFirstMessage;
+    bool     fValid;
     uint16_t fTimeout;
 
@@ -54 +56 @@
                 Error(str);
             }
-            PostClose(err!=ba::error::basic_errors::operation_aborted);
+            PostClose(false);//err!=ba::error::basic_errors::operation_aborted);
             return;
         }
@@ -71 +73 @@
 
         if (fIsVerbose)
-            Out() << Time().GetAsStr("%H:%M:%S.%f") << "[" << buffer.size() << "]: " << buffer << endl;
+        {
+            Out() << Time().GetAsStr("%H:%M:%S.%f") << "[" << buffer.size() << "]: " << buffer << "|" << endl;
+            // Out() << Time().GetAsStr("%H:%M:%S.%f") << "[ " << vec.size() << "]: ";
+            // for (auto it=vec.begin(); it!=vec.end(); it++)
+            //     Out() << *it << "|";
+            // Out() << endl;
+        }
 
         vector<string> vec;
         boost::split(vec, buffer, boost::is_any_of(","));
 
-        // Send next request in fTimeout milliseconds calculated from
-        // the last request onwards.
-        fTrigger.expires_at(fTrigger.expires_at()+boost::posix_time::milliseconds(fTimeout));
-        fTrigger.async_wait(boost::bind(&ConnectionSQM::HandleRequestTrigger,
-                                          this, dummy::error));
         try
         {
@@ -87 +90 @@
 
             if (vec[0]!="r")
-                throw runtime_error("Not a 'reading request' answer: "+vec[0]);
+                throw runtime_error("Not a proper answer["+to_string(vec[0].size())+"]: "+vec[0]);
 
             SQM::Data data;
@@ -98 +101 @@
 
             Update(data);
+
+            fValid = true;
         }
         catch (const exception &e)
         {
-            Error("Parsing received data failed ["+string(e.what())+"]");
-            Error("Received: "+buffer);
-            PostClose(true);
-        }
+            if (fFirstMessage)
+                Warn("Parsing first message failed ["+string(e.what())+"]");
+            else
+            {
+                Error("Parsing received message failed ["+string(e.what())+"]");
+                Error("Received: "+buffer);
+                PostClose(false);
+                return;
+            }
+        }
+
+        // Send next request in fTimeout milliseconds calculated from
+        // the last request onwards.
+        fTrigger.expires_at(fTrigger.expires_at()+boost::posix_time::milliseconds(fTimeout));
+        fTrigger.async_wait(boost::bind(&ConnectionSQM::HandleRequestTrigger,
+                                        this, dummy::error));
+
+        fFirstMessage = false;
     }
 
@@ -116 +135 @@
             Error(str);
 
-            PostClose(true);
+            PostClose(false);
             return;
         }
@@ -124 +143 @@
             // For example: Here we could schedule a new accept if we
             // would not want to allow two connections at the same time.
-            //PostClose(true);
-            return;
-        }
+            PostClose(true);
+            return;
+        }
+
+        // This is called if the deadline has been shifted
+        if (error==ba::error::basic_errors::operation_aborted)
+            return;
 
         // Check whether the deadline has passed. We compare the deadline
@@ -139 +162 @@
         Error(str);
 
-        PostClose(true);
+        PostClose(false);
+
+        fInTimeout.expires_from_now(boost::posix_time::milliseconds(1000));
+        fInTimeout.async_wait(boost::bind(&ConnectionSQM::HandleReadTimeout,
+                                          this, dummy::error));
     }
 
     void HandleRequestTrigger(const bs::error_code &error)
     {
+
         // 125: Operation canceled (bs::error_code(125, bs::system_category))
         if (error && error!=ba::error::basic_errors::operation_aborted)
@@ -151 +179 @@
             Error(str);
 
-            PostClose(true);
+            PostClose(false);
             return;
         }
@@ -177 +205 @@
         PostMessage(string("rx\n"), 3);
 
+        // Do not schedule two reads
+        if (!fFirstMessage)
+        {
+            async_read_until(*this, fBuffer, '\n',
+                             boost::bind(&ConnectionSQM::HandleRead, this,
+                                         dummy::error, dummy::bytes_transferred));
+        }
+
+        fInTimeout.expires_from_now(boost::posix_time::milliseconds(fTimeout*1.5));
+        fInTimeout.async_wait(boost::bind(&ConnectionSQM::HandleReadTimeout,
+                                          this, dummy::error));
+    }
+
+private:
+    // This is called when a connection was established
+    void ConnectionEstablished()
+    {
+        fValid = false;
+        fFirstMessage = true;
+
+        // Empty a possible buffer first before we start reading
+        // otherwise reading and writing might not be consecutive
         async_read_until(*this, fBuffer, '\n',
                          boost::bind(&ConnectionSQM::HandleRead, this,
                                      dummy::error, dummy::bytes_transferred));
 
-        fInTimeout.expires_from_now(boost::posix_time::milliseconds(fTimeout*1.5));
-        fInTimeout.async_wait(boost::bind(&ConnectionSQM::HandleReadTimeout,
-                                          this, dummy::error));
-    }
-
-private:
-    // This is called when a connection was established
-    void ConnectionEstablished()
-    {
-        StartReadReport();
-        fTrigger.expires_at(Time());
+        // If there was no immediate answer, send a request
+        fTrigger.expires_at(Time()+boost::posix_time::milliseconds(1000));
+        fTrigger.async_wait(boost::bind(&ConnectionSQM::HandleRequestTrigger,
+                                        this, dummy::error));
     }
 
@@ -217 +260 @@
     int GetState() const
     {
-        return is_open() ? SQM::State::kConnected : SQM::State::kDisconnected;
+        if (!is_open())
+            return  SQM::State::kDisconnected;
+
+        return fValid ? SQM::State::kValid : SQM::State::kConnected;
     }
 };