More improvements

stgates · stgates · commit 3e7ac07ab780 · 2014-09-18T20:42:57.000-07:00
diff --git a/Release/include/cpprest/ws_msg.h b/Release/include/cpprest/ws_msg.h
@@ -59,6 +59,7 @@ namespace client
 namespace details
 {
     class winrt_client;
+    class ReceiveContext;
     class ws_desktop_client;
 }
 
@@ -83,10 +84,6 @@ class _websocket_message
 {
 public:
 
-    concurrency::streams::streambuf<uint8_t> & body() { return m_buf; }
-
-    void set_body(const concurrency::streams::streambuf<uint8_t> &buf) { m_buf = buf; }
-
     void set_msg_type(websocket_message_type msg_type) { m_msg_type = msg_type; }
 
     void set_length(size_t len) { m_length = len; }
@@ -97,10 +94,7 @@ class _websocket_message
 
 private:
 
-    concurrency::streams::streambuf<uint8_t> m_buf;
-
     websocket_message_type m_msg_type;
-
     size_t m_length;
 };
 }
@@ -176,13 +170,13 @@ class websocket_outgoing_message
     }
 
 private:
-
     friend class details::winrt_client;
     friend class details::ws_desktop_client;
 
     std::shared_ptr<details::_websocket_message> _m_impl;
 
     pplx::task_completion_event<void> m_body_sent;
+    concurrency::streams::streambuf<uint8_t> m_body;
 
     void signal_body_sent()
     {
@@ -200,21 +194,21 @@ class websocket_outgoing_message
     {
         _m_impl->set_msg_type(msg_type);
         _m_impl->set_length(data.length());
-        _m_impl->set_body(concurrency::streams::container_buffer<std::string>(std::move(data)));
+        m_body = concurrency::streams::container_buffer<std::string>(std::move(data));
     }
 
     void _set_message(const std::string &data, websocket_message_type msg_type)
     {
         _m_impl->set_msg_type(msg_type);
         _m_impl->set_length(data.length());
-        _m_impl->set_body(concurrency::streams::container_buffer<std::string>(data));
+        m_body = concurrency::streams::container_buffer<std::string>(data);
     }
 
     void _set_message(const concurrency::streams::istream &istream, size_t len, websocket_message_type msg_type)
     {
         _m_impl->set_msg_type(msg_type);
-        _m_impl->set_body(istream.streambuf());
         _m_impl->set_length(len);
+        m_body = istream.streambuf();
     }
 };
 
