upgrade slick-net dependency, and refactor WebSocketClient management so that WebsocketClient can reconnect without creating a new object

kzhdev · kzhdev · commit 1166f0737018 · 2026-06-06T16:43:10.000-05:00
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -12,6 +12,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 - Change log file opening mode to append for data logging
+- Upgraded slick-net dependency from v2.0.0 to v2.1.0
+- Changed `market_data_websocket_` and `user_data_websocket_` members from `shared_ptr` to `unique_ptr`
+- Refactored WebSocketClient destructor to call `reset_callbacks()` before closing sockets, eliminating busy-wait polling loops on disconnect
+- Removed atomic `pending_md_socket_close_` and `pending_user_socket_close_` counters
+- `stop()` no longer resets websocket pointers; connection state is preserved for reconnect
+- Disconnect callbacks no longer reset websocket pointers
+- `subscribe()` now checks socket status to reopen a disconnected (but existing) connection instead of only creating on null
+- Heartbeat subscription is now sent immediately after `open()` on user data socket creation
+
+### Fixed
+- `unsubscribe()` now holds a reference to the `unique_ptr` instead of copying it
 
 ## [0.2.2] - 2026-03-05
 
diff --git a/CMakeLists.txt b/CMakeLists.txt
@@ -18,14 +18,14 @@ find_package(nlohmann_json CONFIG REQUIRED)
 find_package(OpenSSL CONFIG REQUIRED)
 find_package(jwt-cpp CONFIG REQUIRED)
 
-find_package(slick-net 2.0.0 CONFIG QUIET)
+find_package(slick-net 2.1.0 CONFIG QUIET)
 if (NOT slick-net_FOUND)
     message(STATUS "fetching slick-net...")
     include(FetchContent)
     FetchContent_Declare(
         slick-net
         GIT_REPOSITORY https://github.com/SlickQuant/slick-net.git
-        GIT_TAG v2.0.0
+        GIT_TAG v2.1.0
     )
     FetchContent_MakeAvailable(slick-net)
 else()
