Handle fragmented message. (#6)

Author: hokyungh

chime-sdk-signaling-cpp/src/websocket/libwebsockets_websocket.cc

@@ -83,10 +83,15 @@ int LibwebsocketsWebsocket::Callback(struct lws* wsi, enum lws_callback_reasons
     case LWS_CALLBACK_CLIENT_RECEIVE: {
       lwsl_debug("Data received.\n");
       lwsl_hexdump_debug(in, len);
+      const size_t remaining = lws_remaining_packet_payload(wsi);
       auto* uint8_ptr = static_cast<uint8_t*>(in);
-      std::vector<uint8_t> data;
-      data.assign(uint8_ptr, uint8_ptr + len);
-      self->observer_->OnWebsocketBinaryReceived(data);
+      // Messages can be fragmented if the size exceeds max bytes
+      // Therefore, it needs to handle fragmented message.
+      self->received_data_buffer_.insert(self->received_data_buffer_.end(), uint8_ptr, uint8_ptr + len);
+      if (!remaining && lws_is_final_fragment(wsi)) {
+        self->observer_->OnWebsocketBinaryReceived(self->received_data_buffer_);
+        self->received_data_buffer_.clear();
+      }
       break;
     }
 
@@ -136,6 +141,7 @@ int LibwebsocketsWebsocket::Callback(struct lws* wsi, enum lws_callback_reasons
       break;
     }
     case LWS_CALLBACK_CLIENT_WRITEABLE: {
+      // TODO: Check if this can be also fragmented
       if (!self->message_queue_.empty()) {
         std::vector<uint8_t> data = self->message_queue_.front();
         self->message_queue_.pop();
@@ -238,6 +244,7 @@ void LibwebsocketsWebsocket::Poll() { lws_service(context_, 0); }
 
 void LibwebsocketsWebsocket::Close() {
   if (!context_) return;
+
   lws_context_destroy(context_);
   lwsl_user("Closed\n");
 }
@@ -248,6 +255,7 @@ void LibwebsocketsWebsocket::SendBinary(const std::vector<uint8_t>& data) {
   // Triggers LWS_CALLBACK_CLIENT_WRITEABLE event in Callback when socket is ready to accept data.
   lws_callback_on_writable(wsi_);
 }
+
 int LibwebsocketsWebsocket::ConvertLogLevel(LogLevel level) {
   int lws_level = 0;
   switch (level) {
@@ -267,7 +275,7 @@ int LibwebsocketsWebsocket::ConvertLogLevel(LogLevel level) {
   return lws_level;
 }
 
-void LibwebsocketsWebsocket::HandleError(std::string error_description) {
+void LibwebsocketsWebsocket::HandleError(const std::string& error_description) {
   lwsl_err("%s", error_description.c_str());
   WebsocketErrorStatus error_status;
   error_status.description = error_description;

chime-sdk-signaling-cpp/src/websocket/libwebsockets_websocket.h

@@ -27,7 +27,7 @@ struct LibwebsocketsWebsocketConfiguration : public WebsocketConfiguration {
 class LibwebsocketsWebsocket : public Websocket {
  public:
   LibwebsocketsWebsocket(LibwebsocketsWebsocketConfiguration configuration, WebsocketObserver* observer);
-
+  ~LibwebsocketsWebsocket() = default;
   // chime::signaling::Websocket overrides
   void Connect() override;
   void Close() override;
@@ -41,6 +41,9 @@ class LibwebsocketsWebsocket : public Websocket {
   // Messages are buffered here first until the socket will accept packets without blocking.
   std::queue<std::vector<uint8_t>> message_queue_;
 
+  // Internal data
+  std::vector<uint8_t> received_data_buffer_;
+
   // This policy sets both connection attempt retry parameters and ping/pong parameters.
   // The retry parameters apply to failures that happen before a websocket connection is established.
   lws_retry_bo_t retry_and_idle_policy_;
@@ -83,7 +86,7 @@ class LibwebsocketsWebsocket : public Websocket {
   int ConvertLogLevel(LogLevel level);
 
   // Log error and notify observer.
-  void HandleError(std::string error_description);
+  void HandleError(const std::string& error_description);
 };
 
 }  // namespace chime