fix: Tweak ccapi deribit md so it sends hbs for generic subs

This commit is contained in:
John Colvin committed 2023-06-04 15:11:28 +00:00
1 parent f21c6594c0
commit 591efbc06b
1 file changed
+149
@@ -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<std::chrono::nanoseconds>(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<std::chrono::nanoseconds>(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<std::int64_t> activeHeartbeatReqs;
std::unordered_set<std::int64_t> activeReqs;
template <class T>
static std::pair<T, bool> readJson(rapidjson::Document &doc, const std::string &key);
template <>
std::pair<std::string, bool> readJson<std::string>(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<std::int64_t, bool> readJson<std::int64_t>(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<std::int64_t>(v.GetUint64()), true};
}
if (v.IsUint()) {
return {static_cast<std::int64_t>(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<rj::kParseNumbersAsStringsFlag>(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<std::chrono::nanoseconds>(now.time_since_epoch()).count();
appendParam(document, allocator, rpcId, "/public/test", {});
rj::StringBuffer stringBuffer;
rj::Writer<rj::StringBuffer> 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<rj::kParseNumbersAsStringsFlag>(textMessage.c_str());
const auto [rpcId, hasRpcId] = readJson<std::int64_t>(doc, "id");
if (hasRpcId) {
if (activeHeartbeatReqs.erase(rpcId) != 0) {
return;
}
const auto [resStr, hasResStr] = readJson<std::string>(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<MarketDataMessage>& marketDataMessageList) override {
rj::Document document;