Saving some space in websocket messages and container_buffer.

Author: stgates

Release/include/cpprest/containerstream.h

@@ -120,7 +120,7 @@ namespace Concurrency { namespace streams {
         /// </summary>
         virtual utility::size64_t size() const
         {
-            return utility::size64_t(m_size);
+            return utility::size64_t(m_data.size());
         }
 
         /// <summary>
@@ -153,10 +153,10 @@ namespace Concurrency { namespace streams {
         {
             // See the comment in seek around the restriction that we do not allow read head to
             // seek beyond the current write_end.
-            _ASSERTE(m_current_position <= m_size);
+            _ASSERTE(m_current_position <= m_data.size());
 
             SafeSize readhead(m_current_position);
-            SafeSize writeend(m_size);
+            SafeSize writeend(m_data.size());
             return (size_t)(writeend - readhead);
         }
 
@@ -330,7 +330,7 @@ namespace Concurrency { namespace streams {
             // For now, we assume that the current write_end is the end of the buffer. We use this artificial
             // end to restrict the read head from seeking beyond what is available.
 
-            pos_type end(m_size);
+            pos_type end(m_data.size());
 
             if (position >= beg)
             {
@@ -378,7 +378,7 @@ namespace Concurrency { namespace streams {
         {
             pos_type beg = 0;
             pos_type cur = static_cast<pos_type>(m_current_position);
-            pos_type end = static_cast<pos_type>(m_size);
+            pos_type end = static_cast<pos_type>(m_data.size());
 
             switch ( way )
             {
@@ -404,8 +404,7 @@ namespace Concurrency { namespace streams {
         /// </summary>
         basic_container_buffer(std::ios_base::openmode mode)
             : streambuf_state_manager<typename _CollectionType::value_type>(mode),
-              m_current_position(0),
-              m_size(0)
+              m_current_position(0)
         {
             validate_mode(mode);
         }
@@ -416,8 +415,7 @@ namespace Concurrency { namespace streams {
         basic_container_buffer(_CollectionType data, std::ios_base::openmode mode)
             : streambuf_state_manager<typename _CollectionType::value_type>(mode),
               m_data(std::move(data)),
-              m_current_position((mode & std::ios_base::in) ? 0 : m_data.size()),
-              m_size(m_data.size())
+              m_current_position((mode & std::ios_base::in) ? 0 : m_data.size())
         {
             validate_mode(mode);
         }
@@ -509,10 +507,8 @@ namespace Concurrency { namespace streams {
         /// </summary>
         void resize_for_write(size_t newPos)
         {
-            _ASSERTE(m_size <= m_data.size());
-
             // Resize the container if required
-            if (newPos > m_size)
+            if (newPos > m_data.size())
             {
                 m_data.resize(newPos);
             }
@@ -525,23 +521,14 @@ namespace Concurrency { namespace streams {
         {
             // The new write head
             m_current_position = newPos;
-
-            if ( this->can_write() && m_size < m_current_position)
-            {
-                // Update the size of the buffer with valid data if required
-                m_size = m_current_position;
-            }
-
-            _ASSERTE(m_current_position <= m_size);
-            _ASSERTE(m_size <= m_data.size());
+            _ASSERTE(m_current_position <= m_data.size());
         }
 
         // The actual data store
         _CollectionType m_data;
 
         // Read/write head
         size_t m_current_position;
-        size_t m_size;
     };
 
     } // namespace details

Release/include/cpprest/ws_msg.h

@@ -77,46 +77,20 @@ enum class websocket_message_type
     pong
 };
 
-namespace details
-{
-class _websocket_message
-{
-public:
-
-    void set_msg_type(websocket_message_type msg_type) { m_msg_type = msg_type; }
-
-    void set_length(size_t len) { m_length = len; }
-
-    size_t length() const { return m_length; }
-
-    websocket_message_type message_type() const { return m_msg_type; }
-
-private:
-
-    websocket_message_type m_msg_type;
-    size_t m_length;
-};
-}
-
 /// <summary>
 /// Represents an outgoing websocket message
 /// </summary>
 class websocket_outgoing_message
 {
 public:
 
-    /// <summary>
-    /// Creates an initially empty message for sending.
-    /// </summary>
-    websocket_outgoing_message() : _m_impl(std::make_shared<details::_websocket_message>()) {}
-
     /// <summary>
     /// Sets a UTF-8 message as the message body.
     /// </summary>
     /// <param name="data">UTF-8 String containing body of the message.</param>
     void set_utf8_message(std::string &&data)
     {
-        this->_set_message(std::move(data), websocket_message_type::text_message);
+        this->set_message(concurrency::streams::container_buffer<std::string>(std::move(data)));
     }
 
     /// <summary>
@@ -125,7 +99,7 @@ class websocket_outgoing_message
     /// <param name="data">UTF-8 String containing body of the message.</param>
     void set_utf8_message(const std::string &data)
     {
-        this->_set_message(data, websocket_message_type::text_message);
+        this->set_message(concurrency::streams::container_buffer<std::string>(data));
     }
 
     /// <summary>
@@ -135,7 +109,7 @@ class websocket_outgoing_message
     /// <remarks>Upon sending, the entire stream may be buffered to determine the length.</remarks>
     void set_utf8_message(const concurrency::streams::istream &istream)
     {
-        this->_set_message(istream, SIZE_MAX, websocket_message_type::text_message);
+        this->set_message(istream, SIZE_MAX, websocket_message_type::text_message);
     }
 
     /// <summary>
@@ -145,7 +119,7 @@ class websocket_outgoing_message
     /// <param name="len">number of bytes to send.</param>
     void set_utf8_message(const concurrency::streams::istream &istream, size_t len)
     {
-        this->_set_message(istream, len, websocket_message_type::text_message);
+        this->set_message(istream, len, websocket_message_type::text_message);
     }
 
     /// <summary>
@@ -155,7 +129,7 @@ class websocket_outgoing_message
     /// <param name="len">number of bytes to send.</param>
     void set_binary_message(const concurrency::streams::istream &istream, size_t len)
     {
-        this->_set_message(istream, len, websocket_message_type::binary_message);
+        this->set_message(istream, len, websocket_message_type::binary_message);
     }
 
     /// <summary>
@@ -165,17 +139,17 @@ class websocket_outgoing_message
     /// <remarks>Upon sending, the entire stream may be buffered to determine the length.</remarks>
     void set_binary_message(const concurrency::streams::istream &istream)
     {
-        this->_set_message(istream, SIZE_MAX, websocket_message_type::binary_message);
+        this->set_message(istream, SIZE_MAX, websocket_message_type::binary_message);
     }
 
 private:
     friend class details::winrt_client;
     friend class details::wspp_client;
 
-    std::shared_ptr<details::_websocket_message> _m_impl;
-
     pplx::task_completion_event<void> m_body_sent;
     concurrency::streams::streambuf<uint8_t> m_body;
+	websocket_message_type m_msg_type;
+    size_t m_length;
 
     void signal_body_sent() const
     {
@@ -189,24 +163,17 @@ class websocket_outgoing_message
 
     const pplx::task_completion_event<void> & body_sent() const { return m_body_sent; }
 
-    void _set_message(std::string &&data, websocket_message_type msg_type)
-    {
-        _m_impl->set_msg_type(msg_type);
-        _m_impl->set_length(data.length());
-        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_body = concurrency::streams::container_buffer<std::string>(data);
-    }
+	void set_message(const concurrency::streams::container_buffer<std::string> &buffer)
+	{
+        m_msg_type = websocket_message_type::text_message;
+        m_length = static_cast<size_t>(buffer.size());
+        m_body = buffer;
+	}
 
-    void _set_message(const concurrency::streams::istream &istream, size_t len, websocket_message_type msg_type)
+    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_length(len);
+        m_msg_type = msg_type;
+        m_length = len;
         m_body = istream.streambuf();
     }
 };
@@ -217,9 +184,6 @@ class websocket_outgoing_message
 class websocket_incoming_message
 {
 public:
-    websocket_incoming_message() : _m_impl(std::make_shared<details::_websocket_message>())
-    {
-    }
 
     /// <summary>
     /// Extracts the body of the incoming message as a string value, only if the message type is UTF-8.
@@ -250,15 +214,15 @@ class websocket_incoming_message
     /// </summary>
     size_t length() const
     {
-        return _m_impl->length();
+        return static_cast<size_t>(m_body.size());
     }
 
     /// <summary>
     /// Returns the type of the received message.
     /// </summary>
     websocket_message_type messge_type() const
     {
-        return _m_impl->message_type();
+        return m_msg_type;
     }
 
 private:
@@ -271,8 +235,7 @@ class websocket_incoming_message
     // 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;
+	websocket_message_type m_msg_type;
 };
 
 }}}

Release/src/websockets/client/ws_client.cpp

@@ -269,34 +269,31 @@ class wspp_client : public _websocket_client_impl, public std::enable_shared_fro
         client.set_message_handler([this](websocketpp::connection_hdl, const websocketpp::config::asio_client::message_type::ptr &msg)
         {
             _ASSERTE(m_state >= CONNECTED && m_state < CLOSED);
-            websocket_incoming_message ws_incoming_message;
-            auto& incmsg = ws_incoming_message._m_impl;
+            websocket_incoming_message incoming_msg;
 
             switch (msg->get_opcode())
             {
             case websocketpp::frame::opcode::binary:
-                incmsg->set_msg_type(websocket_message_type::binary_message);
+                incoming_msg.m_msg_type = websocket_message_type::binary_message;
                 break;
             case websocketpp::frame::opcode::text:
-                incmsg->set_msg_type(websocket_message_type::text_message);
+                incoming_msg.m_msg_type = websocket_message_type::text_message;
                 break;
             default:
                 // Unknown message type. Since both websocketpp and our code use the RFC codes, we'll just pass it on to the user.
-                incmsg->set_msg_type(static_cast<websocket_message_type>(msg->get_opcode()));
+                incoming_msg.m_msg_type = static_cast<websocket_message_type>(msg->get_opcode());
                 break;
             }
 
             // 'move' the payload into a container buffer to avoid any copies.
             auto &payload = msg->get_raw_payload();
-            const auto len = payload.size();
-            ws_incoming_message.m_body = concurrency::streams::container_buffer<std::string>(std::move(payload));
-            incmsg->set_length(len);
+            incoming_msg.m_body = concurrency::streams::container_buffer<std::string>(std::move(payload));
 
             std::unique_lock<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(ws_incoming_message);
+                m_receive_msg_queue.push(incoming_msg);
                 return;
             }
             else
@@ -306,7 +303,7 @@ class wspp_client : public _websocket_client_impl, public std::enable_shared_fro
                 m_receive_task_queue.pop();
                 // Unlock the lock before setting the completion event to avoid contention
                 lock.unlock();
-                tce.set(ws_incoming_message);
+                tce.set(incoming_msg);
             }
         });
 
@@ -370,7 +367,7 @@ class wspp_client : public _websocket_client_impl, public std::enable_shared_fro
             return pplx::task_from_exception<void>(websocket_exception("Client not connected."));
         }
 
-        switch (msg._m_impl->message_type())
+        switch (msg.m_msg_type)
         {
         case websocket_message_type::text_message:
         case websocket_message_type::binary_message:
@@ -379,7 +376,7 @@ class wspp_client : public _websocket_client_impl, public std::enable_shared_fro
             return pplx::task_from_exception<void>(websocket_exception("Invalid message type"));
         }
 
-        const auto length = msg._m_impl->length();
+        const auto length = msg.m_length;
         if (length == 0)
         {
             return pplx::task_from_exception<void>(websocket_exception("Cannot send empty message."));
@@ -449,7 +446,7 @@ class wspp_client : public _websocket_client_impl, public std::enable_shared_fro
     {
         auto this_client = this->shared_from_this();
         auto& is_buf = msg.m_body;
-        auto length = msg._m_impl->length();
+        auto length = msg.m_length;
 
         if (length == SIZE_MAX)
         {
@@ -475,7 +472,7 @@ class wspp_client : public _websocket_client_impl, public std::enable_shared_fro
                 {
                     try
                     {
-                        msg._m_impl->set_length(t.get());
+                        msg.m_length = t.get();
                         this_client->send_msg(msg);
                     }
                     catch (...)
@@ -528,7 +525,7 @@ class wspp_client : public _websocket_client_impl, public std::enable_shared_fro
         {
         	auto &client = this_client->m_client->client<WebsocketClientType>();
             websocketpp::lib::error_code ec;
-            switch (msg._m_impl->message_type())
+            switch (msg.m_msg_type)
             {
             case websocket_message_type::text_message:
                 client.send(

Release/src/websockets/client/ws_msg.cpp

@@ -71,7 +71,7 @@ std::vector<::utility::string_t> websocket_client_config::subprotocols() const
 
 pplx::task<std::string> websocket_incoming_message::extract_string() const
 {
-    if (_m_impl->message_type() == websocket_message_type::binary_message)
+    if (m_msg_type == websocket_message_type::binary_message)
     {
         return pplx::task_from_exception<std::string>(websocket_exception("Invalid message type"));
     }