draim multiple messages in UserThreadWebsocketCallbacks; Fix various bugs

Author: kzhdev

CHANGELOG.md

@@ -17,10 +17,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Changed header files from .h to .hpp
 - Decoupled UserThreadWebsocketCallbacks from WebSocketeClient to support multiple WebSocketClient
 - Create market data and user data websocket when url is set
+- UserThreadWebsocketCallbacks now drains multiple queued messages per tick for higher throughput
+- Propagate precompile headers
+
 
 ### Fixed
 - Various WebSocket unit tests not waiting for snapshot
 - Fixed duplicated Candle definition
+- WebSocket logger now writes correct payload offsets and labels user data correctly
+- Sequence-number checks now accept first message even if the sequence does not start at 0
+- Level2 book compile issues and trade timestamp handling
+- Trades JSON parsing (pass-by-reference)
+- Empty API secret handling in PEM formatting
+- PriceBookResponse parsing when pricebook is missing
+- Order status string typo and size_ratio field name
+- Missing default return in FCM trading session state parsing
 
 ## [0.1.0] - 2026-01-13 
 

CMakeLists.txt

@@ -1,4 +1,4 @@
-cmake_minimum_required(VERSION 3.20)
+cmake_minimum_required(VERSION 3.30)
 
 set(CMAKE_CXX_STANDARD 20)
 
