websocket client fixes (#2071)

* websocket client fixes

* Update websocket-client.md

* Update websocket_client_test.cpp

* Update websocket_client_test.cpp

* Update websocket_client_test.cpp

* Update websocket_client_test.cpp

* more diagnostics

* Update websocket_client_test.cpp

* Update websocket_connection.hpp

* Update websocket_connection.hpp

* Update websocket_client_test.cpp

* updates

* fix payload length cast for 32bit

Author: stephenberry

docs/networking/websocket-client.md

@@ -74,7 +74,7 @@ namespace glz
       }
 
       std::string protocol(url.substr(0, protocol_end));
-      if (protocol != "http" && protocol != "https") {
+      if (protocol != "http" && protocol != "https" && protocol != "ws" && protocol != "wss") {
          return std::unexpected(std::make_error_code(std::errc::invalid_argument));
       }
 
@@ -120,7 +120,7 @@ namespace glz
 
       uint16_t port = 0;
       if (port_str.empty()) {
-         port = (protocol == "https") ? 443 : 80;
+         port = (protocol == "https" || protocol == "wss") ? 443 : 80;
       }
       else {
          try {

include/glaze/net/http_client.hpp

@@ -1377,7 +1377,7 @@ namespace glz
 
          // Create WebSocket connection and start it
          // Need to include websocket_connection.hpp for this to work
-         auto ws_conn = std::make_shared<websocket_connection>(std::move(*socket), ws_it->second.get());
+         auto ws_conn = std::make_shared<websocket_connection<asio::ip::tcp::socket>>(socket, ws_it->second.get());
          ws_conn->start(req);
       }
 

include/glaze/net/http_server.hpp

@@ -1,6 +1,7 @@
 #pragma once
 
 #include <mutex>
+#include <random>
 #include <variant>
 
 #include "glaze/net/http_client.hpp"
@@ -30,13 +31,13 @@ namespace glz
 
       std::shared_ptr<asio::io_context> ctx_;
       connection_variant connection_;
-      std::mutex connection_mutex_;
+      std::shared_ptr<std::mutex> connection_mutex_;
 
-      // Callbacks
-      message_handler_t on_message_;
-      open_handler_t on_open_;
-      close_handler_t on_close_;
-      error_handler_t on_error_;
+      // Callbacks - wrapped in shared_ptr to survive client destruction
+      std::shared_ptr<message_handler_t> on_message_;
+      std::shared_ptr<open_handler_t> on_open_;
+      std::shared_ptr<close_handler_t> on_close_;
+      std::shared_ptr<error_handler_t> on_error_;
 
       // Keep alive the resolver/socket during connection
       std::shared_ptr<asio::ip::tcp::resolver> resolver_;
@@ -52,12 +53,16 @@ namespace glz
             ctx_ = ctx;
          else
             ctx_ = std::make_shared<asio::io_context>();
+
+         connection_mutex_ = std::make_shared<std::mutex>();
       }
 
-      void on_message(message_handler_t handler) { on_message_ = std::move(handler); }
-      void on_open(open_handler_t handler) { on_open_ = std::move(handler); }
-      void on_close(close_handler_t handler) { on_close_ = std::move(handler); }
-      void on_error(error_handler_t handler) { on_error_ = std::move(handler); }
+      ~websocket_client() = default;
+
+      void on_message(message_handler_t handler) { on_message_ = std::make_shared<message_handler_t>(std::move(handler)); }
+      void on_open(open_handler_t handler) { on_open_ = std::make_shared<open_handler_t>(std::move(handler)); }
+      void on_close(close_handler_t handler) { on_close_ = std::make_shared<close_handler_t>(std::move(handler)); }
+      void on_error(error_handler_t handler) { on_error_ = std::make_shared<error_handler_t>(std::move(handler)); }
 
       void set_max_message_size(size_t size) { max_message_size_ = size; }
 
@@ -67,7 +72,7 @@ namespace glz
       {
          auto url_result = parse_url(url_str);
          if (!url_result) {
-            if (on_error_) on_error_(url_result.error());
+            if (on_error_ && *on_error_) (*on_error_)(url_result.error());
             return;
          }
 
@@ -91,7 +96,7 @@ namespace glz
                return;
             }
 #else
-            if (on_error_) on_error_(std::make_error_code(std::errc::protocol_not_supported));
+            if (on_error_ && *on_error_) (*on_error_)(std::make_error_code(std::errc::protocol_not_supported));
             return;
 #endif
          }
