diff --git a/linux/include/ccapi_cpp/service/ccapi_market_data_service_deribit.h b/linux/include/ccapi_cpp/service/ccapi_market_data_service_deribit.h index 1a777e94f..1f52592b8 100644 --- a/linux/include/ccapi_cpp/service/ccapi_market_data_service_deribit.h +++ b/linux/include/ccapi_cpp/service/ccapi_market_data_service_deribit.h @@ -44,6 +44,25 @@ class MarketDataServiceDeribit : public MarketDataService { document.SetObject(); rj::Document::AllocatorType& allocator = document.GetAllocator(); auto now = UtilTime::now(); + + ////////////////////////////////////////////////// + // BEGIN - BTS Custom Code + ////////////////////////////////////////////////// + const auto rpcId = std::chrono::duration_cast(now.time_since_epoch()).count(); + const auto hbInterval = 10; + CCAPI_LOGGER_INFO( + "MarketDataServiceDeribit::onOpen. Setting heartbeat interval." + "interval = " + toString(hbInterval) + ". " + "rpcId = " + toString(rpcId)); + activeReqs.insert(rpcId); + this->appendParam(document, allocator, rpcId, "public/set_heartbeat", + { + {"interval", toString(hbInterval)}, + }); + ////////////////////////////////////////////////// + // END - BTS Custom Code + ////////////////////////////////////////////////// + this->appendParam(document, allocator, std::chrono::duration_cast(now.time_since_epoch()).count(), "public/set_heartbeat", { {"interval", "10"}, @@ -141,6 +160,136 @@ class MarketDataServiceDeribit : public MarketDataService { sendStringList.push_back(sendString); return sendStringList; } + + ////////////////////////////////////////////////// + // BEGIN - BTS Custom Code + ////////////////////////////////////////////////// + + std::unordered_set activeHeartbeatReqs; + std::unordered_set activeReqs; + + template + static std::pair readJson(rapidjson::Document &doc, const std::string &key); + + template <> + std::pair readJson(rapidjson::Document &doc, const std::string &key) { + auto member = doc.FindMember(key.c_str()); + if (member == doc.MemberEnd() || !member->value.IsString()) { + return {{}, false}; + } + return {member->value.GetString(), true}; + } + + template <> + std::pair readJson(rapidjson::Document &doc, const std::string &key) { + auto member = doc.FindMember(key.c_str()); + if (member == doc.MemberEnd()) { + return {0, false}; + } + const auto &v = member->value; + if (v.IsInt64()) { + return {v.GetInt64(), true}; + } + if (v.IsInt()) { + return {v.GetInt(), true}; + } + if (v.IsUint64()) { + return {static_cast(v.GetUint64()), true}; + } + if (v.IsUint()) { + return {static_cast(v.GetUint()), true}; + } + if (v.IsString()) { + const char *cstr = v.GetString(); + char* endPtr; + const std::int64_t result = std::strtol(cstr, &endPtr, 10); + if (*endPtr == '\0') { + return {result, true}; + } + } + return {0, false}; + } + + static bool isHeartbeatRequest(const std::string &textMessage) { + rj::Document doc; + doc.Parse(textMessage.c_str()); + auto resultIt = doc.FindMember("result"); + if (resultIt != doc.MemberEnd()) { + return false; + } + auto paramsIt = doc.FindMember("params"); + if (paramsIt == doc.MemberEnd() || !paramsIt->value.IsObject()) { + return false; + } + const auto params = paramsIt->value.GetObject(); + auto typeIt = params.FindMember("type"); + if (typeIt == params.MemberEnd() || !typeIt->value.IsString()) { + return false; + } + const std::string type{typeIt->value.GetString()}; + if (type != "test_request") { + return false; + } + return true; + } + + void sendHeartbeat(wspp::connection_hdl hdl) { + CCAPI_LOGGER_TRACE("MarketDataServiceDeribit::sendHeartbeat"); + rj::Document document; + document.SetObject(); + rj::Document::AllocatorType &allocator = document.GetAllocator(); + auto now = UtilTime::now(); + const auto rpcId = std::chrono::duration_cast(now.time_since_epoch()).count(); + appendParam(document, allocator, rpcId, "/public/test", {}); + rj::StringBuffer stringBuffer; + rj::Writer writer(stringBuffer); + document.Accept(writer); + std::string msg = stringBuffer.GetString(); + ErrorCode ec; + static const std::size_t MAX_ACTIVE_HEARTBEATS = 32; + if (activeHeartbeatReqs.size() > MAX_ACTIVE_HEARTBEATS) { + CCAPI_LOGGER_WARN("MarketDataServiceDeribit::sendHeartbeat. Too many active heartbeat reqs (" + + std::to_string(activeHeartbeatReqs.size()) + "): " + + ccapi::toString(activeHeartbeatReqs)); + activeHeartbeatReqs.clear(); + } + activeHeartbeatReqs.insert(rpcId); + CCAPI_LOGGER_INFO("MarketDataServiceDeribit::sendHeartbeat. Sending heartbeat rpcId: " + std::to_string(rpcId)); + this->send(hdl, msg, wspp::frame::opcode::text, ec); + if (ec) { + this->onError(Event::Type::REQUEST_STATUS, Message::Type::REQUEST_FAILURE, ec, "request"); + } + } + + void onTextMessage(wspp::connection_hdl hdl, const std::string& textMessage, const TimePoint& timeReceived) override { + if (isHeartbeatRequest(textMessage)) { + sendHeartbeat(hdl); + return; + } + rj::Document doc; + doc.Parse(textMessage.c_str()); + const auto [rpcId, hasRpcId] = readJson(doc, "id"); + if (hasRpcId) { + if (activeHeartbeatReqs.erase(rpcId) != 0) { + return; + } + const auto [resStr, hasResStr] = readJson(doc, "result"); + if (hasResStr && resStr == "ok") { + if (activeReqs.erase(rpcId) == 0) { + CCAPI_LOGGER_ERROR("MarketDataServiceDeribit::onTextMessage. Received rpc response with unknown id:\n{}" + textMessage); + } else { + CCAPI_LOGGER_DEBUG("MarketDataServiceDeribit::onTextMessage. Received rpc response:\n{}" + textMessage); + } + return; + } + } + MarketDataService::onTextMessage(hdl, textMessage, timeReceived); + } + + ////////////////////////////////////////////////// + // END - BTS Custom Code + ////////////////////////////////////////////////// + void processTextMessage(WsConnection& wsConnection, wspp::connection_hdl hdl, const std::string& textMessage, const TimePoint& timeReceived, Event& event, std::vector& marketDataMessageList) override { rj::Document document;