@@ -42,8 +42,12 @@ target_include_directories(coinbase-advanced-cpp INTERFACE
 target_link_libraries(coinbase-advanced-cpp INTERFACE slick::net jwt-cpp::jwt-cpp)
 target_precompile_headers(coinbase-advanced-cpp INTERFACE 
     <nlohmann/json.hpp>
-    <slick/net/http.h>
-    <slick/net/websocket.h>
+    <random>
+    <fstream>
+)
+
+set_target_properties(coinbase-advanced-cpp PROPERTIES 
+    TRANSITIVE_COMPILE_PROPERTIES "PRECOMPILE_HEADERS"
 )
 
 if (MSVC)

include/coinbase/auth.hpp

@@ -16,8 +16,11 @@
 namespace coinbase {
 
 inline std::string fix_pem_format(std::string key) {
+    if (key.empty()) {
+        return key;
+    }
     // Remove any quotes that might have been added
-    if (key.front() == '"' && key.back() == '"') {
+    if (key.size() >= 2 && key.front() == '"' && key.back() == '"') {
         key = key.substr(1, key.length() - 2);
     }
 

include/coinbase/level2_book.hpp

@@ -6,8 +6,11 @@
 
 #include <string>
 #include <vector>
+#include <map>
+#include <functional>
 #include <nlohmann/json.hpp>
 #include <coinbase/utils.hpp>
+#include <coinbase/market_data.hpp>
 
 using json = nlohmann::json;
 
@@ -21,11 +24,11 @@ struct SideBook {
 
 struct Level2Book {
     std::string product_id;
-    uint64_t last_update_time;
+    uint64_t last_update_time = 0;
     SideBook<Side::BUY> bids;
     SideBook<Side::SELL> asks;
 
-    void onLevel2Snapshot(uin64_t seq_num, const Level2UpdateBatch &batch) {
+    void onLevel2Snapshot(uint64_t seq_num, const Level2UpdateBatch &batch) {
         bids.levels.clear();
         asks.levels.clear();
         for (const auto &update : batch.updates) {
@@ -35,10 +38,12 @@ struct Level2Book {
                 asks.levels.emplace(update.price_level, update.new_quantity);
             }
         }
-        last_update_time = batch.updates.back().event_time;
+        if (!batch.updates.empty()) {
+            last_update_time = batch.updates.back().event_time;
+        }
     }
 
-    void onLevel2Updates(uin64_t seq_num, const Level2UpdateBatch &batch) {
+    void onLevel2Updates(uint64_t seq_num, const Level2UpdateBatch &batch) {
         for (const auto &update : batch.updates) {
             if (update.event_time <= last_update_time) {
                 LOG_TRACE("{} Skipping update event_time {} <= last_update_time {}", product_id, update.event_time, last_update_time);
@@ -71,10 +76,10 @@ struct Level2Book {
         }
     }
 
-    void onMarketTrades(uin64_t seq_num, const std::vector<MarketTrade> &trades) {
+    void onMarketTrades(uint64_t seq_num, const std::vector<MarketTrade> &trades) {
         for (const auto &trade : trades) {
-            if (trade.event_time <= last_update_time) {
-                LOG_TRACE("{} Skipping trade event_time {} <= last_update_time {}", product_id, trade.event_time, last_update_time);
+            if (trade.time <= last_update_time) {
+                LOG_TRACE("{} Skipping trade event_time {} <= last_update_time {}", product_id, trade.time, last_update_time);
                 continue;
             }
             if (trade.side == Side::BUY) {
@@ -94,7 +99,7 @@ struct Level2Book {
                     }
                 }
             }
-            last_update_time = trade.event_time;
+            last_update_time = trade.time;
         }
     }
 };

include/coinbase/order.hpp

@@ -155,7 +155,7 @@ struct ScaledLimitConfig : public MarketConfig{
     std::string price_distribution;
     std::string size_distribution;
     double size_diff;
-    double size_ration;
+    double size_ratio;
 };
 
 inline void from_json(const json &j, ScaledLimitConfig &c) {
@@ -167,7 +167,7 @@ inline void from_json(const json &j, ScaledLimitConfig &c) {
     VARIABLE_FROM_JSON(j, c, price_distribution);
     VARIABLE_FROM_JSON(j, c, size_distribution);
     DOUBLE_FROM_JSON(j, c, size_diff);
-    DOUBLE_FROM_JSON(j, c, size_ration);
+    DOUBLE_FROM_JSON(j, c, size_ratio);
 }
 
 struct OrderConfiguration {
@@ -317,7 +317,7 @@ inline std::string to_string(OrderStatus status) {
     case OrderStatus::QUEUED:
         return "QUEUED";
     case OrderStatus::CANCEL_QUEUED:
-        return "CANCEL_QUEUEQ";
+        return "CANCEL_QUEUED";
     case OrderStatus::EDIT_QUEUED:
         return "EDIT_QUEUED";
     case OrderStatus::UNKNOWN_ORDER_STATUS:

include/coinbase/price_book.hpp

@@ -71,7 +71,6 @@ struct PriceBookResponse {
 
 inline void from_json(const json &j, PriceBookResponse &b) {
     STRUCT_FROM_JSON(j, b, pricebook);
-    b.pricebook = j["pricebook"];
     DOUBLE_FROM_JSON(j, b, last);
     DOUBLE_FROM_JSON(j, b, mid_market);
     DOUBLE_FROM_JSON(j, b, spread_bps);

include/coinbase/product.hpp

@@ -40,7 +40,7 @@ inline FcmTradingSessionState to_fcm_trading_session_state(std::string_view stat
         return FcmTradingSessionState::FCM_TRADING_SESSION_STATE_CLOSE;
     }
 
-    FcmTradingSessionState::FCM_TRADING_SESSION_STATE_UNDEFINED;
+    return FcmTradingSessionState::FCM_TRADING_SESSION_STATE_UNDEFINED;
 }
 
 struct Maintenance {
@@ -152,11 +152,11 @@ inline void from_json(const json &j, FutureProductDetails &d) {
     VARIABLE_FROM_JSON(j, d, twenty_four_by_seven);
 }
 
-struct SettlmentSource {
+struct SettlementSource {
     std::string url;
     std::string name;
 
-    NLOHMANN_DEFINE_TYPE_INTRUSIVE(SettlmentSource, url, name)
+    NLOHMANN_DEFINE_TYPE_INTRUSIVE(SettlementSource, url, name)
 };
 
 struct PredictionMarketProductDetails {
@@ -181,7 +181,7 @@ struct PredictionMarketProductDetails {
     std::string scope;
     std::string yes_titles;
     std::vector<std::string> prehibitions;
-    std::vector<SettlmentSource> settlement_sources;
+    std::vector<SettlementSource> settlement_sources;
     std::vector<std::string> tags;
     uint64_t contract_expiry;
     uint64_t trade_starting_time;
@@ -213,7 +213,7 @@ inline void from_json(const json &j, PredictionMarketProductDetails &p) {
     VARIABLE_FROM_JSON(j, p, market_cbrn);
     VARIABLE_FROM_JSON(j, p, scope);
     VARIABLE_FROM_JSON(j, p, yes_titles);
-    p.yes_titles = j["prehibitions"];
+    p.prehibitions = j["prehibitions"];
     p.settlement_sources = j["settlement_sources"];
     p.tags = j["tags"];
     TIMESTAMP_FROM_JSON(j, p, contract_expiry);
@@ -287,11 +287,11 @@ struct Product {
     bool limit_only;
     bool post_only;
     bool trading_disabled;
-    bool aution_mode;
+    bool auction_mode;
     bool view_only;
     FcmTradingSessionDetails fcm_trading_session_details;
     FutureProductDetails future_product_details;
-    EquityProductDetails equity_project_details;
+    EquityProductDetails equity_product_details;
 };
 
 inline void from_json(const json& j, Product& p) {

include/coinbase/trades.hpp

@@ -25,7 +25,7 @@ struct Trades {
     Side side;
 };
 
-inline void from_json(const json &j, Trades t) {
+inline void from_json(const json &j, Trades &t) {
     VARIABLE_FROM_JSON(j, t, trade_id);
     VARIABLE_FROM_JSON(j, t, product_id);
     VARIABLE_FROM_JSON(j, t, exchange);

include/coinbase/utils.hpp

@@ -149,7 +149,7 @@ inline std::string to_string(double value, Side side, double min_increment) {
 #define DOUBLE_FROM_JSON(j, o, field) o.##field = double_from_json(j, #field)
 #define INT_FROM_JSON(j, o, field) o.##field = int_from_json(j, #field)
 #define VARIABLE_FROM_JSON(j, o, field) if (j.contains(#field)) try { j.at(#field).get_to(o.##field); } catch(const std::exception &e) { LOG_INFO(#field " {}", j.dump()); LOG_ERROR(e.what()); }
-#define BOOL_FROM_JSON(j, o, field) if (!j[#field].is_null() && j[#field].is_string()) { if (j[#field] == "true") o.##field = true; else if (j[#field] == "false") o.##field = false; } else try { j.at(#field).get_to(o.##field); } catch(const std::exception &e) { LOG_INFO(#field " {}", j.dump()); LOG_ERROR(e.what()); }
-#define STRUCT_FROM_JSON(j, o, field) if (!j[#field].is_null()) from_json(j[#field], o.##field)
+#define BOOL_FROM_JSON(j, o, field) if (j.contains(#field) && !j[#field].is_null() && j[#field].is_string()) { if (j[#field] == "true") o.##field = true; else if (j[#field] == "false") o.##field = false; } else try { j.at(#field).get_to(o.##field); } catch(const std::exception &e) { LOG_INFO(#field " {}", j.dump()); LOG_ERROR(e.what()); }
+#define STRUCT_FROM_JSON(j, o, field) if (j.contains(#field) && !j[#field].is_null()) from_json(j[#field], o.##field)
 #define ENUM_FROM_JSON(j, o, field) if (j.contains(#field)) try { o.##field = to_##field(j.at(#field).get<std::string_view>()); } catch(const std::exception &e) { LOG_INFO(#field " {}", j.dump()); LOG_ERROR(e.what()); }
 }   // end namespace coinbase

include/coinbase/websocket.hpp

@@ -131,13 +131,13 @@ struct UserThreadWebsocketCallbacks : public DataHandler, public WebsocketCallba
     ~UserThreadWebsocketCallbacks() override = default;
 
     // Process data in the user thread. Callbacks will be invoked in the user thread.
-    void processData();
+    void processData(uint32_t max_drain_count = 100);
 
 private:
     bool checkMarketDataSequenceNumber(void *ws_client, int64_t seq_num) override {
         auto it = md_seq_nums_.find(ws_client);
         if (it == md_seq_nums_.end()) [[unlikely]] {
-            it = md_seq_nums_.emplace(ws_client, -1).first;
+            it = md_seq_nums_.emplace(ws_client, seq_num - 1).first;
         }
         auto &seq_atomic = it->second;
         auto last_seq = seq_atomic.load(std::memory_order_acquire);
@@ -153,7 +153,7 @@ struct UserThreadWebsocketCallbacks : public DataHandler, public WebsocketCallba
     bool checkUserDataSequenceNumber(void *ws_client, int64_t seq_num) override {
         auto it = user_seq_nums_.find(ws_client);
         if (it == user_seq_nums_.end()) [[unlikely]] {
-            it = user_seq_nums_.emplace(ws_client, -1).first;
+            it = user_seq_nums_.emplace(ws_client, seq_num - 1).first;
         }
         auto &seq_atomic = it->second;
         auto last_seq = seq_atomic.load(std::memory_order_acquire);
@@ -231,9 +231,13 @@ class WebSocketClient {
 
 
 
-inline void UserThreadWebsocketCallbacks::processData() {
-    auto [data_ptr, data_size] = data_queue_.read(read_cursor_);
-    if (data_ptr && data_size > 0) {
+inline void UserThreadWebsocketCallbacks::processData(uint32_t max_drain_count) {
+    uint32_t i = 0;
+    do {
+        auto [data_ptr, data_size] = data_queue_.read(read_cursor_);
+        if (!data_ptr || data_size == 0) {
+            break;
+        }
         void* client = nullptr;
         memcpy(&client, data_ptr, sizeof(void*));
         char channel = data_ptr[sizeof(void*)];
@@ -245,6 +249,7 @@ inline void UserThreadWebsocketCallbacks::processData() {
             processMarketData(client, data_ptr, data_size - sizeof(void*) - 1);
         }
     }
+    while(++i < max_drain_count);
 }
 
 
@@ -417,8 +422,8 @@ inline void WebSocketClient::runDataLogger() {
         if (!data) {
             break;
         }
-        ++data;     // skip channel identifier
-        data_log_.write(data, size - 1);
+        data += sizeof(void*) + 1;     // skip client and channel identifier
+        data_log_.write(data, size - sizeof(void*) - 1);
         data_log_ << std::endl;
     }
 }
@@ -450,7 +455,7 @@ inline void WebSocketClient::onUserData(const char* data, std::size_t size) {
         // If data_queue_ is not nullptr, data logger is enabled.
         // dispatch data for logging
         if (data_queue_) {
-            dispatchData(data, size, 'M');
+            dispatchData(data, size, 'U');
         }
     }
 }
@@ -517,7 +522,7 @@ inline bool DataHandler::processMarketData(void *ws_client, const char* data, st
             processHeartbeat(j);
         }
         else {
-            LOG_ERROR("unknown channel: {}", to_string(channel));
+            LOG_ERROR("unknown channel: {}", channel.is_string() ? channel.get<std::string_view>() : channel.dump());
         }
     }
     catch (const std::exception &e) {
@@ -545,7 +550,7 @@ inline bool DataHandler::processUserData(void *ws_client, const char* data, std:
         else if (channel == "futures_balance_summary") {
         }
         else {
-            LOG_ERROR("unknown channel: {}", to_string(channel));
+            LOG_ERROR("unknown channel: {}", channel.is_string() ? channel.get<std::string_view>() : channel.dump());
         }
     }
     catch (const std::exception &e) {
@@ -652,6 +657,10 @@ inline bool DataHandler::processHeartbeat([[maybe_unused]] const json& j) {
 }
 
 inline bool DataHandler::checkMarketDataSequenceNumber([[maybe_unused]] void* client, int64_t seq_num) {
+    if (last_md_seq_num_ < 0) [[unlikely]] {
+        last_md_seq_num_ = seq_num;
+        return true;
+    }
     if (seq_num != last_md_seq_num_ + 1) {
         LOG_ERROR("market data message lost. seq_num: {}, last_md_seq_num: {}", seq_num, last_md_seq_num_);
         callbacks_->onMarketDataGap();
@@ -662,6 +671,10 @@ inline bool DataHandler::checkMarketDataSequenceNumber([[maybe_unused]] void* cl
 }
 
 inline bool DataHandler::checkUserDataSequenceNumber([[maybe_unused]] void* client, int64_t seq_num) {
+    if (last_user_seq_num_ < 0) [[unlikely]] {
+        last_user_seq_num_ = seq_num;
+        return true;
+    }
     if (seq_num != last_user_seq_num_ + 1) {
         LOG_ERROR("user data message lost. seq_num: {}, last_user_seq_num: {}", seq_num, last_user_seq_num_);
         callbacks_->onUserDataGap();