@@ -224,11 +218,8 @@ class websocket_outgoing_message
 class websocket_incoming_message
 {
 public:
-    websocket_incoming_message() : _m_impl(std::make_shared<details::_websocket_message>()) 
-    { 
-        // Body defaults to producer_consumer_buffer.
-        // Perhaps in the future options could be exposed to allow the user to set.
-        _m_impl->set_body(concurrency::streams::producer_consumer_buffer<uint8_t>());
+    websocket_incoming_message() : _m_impl(std::make_shared<details::_websocket_message>())
+    {
     }
 
     /// <summary>
@@ -237,7 +228,7 @@ class websocket_incoming_message
     /// </summary>
     /// <returns>String containing body of the message.</returns>
     _ASYNCRTIMP pplx::task<std::string> extract_string() const;
-
+    
     /// <summary>
     /// Produces a stream which the caller may use to retrieve body from an incoming message.
     /// Can be used for both UTF-8 (text) and binary message types.
@@ -248,7 +239,11 @@ class websocket_incoming_message
     /// </remarks>
     concurrency::streams::istream body() const
     {
-        return _m_impl->body().create_istream();
+        auto to_uint8_t_stream = [](const concurrency::streams::streambuf<uint8_t> &buf) -> concurrency::streams::istream
+        {
+            return buf.create_istream();
+        };
+        return to_uint8_t_stream(m_body);
     }
 
     /// <summary>
@@ -270,6 +265,12 @@ class websocket_incoming_message
 private:
     friend class details::winrt_client;
     friend class details::ws_desktop_client;
+    friend class details::ReceiveContext;
+    
+    // Store message body in a container buffer backed by a string.
+    // Allows for optimization in the string message cases.
+    concurrency::streams::container_buffer<std::string> m_body;
+    
     std::shared_ptr<details::_websocket_message> _m_impl;
 };
 
diff --git a/Release/src/websockets/client/ws_client.cpp b/Release/src/websockets/client/ws_client.cpp
@@ -220,7 +220,7 @@ class ws_desktop_client : public _websocket_client_impl, public std::enable_shar
             // 'move' the payload into a container buffer to avoid any copies.
             auto &payload = msg->get_raw_payload();
             const auto len = payload.size();
-            incmsg->set_body(concurrency::streams::container_buffer<std::string>(std::move(payload)));
+            ws_incoming_message.m_body = concurrency::streams::container_buffer<std::string>(std::move(payload));
             incmsg->set_length(len);
 
             std::unique_lock<std::mutex> lock(m_receive_queue_lock);
@@ -374,7 +374,7 @@ class ws_desktop_client : public _websocket_client_impl, public std::enable_shar
     void send_msg(websocket_outgoing_message msg)
     {
         auto this_client = this->shared_from_this();
-        auto& is_buf = msg._m_impl->body();
+        auto& is_buf = msg.m_body;
         auto length = msg._m_impl->length();
 
         if (length == SIZE_MAX)
@@ -395,10 +395,9 @@ class ws_desktop_client : public _websocket_client_impl, public std::enable_shar
             else
             {
                 // The stream needs to be buffered.
-                concurrency::streams::container_buffer<std::vector<uint8_t>> stbuf;
                 auto is_buf_istream = is_buf.create_istream();
-                msg._m_impl->set_body(stbuf);
-                is_buf_istream.read_to_end(stbuf).then([this_client, msg](pplx::task<size_t> t) mutable
+                msg.m_body = concurrency::streams::container_buffer<std::vector<uint8_t>>();
+                is_buf_istream.read_to_end(msg.m_body).then([this_client, msg](pplx::task<size_t> t) mutable
                 {
                     try
                     {
@@ -476,7 +475,7 @@ class ws_desktop_client : public _websocket_client_impl, public std::enable_shar
             }
 
             return ec;
-        }).then([this_client, msg, acquired, sp_allocated, length](pplx::task<websocketpp::lib::error_code> previousTask) mutable
+        }).then([this_client, msg, is_buf, acquired, sp_allocated, length](pplx::task<websocketpp::lib::error_code> previousTask) mutable
         {
             std::exception_ptr eptr;
             try
@@ -495,7 +494,7 @@ class ws_desktop_client : public _websocket_client_impl, public std::enable_shar
 
             if (acquired)
             {
-                msg._m_impl->body().release(sp_allocated.get(), length);
+                is_buf.release(sp_allocated.get(), length);
             }
 
             // Set the send_task_completion_event after calling release.
diff --git a/Release/src/websockets/client/ws_msg.cpp b/Release/src/websockets/client/ws_msg.cpp
@@ -78,18 +78,7 @@ pplx::task<std::string> websocket_incoming_message::extract_string() const
     {
         return pplx::task_from_exception<std::string>(websocket_exception("Invalid message type"));
     }
-
-    auto& buf_r = _m_impl->body();
-    if (buf_r.in_avail() == 0)
-    {
-        return pplx::task_from_result<std::string>();
-    }
-
-    // TODO all store as strings than can make optimization...
-    std::string body;
-    body.resize(static_cast<std::string::size_type>(buf_r.in_avail()));
-    buf_r.getn(reinterpret_cast<uint8_t*>(&body[0]), body.size()).get();
-    return pplx::task_from_result(std::move(body));
+    return pplx::task_from_result(std::move(m_body.collection()));
 }
 
 }}}}
diff --git a/Release/src/websockets/client/ws_winrt.cpp b/Release/src/websockets/client/ws_winrt.cpp
@@ -61,10 +61,10 @@ ref class ReceiveContext sealed
 
 private:
     // Public members cannot have native types