@@ -101,30 +106,33 @@ namespace glz
 
          resolver_ = std::make_shared<asio::ip::tcp::resolver>(*ctx_);
 
+         auto error_handler = on_error_;  // Copy shared_ptr
          resolver_->async_resolve(
             url.host, std::to_string(url.port),
-            [this, url](std::error_code ec, asio::ip::tcp::resolver::results_type results) {
+            [this, url, error_handler](std::error_code ec, asio::ip::tcp::resolver::results_type results) {
                if (ec) {
-                  if (on_error_) on_error_(ec);
+                  if (error_handler && *error_handler) (*error_handler)(ec);
                   return;
                }
 
                // Determine which socket to connect
                auto& socket_ref = get_tcp_socket_ref();
 
+               auto error_handler2 = on_error_;  // Copy shared_ptr
                asio::async_connect(
-                  socket_ref, results, [this, url](std::error_code ec, const asio::ip::tcp::endpoint&) {
+                  socket_ref, results, [this, url, error_handler2](std::error_code ec, const asio::ip::tcp::endpoint&) {
                      if (ec) {
-                        if (on_error_) on_error_(ec);
+                        if (error_handler2 && *error_handler2) (*error_handler2)(ec);
                         return;
                      }
 
                      if (url.protocol == "wss") {
 #ifdef GLZ_ENABLE_SSL
                         // Perform SSL Handshake
-                        ssl_socket_->async_handshake(asio::ssl::stream_base::client, [this, url](std::error_code ec) {
+                        auto error_handler3 = on_error_;
+                        ssl_socket_->async_handshake(asio::ssl::stream_base::client, [this, url, error_handler3](std::error_code ec) {
                            if (ec) {
-                              if (on_error_) on_error_(ec);
+                              if (error_handler3 && *error_handler3) (*error_handler3)(ec);
                               return;
                            }
                            perform_handshake(ssl_socket_, url);
@@ -140,7 +148,10 @@ namespace glz
 
       void send(std::string_view msg)
       {
-         std::lock_guard<std::mutex> lock(connection_mutex_);
+         auto mutex = connection_mutex_;  // Copy shared_ptr to keep mutex alive
+         if (!mutex) return;
+
+         std::lock_guard<std::mutex> lock(*mutex);
          std::visit(
             [&](auto&& conn) {
                if constexpr (!std::is_same_v<std::decay_t<decltype(conn)>, std::monostate>) {
@@ -150,9 +161,27 @@ namespace glz
             connection_);
       }
 
+      void send_binary(std::string_view msg)
+      {
+         auto mutex = connection_mutex_;  // Copy shared_ptr to keep mutex alive
+         if (!mutex) return;
+
+         std::lock_guard<std::mutex> lock(*mutex);
+         std::visit(
+            [&](auto&& conn) {
+               if constexpr (!std::is_same_v<std::decay_t<decltype(conn)>, std::monostate>) {
+                  if (conn) conn->send_binary(msg);
+               }
+            },
+            connection_);
+      }
+
       void close()
       {
-         std::lock_guard<std::mutex> lock(connection_mutex_);
+         auto mutex = connection_mutex_;  // Copy shared_ptr to keep mutex alive
+         if (!mutex) return;
+
+         std::lock_guard<std::mutex> lock(*mutex);
          std::visit(
             [&](auto&& conn) {
                if constexpr (!std::is_same_v<std::decay_t<decltype(conn)>, std::monostate>) {
@@ -188,11 +217,12 @@ namespace glz
                                  "\r\n" + "Sec-WebSocket-Version: 13\r\n\r\n";
 
          auto req_buf = std::make_shared<std::string>(std::move(handshake));
+         auto error_handler = on_error_;  // Copy shared_ptr
 
          asio::async_write(*socket, asio::buffer(*req_buf),
-                           [this, socket, req_buf /* keep alive */, key](std::error_code ec, std::size_t) {
+                           [this, socket, req_buf, key, error_handler](std::error_code ec, std::size_t) {
                               if (ec) {
-                                 if (on_error_) on_error_(ec);
+                                 if (error_handler && *error_handler) (*error_handler)(ec);
                                  return;
                               }
                               read_handshake_response(socket, key);
@@ -205,11 +235,17 @@ namespace glz
          // Limit handshake response size to 16KB to prevent DoS
          static constexpr size_t max_handshake_size = 1024 * 16;
          auto response_buf = std::make_shared<asio::streambuf>(max_handshake_size);
+         auto error_handler = on_error_;  // Copy shared_ptr
+         auto message_handler = on_message_;
+         auto close_handler = on_close_;
+         auto open_handler = on_open_;
+         auto mutex = connection_mutex_;
+         auto max_msg_size = max_message_size_;
 
          asio::async_read_until(*socket, *response_buf, "\r\n\r\n",
-                                [this, socket, response_buf, expected_key](std::error_code ec, std::size_t) {
+                                [this, socket, response_buf, expected_key, error_handler, message_handler, close_handler, open_handler, mutex, max_msg_size](std::error_code ec, std::size_t) {
                                    if (ec) {
-                                      if (on_error_) on_error_(ec);
+                                      if (error_handler && *error_handler) (*error_handler)(ec);
                                       return;
                                    }
 
@@ -222,7 +258,7 @@ namespace glz
                                    std::getline(response_stream, status_message);
 
                                    if (!response_stream || status_code != 101) {
-                                      if (on_error_) on_error_(std::make_error_code(std::errc::protocol_error));
+                                      if (error_handler && *error_handler) (*error_handler)(std::make_error_code(std::errc::protocol_error));
                                       return;
                                    }
 
@@ -262,33 +298,35 @@ namespace glz
                                    }
 
                                    if (!upgrade_websocket || !connection_upgrade || !accept_key_valid) {
-                                      if (on_error_) on_error_(std::make_error_code(std::errc::protocol_error));
+                                      if (error_handler && *error_handler) (*error_handler)(std::make_error_code(std::errc::protocol_error));
                                       return;
                                    }
 
                                    // Handshake successful. Transfer socket to websocket_connection.
                                    auto ws_conn =
-                                      std::make_shared<websocket_connection<SocketType>>(std::move(*socket));
+                                      std::make_shared<websocket_connection<SocketType>>(socket);
                                    ws_conn->set_client_mode(true);
-                                   ws_conn->set_max_message_size(max_message_size_);
+                                   ws_conn->set_max_message_size(max_msg_size);
 
                                    if (response_buf->size() > 0) {
                                       std::string_view initial_data{
                                          static_cast<const char*>(response_buf->data().data()), response_buf->size()};
                                       ws_conn->set_initial_data(initial_data);
                                    }
 
-                                   if (on_message_) ws_conn->on_message(on_message_);
-                                   if (on_close_) ws_conn->on_close(on_close_);
-                                   if (on_error_) ws_conn->on_error(on_error_);
+                                   if (message_handler && *message_handler) ws_conn->on_message(*message_handler);
+                                   if (close_handler && *close_handler) ws_conn->on_close(*close_handler);
+                                   if (error_handler && *error_handler) ws_conn->on_error(*error_handler);
 
                                    ws_conn->start_read();
                                    {
-                                      std::lock_guard<std::mutex> lock(connection_mutex_);
-                                      connection_ = ws_conn;
+                                      if (mutex) {
+                                         std::lock_guard<std::mutex> lock(*mutex);
+                                         connection_ = ws_conn;
+                                      }
                                    }
 
-                                   if (on_open_) on_open_();
+                                   if (open_handler && *open_handler) (*open_handler)();
                                 });
       }
    };