Overlapping transport sends no longer corrupt RawSocket messages (fixes #117)

Author: ecorm

cppwamp/include/cppwamp/internal/asiotransport.hpp

@@ -176,35 +176,43 @@ class AsioTransport :
                "Outgoing message is longer than allowed by peer");
 
         message->prepare(type);
-        if (txQueue_.empty())
-            transmit(std::move(message));
-        else
-            txQueue_.push(std::move(message));
+        txQueue_.push(std::move(message));
+        transmit();
     }
 
-    void transmit(Buffer message)
+private:
+    void transmit()
     {
-        if (socket_)
+        if (isReadyToTransmit())
         {
+            txBuffer_ = txQueue_.front();
+            txQueue_.pop();
+
             auto self = this->shared_from_this();
-            boost::asio::async_write(*socket_, message->gatherBuffers(),
-                [this, self, message](AsioErrorCode ec, size_t)
+            boost::asio::async_write(*socket_, txBuffer_->gatherBuffers(),
+                [this, self](AsioErrorCode ec, size_t size)
                 {
+                    txBuffer_.reset();
                     if (ec)
                     {
                         txQueue_ = TransmitQueue();
                         socket_.reset();
                     }
-                    else if (!txQueue_.empty())
+                    else
                     {
-                        auto msg = txQueue_.front();
-                        txQueue_.pop();
-                        transmit(std::move(msg));
+                        transmit();
                     }
                 });
         }
     }
 
+    bool isReadyToTransmit() const
+    {
+        return socket_ &&           // Socket is still open
+               !txBuffer_ &&        // No async_write is in progress
+               !txQueue_.empty();   // One or more messages are enqueued
+    }
+
     void receive()
     {
         if (socket_)
@@ -336,6 +344,7 @@ class AsioTransport :
     Buffer rxBuffer_;
     Buffer pingBuffer_;
     TransmitQueue txQueue_;
+    Buffer txBuffer_;
     TimePoint pingStart_;
     TimePoint pingStop_;
 };

test/.crossbar/config.json

@@ -38,6 +38,7 @@
                         "type": "tcp",
                         "port": 12345
                     },
+                    "max_message_size": 16777216,
                     "serializers": [
                         "json"
                     ],
@@ -49,6 +50,7 @@
                         "type": "unix",
                         "path": "./udstest"
                     },
+                    "max_message_size": 16777216,
                     "serializers": [
                         "msgpack"
                     ],

test/wamptest.cpp

@@ -2644,4 +2644,87 @@ GIVEN( "an IO service and a TCP connector" )
     }
 }}
 
+
+//------------------------------------------------------------------------------
+SCENARIO( "Outbound Messages are Properly Enqueued", "[WAMP]" )
+{
+GIVEN( "these test fixture objects" )
+{
+    using Yield = boost::asio::yield_context;
+
+    AsioService iosvc;
+    auto cnct = tcp(iosvc);
+    auto session1 = CoroSession<>::create(iosvc, cnct);
+    auto session2 = CoroSession<>::create(iosvc, cnct);
+
+    WHEN( "an RPC is invoked while a large event payload is being transmitted" )
+    {
+        auto callee = session1;
+        auto subscriber = session2;
+        std::string eventString;
+
+        // Fill large string with repeating character sequence
+        std::string largeString(1024*1024, ' ');
+        for (size_t i=0; i<largeString.length(); ++i)
+            largeString[i] = '0' + (i % 64);
+
+        auto onEvent = [&iosvc, &eventString](Event, std::string str)
+        {
+            eventString = std::move(str);
+        };
+
+        // Simple RPC that returns the string argument back to the caller.
+        auto echo =
+            [callee](Invocation, std::string str) -> Outcome
+            {
+                return Result({str});
+            };
+
+        // RPC that triggers the publishing of a large event payload
+        auto trigger =
+            [callee, &largeString] (Invocation) -> Outcome
+            {
+                callee->publish(Pub("grapevine").withArgs(largeString));
+                return Result();
+            };
+
+        boost::asio::spawn(iosvc, [&](boost::asio::yield_context yield)
+        {
+            callee->connect(yield);
+            callee->join(Realm(testRealm), yield);
+            callee->enroll(Procedure("echo"), unpackedRpc<std::string>(echo),
+                           yield);
+            callee->enroll(Procedure("trigger"), trigger, yield);
+
+            subscriber->connect(yield);
+            subscriber->join(Realm(testRealm), yield);
+            subscriber->subscribe(Topic("grapevine"),
+                                  unpackedEvent<std::string>(onEvent),
+                                  yield);
+
+            for (int i=0; i<10; ++i)
+            {
+                // Use async call so that it doesn't block until completion.
+                subscriber->call(Rpc("trigger").withArgs("hello"),
+                                 [](AsyncResult<Result>) {});
+
+                /*  Try to get callee to send an RPC response while it's still
+                    transmitting the large event payload. AsioTransport should
+                    properly enqueue the RPC response while the large event
+                    payload is being transmitted. */
+                while (eventString.empty())
+                    subscriber->call(Rpc("echo").withArgs("hello"), yield);
+
+                CHECK_THAT( eventString, Equals(largeString) );
+                eventString.clear();
+            }
+            callee->disconnect();
+            subscriber->disconnect();
+        });
+
+        iosvc.run();
+    }
+}
+}
+
 #endif // #if CPPWAMP_TESTING_WAMP