diff --git a/include/coinbase/websocket.hpp b/include/coinbase/websocket.hpp
@@ -182,8 +182,8 @@ class WebSocketClient {
     DataHandler* data_handler_ = nullptr;
     std::string market_data_url_;
     std::string user_data_url_;
-    std::shared_ptr<Websocket> market_data_websocket_;
-    std::shared_ptr<Websocket> user_data_websocket_;
+    std::unique_ptr<Websocket> market_data_websocket_;
+    std::unique_ptr<Websocket> user_data_websocket_;
     std::array<std::unordered_set<std::string>, static_cast<uint8_t>(WebSocketChannel::__COUNT__)> product_ids_;
     std::vector<std::string> pending_subscriptions_;
     std::string user_id_;
@@ -192,8 +192,6 @@ class WebSocketClient {
     slick::SlickQueue<char> *data_queue_ = nullptr;
     std::thread logger_thread_;
     std::atomic_bool logger_run_ = false;
-    std::atomic_int_fast8_t pending_md_socket_close_ = 0;
-    std::atomic_int_fast8_t pending_user_socket_close_ = 0;
     uint64_t data_cursor_ = 0;
     static inline constexpr char empty_msg = '\0';
 };
diff --git a/src/websocket.cpp b/src/websocket.cpp
@@ -160,24 +160,15 @@ WebSocketClient::WebSocketClient(
 WebSocketClient::~WebSocketClient() {
     if (market_data_websocket_) {
         if (market_data_websocket_->status() != Websocket::Status::DISCONNECTED) {
+            market_data_websocket_->reset_callbacks();
             market_data_websocket_->close();
-            if (pending_md_socket_close_.fetch_add(1, std::memory_order_acq_rel) == 0) {
-                // waiting for on disconnected callback
-                auto start = std::chrono::system_clock::now();
-                while(pending_md_socket_close_.load(std::memory_order_acquire) == 1 &&
-                      (std::chrono::system_clock::now() - start) < std::chrono::seconds(2));
-            }
         }
         market_data_websocket_.reset();
     }
     if (user_data_websocket_) {
         if (user_data_websocket_->status() != Websocket::Status::DISCONNECTED) {
+            user_data_websocket_->reset_callbacks();
             user_data_websocket_->close();
-            if (pending_user_socket_close_.fetch_add(1, std::memory_order_acq_rel) == 0) {
-                auto start = std::chrono::system_clock::now();
-                while(pending_user_socket_close_.load(std::memory_order_acquire) == 1 &&
-                      (std::chrono::system_clock::now() - start) < std::chrono::seconds(2));
-            }
         }
         user_data_websocket_.reset();
     }
@@ -206,55 +197,65 @@ void WebSocketClient::stop() {
         if (market_data_websocket_->status() != Websocket::Status::DISCONNECTED) {
             market_data_websocket_->close();
         }
-        market_data_websocket_.reset();
     }
     if (user_data_websocket_) {
         if (user_data_websocket_->status() != Websocket::Status::DISCONNECTED) {
             user_data_websocket_->close();
         }
-        user_data_websocket_.reset();
     }
 }
 
 void WebSocketClient::subscribe(const std::vector<std::string> &product_ids, const std::vector<WebSocketChannel> &channels) {
     for (auto channel : channels) {
         auto products = product_ids_[static_cast<uint8_t>(channel)];
         auto subscribe_json = json{{"type", "subscribe"}, {"product_ids", product_ids}, {"channel", to_string(channel)}};
-        std::shared_ptr<Websocket> websocket;
+        Websocket* websocket;
         if (channel == WebSocketChannel::USER) {
-            if (!user_data_websocket_ && !user_data_url_.empty()) {
-                user_data_websocket_ = std::make_shared<Websocket>(
-                    user_data_url_,
-                    [this]() { onUserDataConnected(); },
-                    [this]() { onUserDataDisconnected(); },
-                    [this](const char* data, std::size_t size) { onUserData(data, size); },
-                    [this](std::string err) { onUserDataError(std::move(err)); }
-                );
+            if (!user_data_websocket_) {
+                if (!user_data_url_.empty()) {
+                    user_data_websocket_ = std::make_unique<Websocket>(
+                        user_data_url_,
+                        [this]() { onUserDataConnected(); },
+                        [this]() { onUserDataDisconnected(); },
+                        [this](const char* data, std::size_t size) { onUserData(data, size); },
+                        [this](std::string err) { onUserDataError(std::move(err)); }
+                    );
+                    user_data_websocket_->open();
+    
+                    // subscribe heartbeat to keep user channel alive
+                    auto heartbeat_sub = json{{"type", "subscribe"}, {"channel", "heartbeats"}};
+                    heartbeat_sub["jwt"] = generate_coinbase_jwt(user_data_url_.c_str());
+                    auto subscribe_str = heartbeat_sub.dump();
+                    user_data_websocket_->send(subscribe_str.c_str(), subscribe_str.size());
+                }
+            } else if (user_data_websocket_->status() > Websocket::Status::CONNECTED) {
                 user_data_websocket_->open();
-                pending_user_socket_close_.store(0, std::memory_order_release);
 
                 // subscribe heartbeat to keep user channel alive
                 auto heartbeat_sub = json{{"type", "subscribe"}, {"channel", "heartbeats"}};
                 heartbeat_sub["jwt"] = generate_coinbase_jwt(user_data_url_.c_str());
                 auto subscribe_str = heartbeat_sub.dump();
                 user_data_websocket_->send(subscribe_str.c_str(), subscribe_str.size());
             }
-            websocket = user_data_websocket_;
+            websocket = user_data_websocket_.get();
             subscribe_json["jwt"] = generate_coinbase_jwt(user_data_url_.c_str());
         }
         else {
-            if (!market_data_websocket_ && !market_data_url_.empty()) {
-                market_data_websocket_ = std::make_shared<Websocket>(
-                    market_data_url_,
-                    [this]() { onMarketDataConnected(); },
-                    [this]() { onMarketDataDisconnected(); },
-                    [this](const char* data, std::size_t size) { onMarketData(data, size); },
-                    [this](std::string err) { onMarketDataError(std::move(err)); }
-                );
+            if (!market_data_websocket_) {
+                if (!market_data_url_.empty()) {
+                    market_data_websocket_ = std::make_unique<Websocket>(
+                        market_data_url_,
+                        [this]() { onMarketDataConnected(); },
+                        [this]() { onMarketDataDisconnected(); },
+                        [this](const char* data, std::size_t size) { onMarketData(data, size); },
+                        [this](std::string err) { onMarketDataError(std::move(err)); }
+                    );
+                    market_data_websocket_->open();
+                }
+            } else if (market_data_websocket_->status() > Websocket::Status::CONNECTED) {
                 market_data_websocket_->open();
-                pending_md_socket_close_.store(0, std::memory_order_release);
             }
-            websocket = market_data_websocket_;
+            websocket = market_data_websocket_.get();
         }
 
         if (websocket == nullptr) {
@@ -269,7 +270,7 @@ void WebSocketClient::subscribe(const std::vector<std::string> &product_ids, con
 
 void WebSocketClient::unsubscribe(const std::vector<std::string> &product_ids, const std::vector<WebSocketChannel> &channels) {
     for (auto channel : channels) {
-        auto websocket = channel == WebSocketChannel::USER ? user_data_websocket_ : market_data_websocket_;
+        auto &websocket = channel == WebSocketChannel::USER ? user_data_websocket_ : market_data_websocket_;
         auto unsubscribe_json = json{{"type", "unsubscribe"}, {"product_ids", product_ids}, {"channel", to_string(channel)}};
         auto unsubscribe_str = unsubscribe_json.dump();
         if (websocket == nullptr) {
@@ -369,8 +370,6 @@ void WebSocketClient::onMarketDataDisconnected() {
         data_handler_->callbacks_->onMarketDataDisconnected(this);
         data_handler_->resetMarketDataSequence(this);
     }
-    pending_md_socket_close_.fetch_add(1, std::memory_order_acq_rel);
-    market_data_websocket_.reset();
 }
 
 
@@ -391,8 +390,6 @@ void WebSocketClient::onUserDataDisconnected() {
         data_handler_->callbacks_->onUserDataDisconnected(this);
         data_handler_->resetUserDataSequence(this);
     }
-    pending_user_socket_close_.fetch_add(1, std::memory_order_acq_rel);
-    user_data_websocket_.reset();
 }
 
 void WebSocketClient::onMarketData(const char* data, std::size_t size) {
diff --git a/tests/websocket_tests.cpp b/tests/websocket_tests.cpp
@@ -439,18 +439,18 @@ namespace coinbase::tests {
 
     TEST_F(WebSocketTests, RepeatedConnectDisconnect) {
         constexpr int kIterations = 5;
+        WebSocketClient client_(this);
         for (int i = 0; i < kIterations; ++i) {
-            client_ = std::make_unique<WebSocketClient>(this);
             auto connected_before = md_connected_count_.load(std::memory_order_relaxed);
             auto disconnected_before = md_disconnected_count_.load(std::memory_order_relaxed);
-            client_->subscribe({"BTC-USD"}, {WebSocketChannel::LEVEL2});
+            client_.subscribe({"BTC-USD"}, {WebSocketChannel::LEVEL2});
             auto start = std::chrono::steady_clock::now();
             while (md_connected_count_.load(std::memory_order_relaxed) == connected_before &&
                    (std::chrono::steady_clock::now() - start) < std::chrono::seconds(5)) {
                 std::this_thread::sleep_for(std::chrono::milliseconds(50));
             }
             const bool connected = md_connected_count_.load(std::memory_order_relaxed) > connected_before;
-            client_.reset();
+            client_.stop();
             if (connected) {
                 start = std::chrono::steady_clock::now();
                 while (md_disconnected_count_.load(std::memory_order_relaxed) == disconnected_before &&