-    ReceiveContext(std::function<void(std::shared_ptr<websocket_incoming_message>)> receive_handler, std::function<void()> close_handler): m_receive_handler(receive_handler), m_close_handler(close_handler) {}
+    ReceiveContext(std::function<void(websocket_incoming_message &>)> receive_handler, std::function<void()> close_handler): m_receive_handler(receive_handler), m_close_handler(close_handler) {}
 
     // Handler to be executed when a message has been received by the client
-    std::function<void(std::shared_ptr<websocket_incoming_message>)> m_receive_handler;
+    std::function<void(websocket_incoming_message &)> m_receive_handler;
 
     // Handler to be executed when a close message has been received by the client
     std::function<void()> m_close_handler;
@@ -108,15 +108,14 @@ class winrt_client : public _websocket_client_impl, public std::enable_shared_fr
                 Platform::StringReference(client_config.credentials().password().c_str()));
         }
 
-        m_context = ref new ReceiveContext([=](std::shared_ptr<websocket_incoming_message> msg)
+        m_context = ref new ReceiveContext([=](websocket_incoming_message &msg)
         {
-            _ASSERTE(msg != nullptr);
             pplx::task_completion_event<websocket_incoming_message> tce; // This will be set if there are any tasks waiting to receive a message
             {
                 std::lock_guard<std::mutex> lock(m_receive_queue_lock);
                 if (m_receive_task_queue.empty()) // Push message to the queue as no one is waiting to receive
                 {
-                    m_receive_msg_queue.push(std::move(*msg));
+                    m_receive_msg_queue.push(msg);
                     return;
                 }
                 else // There are tasks waiting to receive a message.
@@ -126,7 +125,7 @@ class winrt_client : public _websocket_client_impl, public std::enable_shared_fr
                 }
             }
             // Setting the tce outside the receive lock for better performance
-            tce.set(*msg);
+            tce.set(msg);
         }, 
         [=]() // Close handler called upon receiving a close frame from the server.
         {
@@ -253,7 +252,7 @@ class winrt_client : public _websocket_client_impl, public std::enable_shared_fr
     void send_msg(websocket_outgoing_message &msg)
     {
         auto this_client = this->shared_from_this();
-        auto& is_buf = msg._m_impl->body().create_istream();
+        auto& is_buf = msg.m_body.create_istream();
         auto length = msg._m_impl->length();
 
         if (length == SIZE_MAX)
@@ -274,10 +273,9 @@ class winrt_client : public _websocket_client_impl, public std::enable_shared_fr
             else
             {
                 // The stream needs to be buffered.
-                concurrency::streams::container_buffer<std::vector<uint8_t>> stbuf;
                 auto is_buf_istream = is_buf.create_istream();
-                msg._m_impl->set_body(stbuf);
-                is_buf_istream.read_to_end(stbuf).then([this_client, msg](pplx::task<size_t> t)
+                msg.m_body = concurrency::streams::container_buffer<std::vector<uint8_t>>();
+                is_buf_istream.read_to_end(msg.m_body).then([this_client, msg](pplx::task<size_t> t)
                 {
                     try
                     {
@@ -336,7 +334,7 @@ class winrt_client : public _websocket_client_impl, public std::enable_shared_fr
 
             // Send the data as one complete message, in WinRT we do not have an option to send fragments.
             return pplx::task<unsigned int>(this_client->m_messageWriter->StoreAsync());
-        }).then([this_client, msg, acquired, sp_allocated, length] (pplx::task<unsigned int> previousTask)
+        }).then([this_client, msg, is_buf, acquired, sp_allocated, length](pplx::task<unsigned int> previousTask)
         {
             std::exception_ptr eptr;
             unsigned int bytes_written = 0;
@@ -360,7 +358,7 @@ class winrt_client : public _websocket_client_impl, public std::enable_shared_fr
 
             if (acquired)
             {
-                msg._m_impl->body().release(sp_allocated.get(), bytes_written);
+                is_buf.release(sp_allocated.get(), bytes_written);
             }
 
             // Set the send_task_completion_event after calling release.
@@ -420,11 +418,6 @@ class winrt_client : public _websocket_client_impl, public std::enable_shared_fr
         return pplx::create_task(m_close_tce);
     }
 
-    static std::shared_ptr<details::_websocket_message> &get_impl(websocket_incoming_message msg)
-    {
-        return msg._m_impl;
-    }
-
 private:
 
     // WinRT MessageWebSocket object
@@ -469,7 +462,7 @@ class winrt_client : public _websocket_client_impl, public std::enable_shared_fr
 void ReceiveContext::OnReceive(MessageWebSocket^ sender, MessageWebSocketMessageReceivedEventArgs^ args)
 {
     websocket_incoming_message ws_incoming_message;
-    auto &msg = winrt_client::get_impl(ws_incoming_message);
+    auto &msg = ws_incoming_message->_m_impl;
 
     switch(args->MessageType)
     {
@@ -481,18 +474,16 @@ void ReceiveContext::OnReceive(MessageWebSocket^ sender, MessageWebSocketMessage
         break;
     }
 
-    auto &writebuf = msg->body();
-
     try
     {
         DataReader^ reader = args->GetDataReader();
         const auto len = reader->UnconsumedBufferLength;
-        auto block = writebuf.alloc(len);
-        reader->ReadBytes(Platform::ArrayReference<uint8_t>(block, len)); 
-        writebuf.commit(len);
-        writebuf.close(std::ios::out).wait(); // Since this is an in-memory stream, this call is not blocking.
+        std::string payload;
+        payload.resize(len);
+        reader->ReadBytes(Platform::ArrayReference<uint8_t>(payload.c_str(), len));
+        ws_incoming_message.m_body = concurrency::streams::container_buffer<std::string>(std::move(payload));
         msg->signal_msg_received(len);
-        m_receive_handler(std::make_shared<websocket_incoming_message>(ws_incoming_message));
+        m_receive_handler(ws_incoming_message);
     }
     catch(...)
     {
diff --git a/Release/tests/Functional/websockets/client/receive_msg_tests.cpp b/Release/tests/Functional/websockets/client/receive_msg_tests.cpp
@@ -108,6 +108,9 @@ TEST_FIXTURE(uri_address, receive_text_msg)
 
     receive_text_msg_helper(client, server, m_uri, "hello").wait();
     client.close().wait();
+
+    // TODO stgates
+    VERIFY_ARE_EQUAL(false, true);
 }
 
 // Receive text message (no fragmentation)

Original file line number	Diff line number	Diff line change
`@@ -220,7 +220,7 @@ class ws_desktop_client : public _websocket_client_impl, public std::enable_shar`
`220`	`220`	`// 'move' the payload into a container buffer to avoid any copies.`
`221`	`221`	`auto &payload = msg->get_raw_payload();`
`222`	`222`	`const auto len = payload.size();`
`223`		`- incmsg->set_body(concurrency::streams::container_buffer<std::string>(std::move(payload)));`
	`223`	`+ ws_incoming_message.m_body = concurrency::streams::container_buffer<std::string>(std::move(payload));`
`224`	`224`	`incmsg->set_length(len);`
`225`	`225`
`226`	`226`	`std::unique_lock<std::mutex> lock(m_receive_queue_lock);`
`@@ -374,7 +374,7 @@ class ws_desktop_client : public _websocket_client_impl, public std::enable_shar`
`374`	`374`	`void send_msg(websocket_outgoing_message msg)`
`375`	`375`	`{`
`376`	`376`	`auto this_client = this->shared_from_this();`
`377`		`- auto& is_buf = msg._m_impl->body();`
	`377`	`+ auto& is_buf = msg.m_body;`
`378`	`378`	`auto length = msg._m_impl->length();`
`379`	`379`
`380`	`380`	`if (length == SIZE_MAX)`
`@@ -395,10 +395,9 @@ class ws_desktop_client : public _websocket_client_impl, public std::enable_shar`
`395`	`395`	`else`
`396`	`396`	`{`
`397`	`397`	`// The stream needs to be buffered.`
`398`		`- concurrency::streams::container_buffer<std::vector<uint8_t>> stbuf;`
`399`	`398`	`auto is_buf_istream = is_buf.create_istream();`
`400`		`- msg._m_impl->set_body(stbuf);`
`401`		`- is_buf_istream.read_to_end(stbuf).then([this_client, msg](pplx::task<size_t> t) mutable`
	`399`	`+ msg.m_body = concurrency::streams::container_buffer<std::vector<uint8_t>>();`
	`400`	`+ is_buf_istream.read_to_end(msg.m_body).then([this_client, msg](pplx::task<size_t> t) mutable`
`402`	`401`	`{`
`403`	`402`	`try`
`404`	`403`	`{`
`@@ -476,7 +475,7 @@ class ws_desktop_client : public _websocket_client_impl, public std::enable_shar`
`476`	`475`	`}`
`477`	`476`
`478`	`477`	`return ec;`
`479`		`- }).then([this_client, msg, acquired, sp_allocated, length](pplx::task<websocketpp::lib::error_code> previousTask) mutable`
	`478`	`+ }).then([this_client, msg, is_buf, acquired, sp_allocated, length](pplx::task<websocketpp::lib::error_code> previousTask) mutable`
`480`	`479`	`{`
`481`	`480`	`std::exception_ptr eptr;`
`482`	`481`	`try`
`@@ -495,7 +494,7 @@ class ws_desktop_client : public _websocket_client_impl, public std::enable_shar`
`495`	`494`
`496`	`495`	`if (acquired)`
`497`	`496`	`{`
`498`		`- msg._m_impl->body().release(sp_allocated.get(), length);`
	`497`	`+ is_buf.release(sp_allocated.get(), length);`
`499`	`498`	`}`
`500`	`499`
`501`	`500`	`// Set the send_task_completion_event after calling release.`
Original file line number	Diff line number	Diff line change
`@@ -78,18 +78,7 @@ pplx::task<std::string> websocket_incoming_message::extract_string() const`
`78`	`78`	`{`
`79`	`79`	`return pplx::task_from_exception<std::string>(websocket_exception("Invalid message type"));`
`80`	`80`	`}`
`81`		`-`
`82`		`- auto& buf_r = _m_impl->body();`
`83`		`- if (buf_r.in_avail() == 0)`
`84`		`- {`
`85`		`- return pplx::task_from_result<std::string>();`
`86`		`- }`
`87`		`-`
`88`		`- // TODO all store as strings than can make optimization...`
`89`		`- std::string body;`
`90`		`- body.resize(static_cast<std::string::size_type>(buf_r.in_avail()));`
`91`		`- buf_r.getn(reinterpret_cast<uint8_t*>(&body[0]), body.size()).get();`
`92`		`- return pplx::task_from_result(std::move(body));`
	`81`	`+ return pplx::task_from_result(std::move(m_body.collection()));`
`93`	`82`	`}`
`94`	`83`
`95`	`84`	`}}}}`
Original file line number	Diff line number	Diff line change
`@@ -61,10 +61,10 @@ ref class ReceiveContext sealed`
`61`	`61`
`62`	`62`	`private:`
`63`	`63`	`// Public members cannot have native types`
`64`		`- ReceiveContext(std::function<void(std::shared_ptr<websocket_incoming_message>)> receive_handler, std::function<void()> close_handler): m_receive_handler(receive_handler), m_close_handler(close_handler) {}`
	`64`	`+ ReceiveContext(std::function<void(websocket_incoming_message &>)> receive_handler, std::function<void()> close_handler): m_receive_handler(receive_handler), m_close_handler(close_handler) {}`
`65`	`65`
`66`	`66`	`// Handler to be executed when a message has been received by the client`
`67`		`- std::function<void(std::shared_ptr<websocket_incoming_message>)> m_receive_handler;`
	`67`	`+ std::function<void(websocket_incoming_message &)> m_receive_handler;`
`68`	`68`
`69`	`69`	`// Handler to be executed when a close message has been received by the client`
`70`	`70`	`std::function<void()> m_close_handler;`
`@@ -108,15 +108,14 @@ class winrt_client : public _websocket_client_impl, public std::enable_shared_fr`
`108`	`108`	`Platform::StringReference(client_config.credentials().password().c_str()));`
`109`	`109`	`}`
`110`	`110`
`111`		`- m_context = ref new ReceiveContext([=](std::shared_ptr<websocket_incoming_message> msg)`
	`111`	`+ m_context = ref new ReceiveContext([=](websocket_incoming_message &msg)`
`112`	`112`	`{`
`113`		`- _ASSERTE(msg != nullptr);`
`114`	`113`	`pplx::task_completion_event<websocket_incoming_message> tce; // This will be set if there are any tasks waiting to receive a message`
`115`	`114`	`{`
`116`	`115`	`std::lock_guard<std::mutex> lock(m_receive_queue_lock);`
`117`	`116`	`if (m_receive_task_queue.empty()) // Push message to the queue as no one is waiting to receive`
`118`	`117`	`{`
`119`		`- m_receive_msg_queue.push(std::move(*msg));`
	`118`	`+ m_receive_msg_queue.push(msg);`
`120`	`119`	`return;`
`121`	`120`	`}`
`122`	`121`	`else // There are tasks waiting to receive a message.`
`@@ -126,7 +125,7 @@ class winrt_client : public _websocket_client_impl, public std::enable_shared_fr`
`126`	`125`	`}`
`127`	`126`	`}`
`128`	`127`	`// Setting the tce outside the receive lock for better performance`
`129`		`- tce.set(*msg);`
	`128`	`+ tce.set(msg);`
`130`	`129`	`},`
`131`	`130`	`[=]() // Close handler called upon receiving a close frame from the server.`
`132`	`131`	`{`
`@@ -253,7 +252,7 @@ class winrt_client : public _websocket_client_impl, public std::enable_shared_fr`
`253`	`252`	`void send_msg(websocket_outgoing_message &msg)`
`254`	`253`	`{`
`255`	`254`	`auto this_client = this->shared_from_this();`
`256`		`- auto& is_buf = msg._m_impl->body().create_istream();`
	`255`	`+ auto& is_buf = msg.m_body.create_istream();`
`257`	`256`	`auto length = msg._m_impl->length();`
`258`	`257`
`259`	`258`	`if (length == SIZE_MAX)`
`@@ -274,10 +273,9 @@ class winrt_client : public _websocket_client_impl, public std::enable_shared_fr`
`274`	`273`	`else`
`275`	`274`	`{`
`276`	`275`	`// The stream needs to be buffered.`
`277`		`- concurrency::streams::container_buffer<std::vector<uint8_t>> stbuf;`
`278`	`276`	`auto is_buf_istream = is_buf.create_istream();`
`279`		`- msg._m_impl->set_body(stbuf);`
`280`		`- is_buf_istream.read_to_end(stbuf).then([this_client, msg](pplx::task<size_t> t)`
	`277`	`+ msg.m_body = concurrency::streams::container_buffer<std::vector<uint8_t>>();`
	`278`	`+ is_buf_istream.read_to_end(msg.m_body).then([this_client, msg](pplx::task<size_t> t)`
`281`	`279`	`{`
`282`	`280`	`try`
`283`	`281`	`{`
`@@ -336,7 +334,7 @@ class winrt_client : public _websocket_client_impl, public std::enable_shared_fr`
`336`	`334`
`337`	`335`	`// Send the data as one complete message, in WinRT we do not have an option to send fragments.`
`338`	`336`	`return pplx::task<unsigned int>(this_client->m_messageWriter->StoreAsync());`
`339`		`- }).then([this_client, msg, acquired, sp_allocated, length] (pplx::task<unsigned int> previousTask)`
	`337`	`+ }).then([this_client, msg, is_buf, acquired, sp_allocated, length](pplx::task<unsigned int> previousTask)`
`340`	`338`	`{`
`341`	`339`	`std::exception_ptr eptr;`
`342`	`340`	`unsigned int bytes_written = 0;`
`@@ -360,7 +358,7 @@ class winrt_client : public _websocket_client_impl, public std::enable_shared_fr`
`360`	`358`
`361`	`359`	`if (acquired)`
`362`	`360`	`{`
`363`		`- msg._m_impl->body().release(sp_allocated.get(), bytes_written);`
	`361`	`+ is_buf.release(sp_allocated.get(), bytes_written);`
`364`	`362`	`}`
`365`	`363`
`366`	`364`	`// Set the send_task_completion_event after calling release.`
`@@ -420,11 +418,6 @@ class winrt_client : public _websocket_client_impl, public std::enable_shared_fr`
`420`	`418`	`return pplx::create_task(m_close_tce);`
`421`	`419`	`}`
`422`	`420`
`423`		`- static std::shared_ptr<details::_websocket_message> &get_impl(websocket_incoming_message msg)`
`424`		`- {`
`425`		`- return msg._m_impl;`
`426`		`- }`
`427`		`-`
`428`	`421`	`private:`
`429`	`422`
`430`	`423`	`// WinRT MessageWebSocket object`
`@@ -469,7 +462,7 @@ class winrt_client : public _websocket_client_impl, public std::enable_shared_fr`
`469`	`462`	`void ReceiveContext::OnReceive(MessageWebSocket^ sender, MessageWebSocketMessageReceivedEventArgs^ args)`
`470`	`463`	`{`
`471`	`464`	`websocket_incoming_message ws_incoming_message;`
`472`		`- auto &msg = winrt_client::get_impl(ws_incoming_message);`
	`465`	`+ auto &msg = ws_incoming_message->_m_impl;`
`473`	`466`
`474`	`467`	`switch(args->MessageType)`
`475`	`468`	`{`
`@@ -481,18 +474,16 @@ void ReceiveContext::OnReceive(MessageWebSocket^ sender, MessageWebSocketMessage`
`481`	`474`	`break;`
`482`	`475`	`}`
`483`	`476`
`484`		`- auto &writebuf = msg->body();`
`485`		`-`
`486`	`477`	`try`
`487`	`478`	`{`
`488`	`479`	`DataReader^ reader = args->GetDataReader();`
`489`	`480`	`const auto len = reader->UnconsumedBufferLength;`
`490`		`- auto block = writebuf.alloc(len);`
`491`		`- reader->ReadBytes(Platform::ArrayReference<uint8_t>(block, len));`
`492`		`- writebuf.commit(len);`
`493`		`- writebuf.close(std::ios::out).wait(); // Since this is an in-memory stream, this call is not blocking.`
	`481`	`+ std::string payload;`
	`482`	`+ payload.resize(len);`
	`483`	`+ reader->ReadBytes(Platform::ArrayReference<uint8_t>(payload.c_str(), len));`
	`484`	`+ ws_incoming_message.m_body = concurrency::streams::container_buffer<std::string>(std::move(payload));`
`494`	`485`	`msg->signal_msg_received(len);`
`495`		`- m_receive_handler(std::make_shared<websocket_incoming_message>(ws_incoming_message));`
	`486`	`+ m_receive_handler(ws_incoming_message);`
`496`	`487`	`}`
`497`	`488`	`catch(...)`
`498`	`489`	`{`
Original file line number	Diff line number	Diff line change
`@@ -108,6 +108,9 @@ TEST_FIXTURE(uri_address, receive_text_msg)`
`108`	`108`
`109`	`109`	`receive_text_msg_helper(client, server, m_uri, "hello").wait();`
`110`	`110`	`client.close().wait();`
	`111`	`+`
	`112`	`+ // TODO stgates`
	`113`	`+ VERIFY_ARE_EQUAL(false, true);`
`111`	`114`	`}`
`112`	`115`
`113`	`116`	`// Receive text message (no fragmentation)`