Interlieved data fix

DXMPP/Network/AsyncTCPXMLClient.cpp

@@ -169,8 +169,8 @@ else std::cout
             int  NrSubRootTagNamesFound;
 
             ExpatStructure( AsyncTCPXMLClient *Sender, XML_Parser Parser)
-                : 
-                  Sender(Sender), 
+                :
+                  Sender(Sender),
                   Parser(Parser)
             {
                 RootTagName = "";
@@ -198,22 +198,22 @@ else std::cout
             CurrentConnectionState = ConnectionState::Error;
             ErrorCallback();
         }
-        
+
         void SAXEndElementHandler(void* data, const XML_Char* el)
         {
             ExpatStructure *ES = static_cast<ExpatStructure*>(data);
             if( ES->EndPosition > 0 )
                 return;
 
             string StrEl(el);
-            
+
             if(StrEl == string("</stream:stream>"))
             {
                 std::cerr << "Received end of stream" << std::endl;
                 ES->Sender->SignalError();
                 return;
             }
-            
+
             if(StrEl != ES->RootTagName )
                 return;
 
@@ -306,16 +306,23 @@ else std::cout
         }
 
 
-
-        void AsyncTCPXMLClient::WriteTextToSocket(const string &Data)
+        void AsyncTCPXMLClient::FlushOutgoingDataUnsafe()
         {
-            if( CurrentConnectionState != ConnectionState::Connected )
+            if(OutgoingData.empty())
+            {
+                Flushing = false;
                 return;
+            }
 
-            boost::system::error_code ec;
+            Flushing = true;
 
-            std::shared_ptr<std::string> SharedData (new std::string(Data));
+            std::shared_ptr<std::string> SharedData = OutgoingData.front();
+            OutgoingData.pop();
 
+            if(OutgoingData.empty())
+                Flushing = false;
+
+            boost::system::error_code ec;
             if(SSLConnection)
             {
                 SynchronizationStrand.post([&,SharedData, this]
@@ -342,13 +349,9 @@ else std::cout
 
                     ErrorCallback();
                 }*/
-
-
             }
             else
             {
-
-                std::shared_ptr<std::string> SharedData = std::make_shared<std::string>(Data);
                 boost::asio::async_write(*tcp_socket, boost::asio::buffer(*SharedData),
                                          SynchronizationStrand.wrap(
                                             boost::bind(&AsyncTCPXMLClient::HandleWrite,
@@ -375,7 +378,36 @@ else std::cout
             LastWrite = boost::posix_time::microsec_clock::universal_time();
 
             DebugOut(DebugOutputTreshold::Debug) << "Write text to socket:" <<
-                std::endl << Data << std::endl;
+                std::endl << "+++" << std::endl << *SharedData << std::endl << " --- " << std::endl;
+        }
+
+        void AsyncTCPXMLClient::FlushOutgoingData()
+        {
+            boost::unique_lock<boost::shared_mutex> WriteLock(OutgoingDataMutex);
+
+            if(Flushing)
+                return;
+
+            FlushOutgoingDataUnsafe();
+        }
+
+
+        void AsyncTCPXMLClient::WriteTextToSocket(const string &Data)
+        {
+            if( CurrentConnectionState != ConnectionState::Connected )
+                return;
+
+            {
+                boost::unique_lock<boost::shared_mutex> WriteLock(OutgoingDataMutex);
+                std::shared_ptr<std::string> SharedData (new std::string(Data));
+                OutgoingData.push( SharedData );
+
+                if(!Flushing)
+                    FlushOutgoingDataUnsafe();  /* we have lock */
+            }
+
+
+            return;
         }
 
         bool AsyncTCPXMLClient::EnsureTCPKeepAlive()
@@ -531,6 +563,8 @@ else std::cout
                 return;
             }
 
+            FlushOutgoingData();
+
             //DebugOut(DebugOutputTreshold::Debug) << "Got ack for " << *Data << std::endl;
         }
 

DXMPP/Network/AsyncTCPXMLClient.hpp

@@ -68,11 +68,17 @@ namespace DXMPP
 
             boost::shared_mutex IncomingDocumentsMutex;
             std::queue<pugi::xml_document*> IncomingDocuments;
-
-
+
+            std::queue<std::shared_ptr<std::string>> OutgoingData;
+            boost::shared_mutex OutgoingDataMutex;
+            bool Flushing;
+
+            void FlushOutgoingDataUnsafe();
 
         public:
 
+            void FlushOutgoingData();
+
             void SignalError();
             std::stringstream *ReadDataStream;
             char * ReadDataBuffer;
@@ -164,6 +170,7 @@ namespace DXMPP
                 this->io_service = IOService;
                 this->Hostname = Hostname;
                 this->Portnumber = Portnumber;
+                Flushing = false;
             }
         };
     }

Examples/EchoBot.cpp

@@ -92,7 +92,7 @@ int main(int, const char **)
     while(!Handler.Quit)
     {
         boost::this_thread::sleep(boost::posix_time::milliseconds(10));
-    }   
+    }
 
     return 0;
 }