#ifndef INCLUDE_CCAPI_CPP_SERVICE_CCAPI_SERVICE_H_ #define INCLUDE_CCAPI_CPP_SERVICE_CCAPI_SERVICE_H_ #include "ccapi_cpp/ccapi_logger.h" #ifndef RAPIDJSON_HAS_CXX11_NOEXCEPT #define RAPIDJSON_HAS_CXX11_NOEXCEPT 0 #endif #ifndef RAPIDJSON_ASSERT #define RAPIDJSON_ASSERT(x) \ if (!(x)) { \ throw std::runtime_error("rapidjson internal assertion failure"); \ } #endif #ifndef RAPIDJSON_PARSE_ERROR_NORETURN #define RAPIDJSON_PARSE_ERROR_NORETURN(parseErrorCode, offset) throw std::runtime_error(#parseErrorCode) #endif #ifndef CCAPI_WEBSOCKET_WRITE_BUFFER_SIZE #define CCAPI_WEBSOCKET_WRITE_BUFFER_SIZE 1 << 20 #endif #include #include "boost/asio/strand.hpp" #include "boost/beast/core.hpp" #include "boost/beast/http.hpp" #include "boost/beast/ssl.hpp" #include "boost/beast/version.hpp" #include "ccapi_cpp/ccapi_decimal.h" #include "ccapi_cpp/ccapi_event.h" #include "ccapi_cpp/ccapi_macro.h" #include "ccapi_cpp/ccapi_market_data_message.h" #include "rapidjson/document.h" #include "rapidjson/stringbuffer.h" #include "rapidjson/writer.h" // clang-format off #ifdef CCAPI_LEGACY_USE_WEBSOCKETPP #include "websocketpp/config/boost_config.hpp" #include "websocketpp/client.hpp" #include "websocketpp/common/connection_hdl.hpp" #include "websocketpp/config/asio_client.hpp" #else #include "boost/beast/websocket.hpp" #endif // clang-format on #ifdef CCAPI_LEGACY_USE_WEBSOCKETPP #if defined(CCAPI_ENABLE_SERVICE_MARKET_DATA) && \ (defined(CCAPI_ENABLE_EXCHANGE_HUOBI) || defined(CCAPI_ENABLE_EXCHANGE_HUOBI_USDT_SWAP) || defined(CCAPI_ENABLE_EXCHANGE_HUOBI_COIN_SWAP)) || \ defined(CCAPI_ENABLE_SERVICE_EXECUTION_MANAGEMENT) && \ (defined(CCAPI_ENABLE_EXCHANGE_HUOBI_USDT_SWAP) || defined(CCAPI_ENABLE_EXCHANGE_HUOBI_COIN_SWAP) || defined(CCAPI_ENABLE_EXCHANGE_BITMART)) #include #include #include "ccapi_cpp/websocketpp_decompress_workaround.h" #endif #else #include "ccapi_cpp/ccapi_inflate_stream.h" #endif #include "ccapi_cpp/ccapi_fix_connection.h" #include "ccapi_cpp/ccapi_http_connection.h" #include "ccapi_cpp/ccapi_http_retry.h" #include "ccapi_cpp/ccapi_queue.h" #include "ccapi_cpp/ccapi_request.h" #include "ccapi_cpp/ccapi_session_configs.h" #include "ccapi_cpp/ccapi_session_options.h" #include "ccapi_cpp/ccapi_subscription.h" #include "ccapi_cpp/ccapi_url.h" #include "ccapi_cpp/ccapi_ws_connection.h" #include "ccapi_cpp/service/ccapi_service_context.h" namespace beast = boost::beast; namespace http = beast::http; namespace net = boost::asio; namespace ssl = net::ssl; using tcp = net::ip::tcp; #ifdef CCAPI_LEGACY_USE_WEBSOCKETPP namespace wspp = websocketpp; #endif namespace rj = rapidjson; namespace ccapi { /** * Defines a service which provides access to exchange API and normalizes them. This is a base class that implements generic functionalities for dealing with * exchange REST and Websocket APIs. The Session object is responsible for routing requests and subscriptions to the desired concrete service. */ class Service : public std::enable_shared_from_this { public: typedef ServiceContext* ServiceContextPtr; #ifdef CCAPI_LEGACY_USE_WEBSOCKETPP typedef wspp::lib::error_code ErrorCode; #else typedef boost::system::error_code ErrorCode; // a.k.a. beast::error_code #endif enum class PingPongMethod { WEBSOCKET_PROTOCOL_LEVEL, WEBSOCKET_APPLICATION_LEVEL, FIX_PROTOCOL_LEVEL, }; static std::string pingPongMethodToString(PingPongMethod pingPongMethod) { std::string output; switch (pingPongMethod) { case PingPongMethod::WEBSOCKET_PROTOCOL_LEVEL: output = "WEBSOCKET_PROTOCOL_LEVEL"; break; case PingPongMethod::WEBSOCKET_APPLICATION_LEVEL: output = "WEBSOCKET_APPLICATION_LEVEL"; break; case PingPongMethod::FIX_PROTOCOL_LEVEL: output = "FIX_PROTOCOL_LEVEL"; break; default: CCAPI_LOGGER_FATAL(CCAPI_UNSUPPORTED_VALUE); } return output; } Service(std::function*)> eventHandler, SessionOptions sessionOptions, SessionConfigs sessionConfigs, ServiceContextPtr serviceContextPtr) : eventHandler(eventHandler), sessionOptions(sessionOptions), sessionConfigs(sessionConfigs), serviceContextPtr(serviceContextPtr), resolver(*serviceContextPtr->ioContextPtr), resolverWs(*serviceContextPtr->ioContextPtr) { this->enableCheckPingPongWebsocketProtocolLevel = this->sessionOptions.enableCheckPingPongWebsocketProtocolLevel; this->enableCheckPingPongWebsocketApplicationLevel = this->sessionOptions.enableCheckPingPongWebsocketApplicationLevel; // this->pingIntervalMillisecondsByMethodMap[PingPongMethod::WEBSOCKET_PROTOCOL_LEVEL] = sessionOptions.pingWebsocketProtocolLevelIntervalMilliseconds; // this->pongTimeoutMillisecondsByMethodMap[PingPongMethod::WEBSOCKET_PROTOCOL_LEVEL] = sessionOptions.pongWebsocketProtocolLevelTimeoutMilliseconds; this->pingIntervalMillisecondsByMethodMap[PingPongMethod::WEBSOCKET_APPLICATION_LEVEL] = sessionOptions.pingWebsocketApplicationLevelIntervalMilliseconds; this->pongTimeoutMillisecondsByMethodMap[PingPongMethod::WEBSOCKET_APPLICATION_LEVEL] = sessionOptions.pongWebsocketApplicationLevelTimeoutMilliseconds; this->pingIntervalMillisecondsByMethodMap[PingPongMethod::FIX_PROTOCOL_LEVEL] = sessionOptions.heartbeatFixIntervalMilliseconds; this->pongTimeoutMillisecondsByMethodMap[PingPongMethod::FIX_PROTOCOL_LEVEL] = sessionOptions.heartbeatFixTimeoutMilliseconds; } virtual ~Service() { for (const auto& x : this->pingTimerByMethodByConnectionIdMap) { for (const auto& y : x.second) { y.second->cancel(); } } for (const auto& x : this->pongTimeOutTimerByMethodByConnectionIdMap) { for (const auto& y : x.second) { y.second->cancel(); } } for (const auto& x : this->connectRetryOnFailTimerByConnectionIdMap) { x.second->cancel(); } } void purgeHttpConnectionPool() { this->httpConnectionPool.clear(); } void purgeHttpConnectionPool(const std::string& localIpAddress) { this->httpConnectionPool.erase(localIpAddress); } void purgeHttpConnectionPool(const std::string& localIpAddress, const std::string& baseUrl) { this->httpConnectionPool[localIpAddress].erase(baseUrl); } void forceCloseWebsocketConnections() { for (const auto& x : this->wsConnectionByIdMap) { ErrorCode ec; auto wsConnectionPtr = x.second; this->close(wsConnectionPtr, beast::websocket::close_code::normal, beast::websocket::close_reason("force close"), ec); if (ec) { this->onError(Event::Type::SUBSCRIPTION_STATUS, Message::Type::GENERIC_ERROR, ec, "shutdown"); } } } void stop() { for (const auto& x : this->sendRequestDelayTimerByCorrelationIdMap) { x.second->cancel(); } sendRequestDelayTimerByCorrelationIdMap.clear(); this->shouldContinue = false; for (const auto& x : this->wsConnectionByIdMap) { ErrorCode ec; auto wsConnectionPtr = x.second; this->close(wsConnectionPtr, beast::websocket::close_code::normal, beast::websocket::close_reason("stop"), ec); if (ec) { this->onError(Event::Type::SUBSCRIPTION_STATUS, Message::Type::GENERIC_ERROR, ec, "shutdown"); } this->shouldProcessRemainingMessageOnClosingByConnectionIdMap[wsConnectionPtr->id] = false; } } virtual void convertRequestForRestCustom(http::request& req, const Request& request, const TimePoint& now, const std::string& symbolId, const std::map& credential) { auto errorMessage = "REST unimplemented operation " + Request::operationToString(request.getOperation()) + " for exchange " + request.getExchange(); throw std::runtime_error(errorMessage); } virtual void subscribe(std::vector& subscriptionList) {} virtual void convertRequestForRest(http::request& req, const Request& request, const TimePoint& now, const std::string& symbolId, const std::map& credential) {} virtual void processSuccessfulTextMessageRest(int statusCode, const Request& request, const std::string& textMessage, const TimePoint& timeReceived, Queue* eventQueuePtr) {} std::shared_ptr> sendRequest(Request& request, const bool useFuture, const TimePoint& now, long delayMilliseconds, Queue* eventQueuePtr) { CCAPI_LOGGER_FUNCTION_ENTER; CCAPI_LOGGER_DEBUG("request = " + toString(request)); CCAPI_LOGGER_DEBUG("useFuture = " + toString(useFuture)); TimePoint then; if (delayMilliseconds > 0) { then = now + std::chrono::milliseconds(delayMilliseconds); } else { then = now; } http::request req; try { req = this->convertRequest(request, then); } catch (const std::runtime_error& e) { CCAPI_LOGGER_ERROR(std::string("e.what() = ") + e.what()); this->onError(Event::Type::REQUEST_STATUS, Message::Type::REQUEST_FAILURE, e, {request.getCorrelationId()}, eventQueuePtr); std::promise* promisePtrRaw = nullptr; if (useFuture) { promisePtrRaw = new std::promise(); } std::shared_ptr> promisePtr(promisePtrRaw); std::shared_ptr> futurePtr(nullptr); if (useFuture) { futurePtr = std::make_shared>(std::move(promisePtr->get_future())); promisePtr->set_value(); } return futurePtr; } std::promise* promisePtrRaw = nullptr; if (useFuture) { promisePtrRaw = new std::promise(); } std::shared_ptr> promisePtr(promisePtrRaw); HttpRetry retry(0, 0, "", promisePtr); if (delayMilliseconds > 0) { TimerPtr timerPtr(new net::steady_timer(*this->serviceContextPtr->ioContextPtr, std::chrono::milliseconds(delayMilliseconds))); timerPtr->async_wait([that = shared_from_this(), request, req, retry, eventQueuePtr](ErrorCode const& ec) mutable { if (ec) { CCAPI_LOGGER_ERROR("request = " + toString(request) + ", sendRequest timer error: " + ec.message()); that->onError(Event::Type::REQUEST_STATUS, Message::Type::GENERIC_ERROR, ec, "timer", {request.getCorrelationId()}, eventQueuePtr); } else { auto now = UtilTime::now(); request.setTimeSent(now); that->tryRequest(request, req, retry, eventQueuePtr); } that->sendRequestDelayTimerByCorrelationIdMap.erase(request.getCorrelationId()); }); this->sendRequestDelayTimerByCorrelationIdMap[request.getCorrelationId()] = timerPtr; } else { request.setTimeSent(now); net::post(*this->serviceContextPtr->ioContextPtr, [that = shared_from_this(), request, req, retry, eventQueuePtr]() mutable { that->tryRequest(request, req, retry, eventQueuePtr); }); } std::shared_ptr> futurePtr(nullptr); if (useFuture) { futurePtr = std::make_shared>(std::move(promisePtr->get_future())); } CCAPI_LOGGER_FUNCTION_EXIT; return futurePtr; } virtual void sendRequestByWebsocket(Request& request, const TimePoint& now) {} virtual void sendRequestByFix(Request& request, const TimePoint& now) {} virtual void subscribeByFix(Subscription& subscription) {} void onError(const Event::Type eventType, const Message::Type messageType, const std::string& errorMessage, const std::vector correlationIdList = {}, Queue* eventQueuePtr = nullptr) { CCAPI_LOGGER_ERROR("errorMessage = " + errorMessage); CCAPI_LOGGER_ERROR("correlationIdList = " + toString(correlationIdList)); Event event; event.setType(eventType); Message message; auto now = UtilTime::now(); message.setTimeReceived(now); message.setType(messageType); message.setCorrelationIdList(correlationIdList); Element element; element.insert(CCAPI_ERROR_MESSAGE, errorMessage); message.setElementList({element}); event.setMessageList({message}); this->eventHandler(event, eventQueuePtr); } void onError(const Event::Type eventType, const Message::Type messageType, const ErrorCode& ec, const std::string& what, const std::vector correlationIdList = {}, Queue* eventQueuePtr = nullptr) { this->onError(eventType, messageType, what + ": " + ec.message() + ", category: " + ec.category().name(), correlationIdList, eventQueuePtr); } void onError(const Event::Type eventType, const Message::Type messageType, const std::exception& e, const std::vector correlationIdList = {}, Queue* eventQueuePtr = nullptr) { this->onError(eventType, messageType, e.what(), correlationIdList, eventQueuePtr); } void onResponseError(const Request& request, int statusCode, const std::string& errorMessage, Queue* eventQueuePtr) { std::string statusCodeStr = std::to_string(statusCode); CCAPI_LOGGER_ERROR("request = " + toString(request) + ", statusCode = " + statusCodeStr + ", errorMessage = " + errorMessage); Event event; event.setType(Event::Type::RESPONSE); Message message; auto now = UtilTime::now(); message.setTimeReceived(now); message.setType(Message::Type::RESPONSE_ERROR); message.setCorrelationIdList({request.getCorrelationId()}); Element element; element.insert(CCAPI_HTTP_STATUS_CODE, statusCodeStr); element.insert(CCAPI_ERROR_MESSAGE, UtilString::trim(errorMessage)); message.setElementList({element}); event.setMessageList({message}); this->eventHandler(event, eventQueuePtr); } #ifndef CCAPI_EXPOSE_INTERNAL protected: #endif // static std::string printableString(const char* s, size_t n) { // std::string output(s, n); // std::replace(output.begin(), output.end(), '\x01', '^'); // return output; // } // static std::string printableString(const std::string& s) { // std::string output(s); // std::replace(output.begin(), output.end(), '\x01', '^'); // return output; // } typedef ServiceContext::SslContextPtr SslContextPtr; #ifdef CCAPI_LEGACY_USE_WEBSOCKETPP typedef ServiceContext::TlsClient TlsClient; #endif typedef std::shared_ptr TimerPtr; void setHostRestFromUrlRest(std::string baseUrlRest) { auto hostPort = this->extractHostFromUrl(baseUrlRest); this->hostRest = hostPort.first; this->portRest = hostPort.second; } void setHostWsFromUrlWs(std::string baseUrlWs) { auto hostPort = this->extractHostFromUrl(baseUrlWs); this->hostWs = hostPort.first; this->portWs = hostPort.second; } std::pair extractHostFromUrl(std::string baseUrl) { std::string host; std::string port; if (!baseUrl.empty()) { auto splitted1 = UtilString::split(baseUrl, "://"); auto splitted2 = UtilString::split(UtilString::split(splitted1.at(1), "/").at(0), ":"); host = splitted2.at(0); if (splitted2.size() == 2) { port = splitted2.at(1); } else { if (splitted1.at(0) == "https" || splitted1.at(0) == "wss") { port = CCAPI_HTTPS_PORT_DEFAULT; } else { port = CCAPI_HTTP_PORT_DEFAULT; } } } return std::make_pair(host, port); } template std::shared_ptr shared_from_base() { return std::static_pointer_cast(shared_from_this()); } void sendRequest(const http::request& req, std::function errorHandler, std::function&)> responseHandler, long timeoutMilliseconds) { #if defined(CCAPI_ENABLE_LOG_DEBUG) || defined(CCAPI_ENABLE_LOG_TRACE) std::ostringstream oss; oss << req; CCAPI_LOGGER_DEBUG("req = \n" + oss.str()); #endif std::shared_ptr> streamPtr(nullptr); try { streamPtr = this->createStream>(this->serviceContextPtr->ioContextPtr, this->serviceContextPtr->sslContextPtr, this->hostRest); } catch (const beast::error_code& ec) { CCAPI_LOGGER_TRACE("fail"); errorHandler(ec); return; } std::shared_ptr httpConnectionPtr(new HttpConnection(this->hostRest, this->portRest, streamPtr)); CCAPI_LOGGER_DEBUG("httpConnection = " + toString(*httpConnectionPtr)); std::shared_ptr newResolverPtr(new tcp::resolver(*this->serviceContextPtr->ioContextPtr)); CCAPI_LOGGER_TRACE("this->hostRest = " + this->hostRest); CCAPI_LOGGER_TRACE("this->portRest = " + this->portRest); newResolverPtr->async_resolve(this->hostRest, this->portRest, beast::bind_front_handler(&Service::onResolve, shared_from_this(), httpConnectionPtr, newResolverPtr, req, errorHandler, responseHandler, timeoutMilliseconds)); // this->startConnect(httpConnectionPtr, req, errorHandler, responseHandler, timeoutMilliseconds, this->tcpResolverResultsRest); } void sendRequest(const std::string& host, const std::string& port, const http::request& req, std::function errorHandler, std::function&)> responseHandler, long timeoutMilliseconds) { #if defined(CCAPI_ENABLE_LOG_DEBUG) || defined(CCAPI_ENABLE_LOG_TRACE) std::ostringstream oss; oss << req; CCAPI_LOGGER_DEBUG("req = \n" + oss.str()); #endif std::shared_ptr> streamPtr(nullptr); try { streamPtr = this->createStream>(this->serviceContextPtr->ioContextPtr, this->serviceContextPtr->sslContextPtr, host); } catch (const beast::error_code& ec) { CCAPI_LOGGER_TRACE("fail"); errorHandler(ec); return; } std::shared_ptr httpConnectionPtr(new HttpConnection(host, port, streamPtr)); CCAPI_LOGGER_DEBUG("httpConnection = " + toString(*httpConnectionPtr)); std::shared_ptr newResolverPtr(new tcp::resolver(*this->serviceContextPtr->ioContextPtr)); CCAPI_LOGGER_TRACE("host = " + host); CCAPI_LOGGER_TRACE("port = " + port); newResolverPtr->async_resolve(host, port, beast::bind_front_handler(&Service::onResolve, shared_from_this(), httpConnectionPtr, newResolverPtr, req, errorHandler, responseHandler, timeoutMilliseconds)); } void onResolve(std::shared_ptr httpConnectionPtr, std::shared_ptr newResolverPtr, http::request req, std::function errorHandler, std::function&)> responseHandler, long timeoutMilliseconds, beast::error_code ec, tcp::resolver::results_type tcpNewResolverResults) { if (ec) { CCAPI_LOGGER_TRACE("fail"); errorHandler(ec); return; } this->startConnect(httpConnectionPtr, req, errorHandler, responseHandler, timeoutMilliseconds, tcpNewResolverResults); } void startConnect(std::shared_ptr httpConnectionPtr, http::request req, std::function errorHandler, std::function&)> responseHandler, long timeoutMilliseconds, tcp::resolver::results_type tcpNewResolverResults) { beast::ssl_stream& stream = *httpConnectionPtr->streamPtr; if (timeoutMilliseconds > 0) { beast::get_lowest_layer(stream).expires_after(std::chrono::milliseconds(timeoutMilliseconds)); } CCAPI_LOGGER_TRACE("before async_connect"); beast::get_lowest_layer(stream).async_connect( tcpNewResolverResults, beast::bind_front_handler(&Service::onConnect, shared_from_this(), httpConnectionPtr, req, errorHandler, responseHandler)); CCAPI_LOGGER_TRACE("after async_connect"); } void onConnect(std::shared_ptr httpConnectionPtr, http::request req, std::function errorHandler, std::function&)> responseHandler, beast::error_code ec, tcp::resolver::results_type::endpoint_type) { CCAPI_LOGGER_TRACE("async_connect callback start"); if (ec) { CCAPI_LOGGER_TRACE("fail"); errorHandler(ec); return; } CCAPI_LOGGER_TRACE("connected"); beast::ssl_stream& stream = *httpConnectionPtr->streamPtr; // #ifdef CCAPI_DISABLE_NAGLE_ALGORITHM beast::get_lowest_layer(stream).socket().set_option(tcp::no_delay(true)); // #endif CCAPI_LOGGER_TRACE("before ssl async_handshake"); stream.async_handshake(ssl::stream_base::client, beast::bind_front_handler(&Service::onSslHandshake, shared_from_this(), httpConnectionPtr, req, errorHandler, responseHandler)); CCAPI_LOGGER_TRACE("after ssl async_handshake"); } void onSslHandshake(std::shared_ptr httpConnectionPtr, http::request req, std::function errorHandler, std::function&)> responseHandler, beast::error_code ec) { CCAPI_LOGGER_TRACE("ssl async_handshake callback start"); if (ec) { CCAPI_LOGGER_TRACE("fail"); errorHandler(ec); return; } CCAPI_LOGGER_TRACE("ssl handshaked"); beast::ssl_stream& stream = *httpConnectionPtr->streamPtr; std::shared_ptr> reqPtr(new http::request(std::move(req))); CCAPI_LOGGER_TRACE("before async_write"); http::async_write(stream, *reqPtr, beast::bind_front_handler(&Service::onWrite, shared_from_this(), httpConnectionPtr, reqPtr, errorHandler, responseHandler)); CCAPI_LOGGER_TRACE("after async_write"); } void onWrite(std::shared_ptr httpConnectionPtr, std::shared_ptr> reqPtr, std::function errorHandler, std::function&)> responseHandler, beast::error_code ec, std::size_t bytes_transferred) { CCAPI_LOGGER_TRACE("async_write callback start"); boost::ignore_unused(bytes_transferred); if (ec) { CCAPI_LOGGER_TRACE("fail"); errorHandler(ec); return; } CCAPI_LOGGER_TRACE("written"); std::shared_ptr bufferPtr(new beast::flat_buffer()); std::shared_ptr> resPtr(new http::response()); beast::ssl_stream& stream = *httpConnectionPtr->streamPtr; CCAPI_LOGGER_TRACE("before async_read"); http::async_read( stream, *bufferPtr, *resPtr, beast::bind_front_handler(&Service::onRead, shared_from_this(), httpConnectionPtr, reqPtr, errorHandler, responseHandler, bufferPtr, resPtr)); CCAPI_LOGGER_TRACE("after async_read"); } void onRead(std::shared_ptr httpConnectionPtr, std::shared_ptr> reqPtr, std::function errorHandler, std::function&)> responseHandler, std::shared_ptr bufferPtr, std::shared_ptr> resPtr, beast::error_code ec, std::size_t bytes_transferred) { CCAPI_LOGGER_TRACE("async_read callback start"); boost::ignore_unused(bytes_transferred); if (ec) { CCAPI_LOGGER_TRACE("fail"); errorHandler(ec); return; } beast::ssl_stream& stream = *httpConnectionPtr->streamPtr; #if defined(CCAPI_ENABLE_LOG_DEBUG) || defined(CCAPI_ENABLE_LOG_TRACE) { std::ostringstream oss; oss << *reqPtr; CCAPI_LOGGER_DEBUG("req = \n" + oss.str()); } { std::ostringstream oss; oss << *resPtr; CCAPI_LOGGER_DEBUG("res = \n" + oss.str()); } #endif responseHandler(*resPtr); } template std::shared_ptr createStream(net::io_context* iocPtr, net::ssl::context* ctxPtr, const std::string& host) { std::shared_ptr streamPtr(new T(*iocPtr, *ctxPtr)); // Set SNI Hostname (many hosts need this to handshake successfully) if (!SSL_set_tlsext_host_name(streamPtr->native_handle(), host.c_str())) { beast::error_code ec{static_cast(::ERR_get_error()), net::error::get_ssl_category()}; CCAPI_LOGGER_DEBUG("error SSL_set_tlsext_host_name: " + ec.message()); throw ec; } return streamPtr; } #ifdef CCAPI_LEGACY_USE_WEBSOCKETPP #else std::shared_ptr>> createWsStream(net::io_context* iocPtr, net::ssl::context* ctxPtr) { std::shared_ptr>> streamPtr( new beast::websocket::stream>(*iocPtr, *ctxPtr)); return streamPtr; } #endif // std::shared_ptr> createStream(net::io_context* iocPtr, net::ssl::context* ctxPtr, // const std::string& host) { // std::shared_ptr> streamPtr(new beast::ssl_stream(*iocPtr, *ctxPtr)); // // Set SNI Hostname (many hosts need this to handshake successfully) // if (!SSL_set_tlsext_host_name(streamPtr->native_handle(), host.c_str())) { // beast::error_code ec{static_cast(::ERR_get_error()), net::error::get_ssl_category()}; // CCAPI_LOGGER_DEBUG("error SSL_set_tlsext_host_name: " + ec.message()); // throw ec; // } // return streamPtr; // } void performRequestWithNewHttpConnection(std::shared_ptr httpConnectionPtr, const Request& request, http::request& req, const HttpRetry& retry, Queue* eventQueuePtr) { CCAPI_LOGGER_FUNCTION_ENTER; CCAPI_LOGGER_DEBUG("httpConnection = " + toString(*httpConnectionPtr)); CCAPI_LOGGER_DEBUG("request = " + toString(request)); CCAPI_LOGGER_DEBUG("retry = " + toString(retry)); beast::ssl_stream& stream = *httpConnectionPtr->streamPtr; CCAPI_LOGGER_DEBUG("this->sessionOptions.httpRequestTimeoutMilliseconds = " + toString(this->sessionOptions.httpRequestTimeoutMilliseconds)); if (this->sessionOptions.httpRequestTimeoutMilliseconds > 0) { beast::get_lowest_layer(stream).expires_after(std::chrono::milliseconds(this->sessionOptions.httpRequestTimeoutMilliseconds)); } const auto& localIpAddress = request.getLocalIpAddress(); CCAPI_LOGGER_TRACE("localIpAddress = " + localIpAddress); if (!localIpAddress.empty()) { if (!beast::get_lowest_layer(stream).socket().is_open()) { ErrorCode ec; CCAPI_LOGGER_TRACE("before socket open"); beast::get_lowest_layer(stream).socket().open(net::ip::tcp::v4(), ec); if (ec) { CCAPI_LOGGER_TRACE("fail"); this->onError(Event::Type::REQUEST_STATUS, Message::Type::REQUEST_FAILURE, ec, "socket open", {request.getCorrelationId()}, eventQueuePtr); return; } } ErrorCode ec; tcp::endpoint existingLocalEndpoint = beast::get_lowest_layer(stream).socket().local_endpoint(ec); if (ec) { CCAPI_LOGGER_TRACE("fail"); this->onError(Event::Type::REQUEST_STATUS, Message::Type::REQUEST_FAILURE, ec, "socket get local endpoint", {request.getCorrelationId()}, eventQueuePtr); return; } tcp::endpoint localEndpoint(net::ip::address::from_string(localIpAddress), 0); // Note: Setting the port to 0 means the OS will select a free port for you if (localEndpoint != existingLocalEndpoint) { ErrorCode ec; CCAPI_LOGGER_TRACE("before socket bind"); beast::get_lowest_layer(stream).socket().bind(localEndpoint, ec); if (ec) { CCAPI_LOGGER_TRACE("fail"); this->onError(Event::Type::REQUEST_STATUS, Message::Type::REQUEST_FAILURE, ec, "socket bind", {request.getCorrelationId()}, eventQueuePtr); return; } } } std::shared_ptr newResolverPtr(new tcp::resolver(*this->serviceContextPtr->ioContextPtr)); CCAPI_LOGGER_TRACE("httpConnectionPtr->host = " + httpConnectionPtr->host); CCAPI_LOGGER_TRACE("httpConnectionPtr->port = " + httpConnectionPtr->port); newResolverPtr->async_resolve( httpConnectionPtr->host, httpConnectionPtr->port, beast::bind_front_handler(&Service::onResolveWorkaround, shared_from_this(), httpConnectionPtr, newResolverPtr, request, req, retry, eventQueuePtr)); CCAPI_LOGGER_FUNCTION_EXIT; } void onResolveWorkaround(std::shared_ptr httpConnectionPtr, std::shared_ptr newResolverPtr, Request request, http::request req, HttpRetry retry, Queue* eventQueuePtr, beast::error_code ec, tcp::resolver::results_type tcpNewResolverResults) { if (ec) { CCAPI_LOGGER_TRACE("fail"); this->onError(Event::Type::REQUEST_STATUS, Message::Type::REQUEST_FAILURE, ec, "DNS resolve", {request.getCorrelationId()}, eventQueuePtr); return; } CCAPI_LOGGER_TRACE("before asyncConnectWorkaround"); TimerPtr timerPtr{nullptr}; if (this->sessionOptions.httpRequestTimeoutMilliseconds > 0) { timerPtr = std::make_shared(*this->serviceContextPtr->ioContextPtr, std::chrono::milliseconds(this->sessionOptions.httpRequestTimeoutMilliseconds)); timerPtr->async_wait([httpConnectionPtr](ErrorCode const& ec) { if (ec) { if (ec != net::error::make_error_code(net::error::basic_errors::operation_aborted)) { CCAPI_LOGGER_ERROR("httpConnectionPtr = " + toString(*httpConnectionPtr) + ", connect timeout timer error: " + ec.message()); } } else { CCAPI_LOGGER_TRACE("httpConnectionPtr = " + toString(*httpConnectionPtr) + ", connect timeout timer triggered"); beast::ssl_stream& stream = *httpConnectionPtr->streamPtr; beast::get_lowest_layer(stream).socket().cancel(); } }); } this->asyncConnectWorkaround(httpConnectionPtr, timerPtr, request, req, retry, eventQueuePtr, tcpNewResolverResults, 0); CCAPI_LOGGER_TRACE("after asyncConnectWorkaround"); } // used to avoid asio close and reopen the socket and therefore losing the bound local ip address void asyncConnectWorkaround(std::shared_ptr httpConnectionPtr, TimerPtr timerPtr, Request request, http::request req, HttpRetry retry, Queue* eventQueuePtr, tcp::resolver::results_type tcpNewResolverResults, size_t tcpNewResolverResultsIndex) { auto it = tcpNewResolverResults.begin(); std::advance(it, tcpNewResolverResultsIndex); if (it == tcpNewResolverResults.end()) { ErrorCode ec = net::error::make_error_code(net::error::misc_errors::not_found); this->onError(Event::Type::REQUEST_STATUS, Message::Type::REQUEST_FAILURE, ec, "connect", {request.getCorrelationId()}, eventQueuePtr); return; } CCAPI_LOGGER_TRACE("before async_connect"); beast::ssl_stream& stream = *httpConnectionPtr->streamPtr; beast::get_lowest_layer(stream).socket().async_connect( *it, beast::bind_front_handler(&Service::onConnect_2, shared_from_this(), httpConnectionPtr, timerPtr, request, req, retry, eventQueuePtr, tcpNewResolverResults, tcpNewResolverResultsIndex)); CCAPI_LOGGER_TRACE("after async_connect"); } void onConnect_2(std::shared_ptr httpConnectionPtr, TimerPtr timerPtr, Request request, http::request req, HttpRetry retry, Queue* eventQueuePtr, tcp::resolver::results_type tcpNewResolverResults, size_t tcpNewResolverResultsIndex, beast::error_code ec) { CCAPI_LOGGER_TRACE("async_connect callback start"); CCAPI_LOGGER_TRACE("local endpoint has address " + beast::get_lowest_layer(*httpConnectionPtr->streamPtr).socket().local_endpoint().address().to_string()); if (ec) { CCAPI_LOGGER_TRACE("fail"); if (ec == net::error::make_error_code(net::error::basic_errors::operation_aborted)) { CCAPI_LOGGER_TRACE("fail"); this->onError(Event::Type::REQUEST_STATUS, Message::Type::REQUEST_FAILURE, ec, "connect attempt timeout", {request.getCorrelationId()}, eventQueuePtr); return; } this->asyncConnectWorkaround(httpConnectionPtr, timerPtr, request, req, retry, eventQueuePtr, tcpNewResolverResults, tcpNewResolverResultsIndex + 1); return; } timerPtr->cancel(); CCAPI_LOGGER_TRACE("connected"); beast::ssl_stream& stream = *httpConnectionPtr->streamPtr; // #ifdef CCAPI_DISABLE_NAGLE_ALGORITHM beast::get_lowest_layer(stream).socket().set_option(tcp::no_delay(true)); // #endif CCAPI_LOGGER_TRACE("before ssl async_handshake"); stream.async_handshake(ssl::stream_base::client, beast::bind_front_handler(&Service::onSslHandshake_2, shared_from_this(), httpConnectionPtr, request, req, retry, eventQueuePtr)); CCAPI_LOGGER_TRACE("after ssl async_handshake"); } void onSslHandshake_2(std::shared_ptr httpConnectionPtr, Request request, http::request req, HttpRetry retry, Queue* eventQueuePtr, beast::error_code ec) { CCAPI_LOGGER_TRACE("ssl async_handshake callback start"); if (ec) { CCAPI_LOGGER_TRACE("fail"); this->onError(Event::Type::REQUEST_STATUS, Message::Type::REQUEST_FAILURE, ec, "ssl handshake", {request.getCorrelationId()}, eventQueuePtr); return; } CCAPI_LOGGER_TRACE("ssl handshaked"); this->startWrite_2(httpConnectionPtr, request, req, retry, eventQueuePtr); } void startWrite_2(std::shared_ptr httpConnectionPtr, Request request, http::request req, HttpRetry retry, Queue* eventQueuePtr) { beast::ssl_stream& stream = *httpConnectionPtr->streamPtr; if (this->sessionOptions.httpRequestTimeoutMilliseconds > 0) { beast::get_lowest_layer(stream).expires_after(std::chrono::milliseconds(this->sessionOptions.httpRequestTimeoutMilliseconds)); } std::shared_ptr> reqPtr(new http::request(std::move(req))); CCAPI_LOGGER_TRACE("before async_write"); http::async_write(stream, *reqPtr, beast::bind_front_handler(&Service::onWrite_2, shared_from_this(), httpConnectionPtr, request, reqPtr, retry, eventQueuePtr)); CCAPI_LOGGER_TRACE("after async_write"); } void onWrite_2(std::shared_ptr httpConnectionPtr, Request request, std::shared_ptr> reqPtr, HttpRetry retry, Queue* eventQueuePtr, beast::error_code ec, std::size_t bytes_transferred) { CCAPI_LOGGER_TRACE("async_write callback start"); boost::ignore_unused(bytes_transferred); if (ec) { CCAPI_LOGGER_TRACE("fail"); this->onError(Event::Type::REQUEST_STATUS, Message::Type::REQUEST_FAILURE, ec, "write", {request.getCorrelationId()}, eventQueuePtr); this->httpConnectionPool[request.getLocalIpAddress()][request.getBaseUrl()].clear(); auto now = UtilTime::now(); auto req = this->convertRequest(request, now); retry.numRetry += 1; this->tryRequest(request, req, retry, eventQueuePtr); return; } CCAPI_LOGGER_TRACE("written"); std::shared_ptr bufferPtr(new beast::flat_buffer()); std::shared_ptr> resPtr(new http::response()); beast::ssl_stream& stream = *httpConnectionPtr->streamPtr; CCAPI_LOGGER_TRACE("before async_read"); http::async_read( stream, *bufferPtr, *resPtr, beast::bind_front_handler(&Service::onRead_2, shared_from_this(), httpConnectionPtr, request, reqPtr, retry, bufferPtr, resPtr, eventQueuePtr)); CCAPI_LOGGER_TRACE("after async_read"); } void onRead_2(std::shared_ptr httpConnectionPtr, Request request, std::shared_ptr> reqPtr, HttpRetry retry, std::shared_ptr bufferPtr, std::shared_ptr> resPtr, Queue* eventQueuePtr, beast::error_code ec, std::size_t bytes_transferred) { CCAPI_LOGGER_TRACE("async_read callback start"); CCAPI_LOGGER_TRACE("local endpoint has address " + beast::get_lowest_layer(*httpConnectionPtr->streamPtr).socket().local_endpoint().address().to_string()); auto now = UtilTime::now(); boost::ignore_unused(bytes_transferred); if (ec) { CCAPI_LOGGER_TRACE("fail"); this->onError(Event::Type::REQUEST_STATUS, Message::Type::REQUEST_FAILURE, ec, "read", {request.getCorrelationId()}, eventQueuePtr); this->httpConnectionPool[request.getLocalIpAddress()][request.getBaseUrl()].clear(); auto now = UtilTime::now(); auto req = this->convertRequest(request, now); retry.numRetry += 1; this->tryRequest(request, req, retry, eventQueuePtr); return; } if (!this->sessionOptions.enableOneHttpConnectionPerRequest) { httpConnectionPtr->lastReceiveDataTp = now; const auto& localIpAddress = request.getLocalIpAddress(); const auto& requestBaseUrl = request.getBaseUrl(); if (this->sessionOptions.httpConnectionPoolMaxSize > 0 && this->httpConnectionPool[localIpAddress][requestBaseUrl].size() >= this->sessionOptions.httpConnectionPoolMaxSize) { CCAPI_LOGGER_TRACE("httpConnectionPool is full for localIpAddress = " + localIpAddress + ", requestBaseUrl = " + toString(requestBaseUrl)); this->httpConnectionPool[localIpAddress][requestBaseUrl].pop_front(); } this->httpConnectionPool[localIpAddress][requestBaseUrl].push_back(httpConnectionPtr); CCAPI_LOGGER_TRACE("pushed back httpConnectionPtr " + toString(*httpConnectionPtr) + " to httpConnectionPool for localIpAddress = " + localIpAddress + ", requestBaseUrl = " + toString(requestBaseUrl)); } #if defined(CCAPI_ENABLE_LOG_DEBUG) || defined(CCAPI_ENABLE_LOG_TRACE) { std::ostringstream oss; oss << *reqPtr; CCAPI_LOGGER_DEBUG("req = \n" + oss.str()); } { std::ostringstream oss; oss << *resPtr; CCAPI_LOGGER_DEBUG("res = \n" + oss.str()); } #endif int statusCode = resPtr->result_int(); std::string body = resPtr->body(); try { if (statusCode / 100 == 2) { this->processSuccessfulTextMessageRest(statusCode, request, body, now, eventQueuePtr); } else if (statusCode / 100 == 3) { if (resPtr->base().find("Location") != resPtr->base().end()) { Url url(resPtr->base() .at("Location") #if BOOST_VERSION < 108100 // Boost Beast 1.81 uses boost::core::string_view which doesn't contain to_string() method .to_string() #endif ); std::string host(url.host); if (!url.port.empty()) { host += ":"; host += url.port; } auto now = UtilTime::now(); auto req = this->convertRequest(request, now); req.set(http::field::host, host); req.target(url.target); retry.numRedirect += 1; CCAPI_LOGGER_WARN("redirect from request " + request.toString() + " to url " + url.toString()); this->tryRequest(request, req, retry, eventQueuePtr); } this->onResponseError(request, statusCode, body, eventQueuePtr); return; } else if (statusCode / 100 == 4) { this->onResponseError(request, statusCode, body, eventQueuePtr); } else if (statusCode / 100 == 5) { this->onResponseError(request, statusCode, body, eventQueuePtr); retry.numRetry += 1; this->tryRequest(request, *reqPtr, retry, eventQueuePtr); return; } else { this->onResponseError(request, statusCode, "unhandled response", eventQueuePtr); } } catch (const std::exception& e) { CCAPI_LOGGER_ERROR(e.what()); { std::ostringstream oss; oss << *reqPtr; CCAPI_LOGGER_ERROR("req = \n" + oss.str()); } { std::ostringstream oss; oss << *resPtr; CCAPI_LOGGER_ERROR("res = " + oss.str()); } this->onError(Event::Type::REQUEST_STATUS, Message::Type::GENERIC_ERROR, e, {request.getCorrelationId()}, eventQueuePtr); } CCAPI_LOGGER_DEBUG("retry = " + toString(retry)); if (retry.promisePtr) { retry.promisePtr->set_value(); } } virtual bool doesHttpBodyContainError(const std::string& body) { return false; } void tryRequest(const Request& request, http::request& req, const HttpRetry& retry, Queue* eventQueuePtr) { CCAPI_LOGGER_FUNCTION_ENTER; #if defined(CCAPI_ENABLE_LOG_DEBUG) || defined(CCAPI_ENABLE_LOG_TRACE) std::ostringstream oss; oss << req; CCAPI_LOGGER_DEBUG("req = \n" + oss.str()); #endif CCAPI_LOGGER_TRACE("retry = " + toString(retry)); if (retry.numRetry <= this->sessionOptions.httpMaxNumRetry && retry.numRedirect <= this->sessionOptions.httpMaxNumRedirect) { try { const auto& localIpAddress = request.getLocalIpAddress(); const auto& requestBaseUrl = request.getBaseUrl(); if (this->sessionOptions.enableOneHttpConnectionPerRequest || this->httpConnectionPool[localIpAddress][requestBaseUrl].empty() || std::chrono::duration_cast(request.getTimeSent() - this->httpConnectionPool[localIpAddress][requestBaseUrl].back()->lastReceiveDataTp) .count() >= this->sessionOptions.httpConnectionKeepAliveTimeoutSeconds) { this->httpConnectionPool[localIpAddress][requestBaseUrl].clear(); std::shared_ptr> streamPtr(nullptr); try { streamPtr = this->createStream>(this->serviceContextPtr->ioContextPtr, this->serviceContextPtr->sslContextPtr, this->hostRest); } catch (const beast::error_code& ec) { CCAPI_LOGGER_TRACE("fail"); this->onError(Event::Type::REQUEST_STATUS, Message::Type::REQUEST_FAILURE, ec, "create stream", {request.getCorrelationId()}, eventQueuePtr); return; } std::string host, port; if (requestBaseUrl.empty()) { host = this->hostRest; port = this->portRest; } else { host = request.getHost(); port = request.getPort(); } std::shared_ptr httpConnectionPtr(new HttpConnection(host, port, streamPtr)); CCAPI_LOGGER_WARN("about to perform request with new httpConnectionPtr " + toString(*httpConnectionPtr) + " for localIpAddress = " + localIpAddress + ", requestBaseUrl = " + toString(requestBaseUrl)); this->performRequestWithNewHttpConnection(httpConnectionPtr, request, req, retry, eventQueuePtr); } else { std::shared_ptr httpConnectionPtr = this->httpConnectionPool[localIpAddress][requestBaseUrl].back(); this->httpConnectionPool[localIpAddress][requestBaseUrl].pop_back(); CCAPI_LOGGER_TRACE("about to perform request with existing httpConnectionPtr " + toString(*httpConnectionPtr) + " for localIpAddress = " + localIpAddress + ", requestBaseUrl = " + toString(requestBaseUrl)); this->startWrite_2(httpConnectionPtr, request, req, retry, eventQueuePtr); } } catch (const std::exception& e) { CCAPI_LOGGER_ERROR(std::string("e.what() = ") + e.what()); this->onError(Event::Type::REQUEST_STATUS, Message::Type::REQUEST_FAILURE, e, {request.getCorrelationId()}, eventQueuePtr); } } else { std::string errorMessage = retry.numRetry > this->sessionOptions.httpMaxNumRetry ? "max retry exceeded" : "max redirect exceeded"; CCAPI_LOGGER_ERROR(errorMessage); CCAPI_LOGGER_DEBUG("retry = " + toString(retry)); this->onError(Event::Type::REQUEST_STATUS, Message::Type::REQUEST_FAILURE, std::runtime_error(errorMessage), {request.getCorrelationId()}, eventQueuePtr); if (retry.promisePtr) { retry.promisePtr->set_value(); } } CCAPI_LOGGER_FUNCTION_EXIT; } http::request convertRequest(const Request& request, const TimePoint& now) { CCAPI_LOGGER_FUNCTION_ENTER; auto credential = request.getCredential(); if (credential.empty()) { credential = this->credentialDefault; } auto instrument = request.getInstrument(); auto symbolId = instrument; CCAPI_LOGGER_TRACE("symbolId = " + symbolId); http::request req; req.version(11); if (this->sessionOptions.enableOneHttpConnectionPerRequest) { req.keep_alive(false); } else { req.keep_alive(true); } req.set(http::field::host, request.getBaseUrl().empty() ? this->hostRest : request.getHost()); req.set(http::field::user_agent, BOOST_BEAST_VERSION_STRING); this->convertRequestForRest(req, request, now, symbolId, credential); CCAPI_LOGGER_FUNCTION_EXIT; return req; } void substituteParam(std::string& target, const std::map& param, const std::map standardizationMap = {}) { for (const auto& kv : param) { auto key = standardizationMap.find(kv.first) != standardizationMap.end() ? standardizationMap.at(kv.first) : kv.first; auto value = kv.second; auto it = target.find(key); if (it != std::string::npos) { target = target.replace(it, key.length(), value); } } } void appendParam(std::string& queryString, const std::map& param, const std::map standardizationMap = {}, const std::map> conversionMap = {}) { int i = 0; for (const auto& kv : param) { std::string key = standardizationMap.find(kv.first) != standardizationMap.end() ? standardizationMap.at(kv.first) : kv.first; queryString += key; queryString += "="; std::string value = conversionMap.find(kv.first) != conversionMap.end() ? conversionMap.at(kv.first)(kv.second) : kv.second; queryString += Url::urlEncode(value); queryString += "&"; ++i; } } void appendSymbolId(rj::Value& rjValue, rj::Document::AllocatorType& allocator, const std::string& symbolId, const std::string symbolIdCalled) { rjValue.AddMember(rj::Value(symbolIdCalled.c_str(), allocator).Move(), rj::Value(symbolId.c_str(), allocator).Move(), allocator); } void appendSymbolId(std::string& queryString, const std::string& symbolId, const std::string symbolIdCalled) { if (!symbolId.empty()) { queryString += symbolIdCalled; queryString += "="; queryString += Url::urlEncode(symbolId); queryString += "&"; } } void setupCredential(std::vector nameList) { for (const auto& x : nameList) { if (this->sessionConfigs.getCredential().find(x) != this->sessionConfigs.getCredential().end()) { this->credentialDefault.insert(std::make_pair(x, this->sessionConfigs.getCredential().at(x))); } else if (!UtilSystem::getEnvAsString(x).empty()) { this->credentialDefault.insert(std::make_pair(x, UtilSystem::getEnvAsString(x))); } } } http::verb convertHttpMethodStringToMethod(const std::string& methodString) { std::string methodStringUpper = UtilString::toUpper(methodString); return http::string_to_verb(methodStringUpper); } #ifdef CCAPI_LEGACY_USE_WEBSOCKETPP SslContextPtr onTlsInit(wspp::connection_hdl hdl) { return this->serviceContextPtr->sslContextPtr; } WsConnection& getWsConnectionFromConnectionPtr(TlsClient::connection_ptr connectionPtr) { return this->wsConnectionByIdMap.at(this->connectionAddressToString(connectionPtr)); } std::string connectionAddressToString(const TlsClient::connection_ptr con) { const void* address = static_cast(con.get()); std::stringstream ss; ss << address; return ss.str(); } void close(WsConnection& wsConnection, wspp::connection_hdl hdl, wspp::close::status::value const code, std::string const& reason, ErrorCode& ec) { if (wsConnection.status == WsConnection::Status::CLOSING) { CCAPI_LOGGER_WARN("websocket connection is already in the state of closing"); return; } wsConnection.status = WsConnection::Status::CLOSING; this->serviceContextPtr->tlsClientPtr->close(hdl, code, reason, ec); } virtual void prepareConnect(WsConnection& wsConnection) { this->connect(wsConnection); } virtual void connect(WsConnection& wsConnection) { CCAPI_LOGGER_FUNCTION_ENTER; wsConnection.status = WsConnection::Status::CONNECTING; CCAPI_LOGGER_DEBUG("connection initialization on dummy id " + wsConnection.id); std::string url = wsConnection.url; CCAPI_LOGGER_DEBUG("url = " + url); this->serviceContextPtr->tlsClientPtr->set_tls_init_handler(std::bind(&Service::onTlsInit, shared_from_this(), std::placeholders::_1)); CCAPI_LOGGER_DEBUG("endpoint tls init handler set"); ErrorCode ec; TlsClient::connection_ptr con = this->serviceContextPtr->tlsClientPtr->get_connection(url, ec); for (const auto& kv : wsConnection.headers) { con->append_header(kv.first, kv.second); } wsConnection.id = this->connectionAddressToString(con); CCAPI_LOGGER_DEBUG("connection initialization on actual id " + wsConnection.id); if (ec) { CCAPI_LOGGER_FATAL("connection initialization error: " + ec.message()); } this->wsConnectionByIdMap.insert(std::pair(wsConnection.id, wsConnection)); CCAPI_LOGGER_DEBUG("this->wsConnectionByIdMap = " + toString(this->wsConnectionByIdMap)); con->set_open_handler(std::bind(&Service::onOpen, shared_from_this(), std::placeholders::_1)); con->set_fail_handler(std::bind(&Service::onFail, shared_from_this(), std::placeholders::_1)); con->set_close_handler(std::bind(&Service::onClose, shared_from_this(), std::placeholders::_1)); con->set_message_handler(std::bind(&Service::onMessage, shared_from_this(), std::placeholders::_1, std::placeholders::_2)); if (this->sessionOptions.enableCheckPingPongWebsocketProtocolLevel) { con->set_pong_handler(std::bind(&Service::onPong, shared_from_this(), std::placeholders::_1, std::placeholders::_2)); } con->set_ping_handler(std::bind(&Service::onPing, shared_from_this(), std::placeholders::_1, std::placeholders::_2)); this->serviceContextPtr->tlsClientPtr->connect(con); CCAPI_LOGGER_FUNCTION_EXIT; } virtual void onOpen(wspp::connection_hdl hdl) { CCAPI_LOGGER_FUNCTION_ENTER; auto now = UtilTime::now(); WsConnection& wsConnection = this->getWsConnectionFromConnectionPtr(this->serviceContextPtr->tlsClientPtr->get_con_from_hdl(hdl)); wsConnection.status = WsConnection::Status::OPEN; wsConnection.hdl = hdl; CCAPI_LOGGER_INFO("connection " + toString(wsConnection) + " established"); auto urlBase = UtilString::split(wsConnection.url, "?").at(0); this->connectNumRetryOnFailByConnectionUrlMap[urlBase] = 0; Event event; event.setType(Event::Type::SESSION_STATUS); Message message; message.setTimeReceived(now); message.setType(Message::Type::SESSION_CONNECTION_UP); std::vector correlationIdList; for (const auto& subscription : wsConnection.subscriptionList) { correlationIdList.push_back(subscription.getCorrelationId()); } CCAPI_LOGGER_DEBUG("correlationIdList = " + toString(correlationIdList)); message.setCorrelationIdList(correlationIdList); Element element; element.insert(CCAPI_CONNECTION_ID, wsConnection.id); element.insert(CCAPI_CONNECTION_URL, wsConnection.url); message.setElementList({element}); event.setMessageList({message}); this->eventHandler(event, nullptr); if (this->enableCheckPingPongWebsocketProtocolLevel) { this->setPingPongTimer(PingPongMethod::WEBSOCKET_PROTOCOL_LEVEL, wsConnection, hdl, [that = shared_from_this()](wspp::connection_hdl hdl, ErrorCode& ec) { that->ping(hdl, "", ec); }); } if (this->enableCheckPingPongWebsocketApplicationLevel) { this->setPingPongTimer(PingPongMethod::WEBSOCKET_APPLICATION_LEVEL, wsConnection, hdl, [that = shared_from_this()](wspp::connection_hdl hdl, ErrorCode& ec) { that->pingOnApplicationLevel(hdl, ec); }); } } virtual void onFail_(WsConnection& wsConnection) { wsConnection.status = WsConnection::Status::FAILED; this->onError(Event::Type::SUBSCRIPTION_STATUS, Message::Type::SUBSCRIPTION_FAILURE, "connection " + toString(wsConnection) + " has failed before opening"); WsConnection thisWsConnection = wsConnection; this->wsConnectionByIdMap.erase(thisWsConnection.id); auto urlBase = UtilString::split(thisWsConnection.url, "?").at(0); long seconds = std::round(UtilAlgorithm::exponentialBackoff(1, 1, 2, std::min(this->connectNumRetryOnFailByConnectionUrlMap[urlBase], 6))); CCAPI_LOGGER_INFO("about to set timer for " + toString(seconds) + " seconds"); if (this->connectRetryOnFailTimerByConnectionIdMap.find(thisWsConnection.id) != this->connectRetryOnFailTimerByConnectionIdMap.end()) { this->connectRetryOnFailTimerByConnectionIdMap.at(thisWsConnection.id)->cancel(); } TimerPtr timerPtr(new net::steady_timer(*this->serviceContextPtr->ioContextPtr, std::chrono::milliseconds(seconds * 1000))); timerPtr->async_wait([thisWsConnection, that = shared_from_this(), urlBase](ErrorCode const& ec) { if (that->wsConnectionByIdMap.find(thisWsConnection.id) == that->wsConnectionByIdMap.end()) { if (ec) { CCAPI_LOGGER_ERROR("wsConnection = " + toString(thisWsConnection) + ", connect retry on fail timer error: " + ec.message()); that->onError(Event::Type::SUBSCRIPTION_STATUS, Message::Type::GENERIC_ERROR, ec, "timer"); } else { CCAPI_LOGGER_INFO("about to retry"); auto thatWsConnection = thisWsConnection; thatWsConnection.assignDummyId(); that->prepareConnect(thatWsConnection); that->connectNumRetryOnFailByConnectionUrlMap[urlBase] += 1; } } }); this->connectRetryOnFailTimerByConnectionIdMap[thisWsConnection.id] = timerPtr; } virtual void onFail(wspp::connection_hdl hdl) { CCAPI_LOGGER_FUNCTION_ENTER; WsConnection& wsConnection = this->getWsConnectionFromConnectionPtr(this->serviceContextPtr->tlsClientPtr->get_con_from_hdl(hdl)); this->clearStates(wsConnection); this->onFail_(wsConnection); CCAPI_LOGGER_FUNCTION_EXIT; } virtual void clearStates(WsConnection& wsConnection) { CCAPI_LOGGER_INFO("clear states for wsConnection " + toString(wsConnection)); this->shouldProcessRemainingMessageOnClosingByConnectionIdMap.erase(wsConnection.id); this->lastPongTpByMethodByConnectionIdMap.erase(wsConnection.id); this->extraPropertyByConnectionIdMap.erase(wsConnection.id); if (this->pingTimerByMethodByConnectionIdMap.find(wsConnection.id) != this->pingTimerByMethodByConnectionIdMap.end()) { for (const auto& x : this->pingTimerByMethodByConnectionIdMap.at(wsConnection.id)) { x.second->cancel(); } this->pingTimerByMethodByConnectionIdMap.erase(wsConnection.id); } if (this->pongTimeOutTimerByMethodByConnectionIdMap.find(wsConnection.id) != this->pongTimeOutTimerByMethodByConnectionIdMap.end()) { for (const auto& x : this->pongTimeOutTimerByMethodByConnectionIdMap.at(wsConnection.id)) { x.second->cancel(); } this->pongTimeOutTimerByMethodByConnectionIdMap.erase(wsConnection.id); } // auto urlBase = UtilString::split(wsConnection.url, "?").at(0); // this->connectNumRetryOnFailByConnectionUrlMap.erase(urlBase); if (this->connectRetryOnFailTimerByConnectionIdMap.find(wsConnection.id) != this->connectRetryOnFailTimerByConnectionIdMap.end()) { this->connectRetryOnFailTimerByConnectionIdMap.at(wsConnection.id)->cancel(); this->connectRetryOnFailTimerByConnectionIdMap.erase(wsConnection.id); } } virtual void onClose(wspp::connection_hdl hdl) { CCAPI_LOGGER_FUNCTION_ENTER; auto now = UtilTime::now(); TlsClient::connection_ptr con = this->serviceContextPtr->tlsClientPtr->get_con_from_hdl(hdl); WsConnection& wsConnection = this->getWsConnectionFromConnectionPtr(con); wsConnection.status = WsConnection::Status::CLOSED; CCAPI_LOGGER_INFO("connection " + toString(wsConnection) + " is closed"); std::stringstream s; s << "close code: " << con->get_remote_close_code() << " (" << websocketpp::close::status::get_string(con->get_remote_close_code()) << "), close reason: " << con->get_remote_close_reason(); std::string reason = s.str(); CCAPI_LOGGER_INFO("reason is " + reason); Event event; event.setType(Event::Type::SESSION_STATUS); Message message; message.setTimeReceived(now); message.setType(Message::Type::SESSION_CONNECTION_DOWN); Element element; element.insert(CCAPI_CONNECTION_ID, wsConnection.id); element.insert(CCAPI_CONNECTION_URL, wsConnection.url); element.insert(CCAPI_REASON, reason); message.setElementList({element}); std::vector correlationIdList; for (const auto& subscription : wsConnection.subscriptionList) { correlationIdList.push_back(subscription.getCorrelationId()); } CCAPI_LOGGER_DEBUG("correlationIdList = " + toString(correlationIdList)); message.setCorrelationIdList(correlationIdList); event.setMessageList({message}); this->eventHandler(event, nullptr); CCAPI_LOGGER_INFO("connection " + toString(wsConnection) + " is closed"); this->clearStates(wsConnection); WsConnection thisWsConnection = wsConnection; this->wsConnectionByIdMap.erase(wsConnection.id); if (this->shouldContinue.load()) { thisWsConnection.assignDummyId(); this->prepareConnect(thisWsConnection); } CCAPI_LOGGER_FUNCTION_EXIT; } void onMessage(wspp::connection_hdl hdl, TlsClient::message_ptr msg) { auto now = UtilTime::now(); WsConnection& wsConnection = this->getWsConnectionFromConnectionPtr(this->serviceContextPtr->tlsClientPtr->get_con_from_hdl(hdl)); CCAPI_LOGGER_DEBUG("received a message from connection " + toString(wsConnection)); if (wsConnection.status != WsConnection::Status::OPEN && !this->shouldProcessRemainingMessageOnClosingByConnectionIdMap[wsConnection.id]) { CCAPI_LOGGER_WARN("should not process remaining message on closing"); return; } auto opcode = msg->get_opcode(); // CCAPI_LOGGER_DEBUG("opcode = " + toString(opcode)); if (msg->get_opcode() == websocketpp::frame::opcode::text) { const std::string& textMessage = msg->get_payload(); CCAPI_LOGGER_DEBUG("received a text message: " + textMessage); try { this->onTextMessage(hdl, textMessage, now); } catch (const std::exception& e) { CCAPI_LOGGER_ERROR("textMessage = " + textMessage); this->onError(Event::Type::SUBSCRIPTION_STATUS, Message::Type::GENERIC_ERROR, e); } } else if (opcode == websocketpp::frame::opcode::binary) { #if defined(CCAPI_ENABLE_SERVICE_MARKET_DATA) && \ (defined(CCAPI_ENABLE_EXCHANGE_HUOBI) || defined(CCAPI_ENABLE_EXCHANGE_HUOBI_USDT_SWAP) || defined(CCAPI_ENABLE_EXCHANGE_HUOBI_COIN_SWAP)) || \ defined(CCAPI_ENABLE_SERVICE_EXECUTION_MANAGEMENT) && \ (defined(CCAPI_ENABLE_EXCHANGE_HUOBI_USDT_SWAP) || defined(CCAPI_ENABLE_EXCHANGE_HUOBI_COIN_SWAP) || defined(CCAPI_ENABLE_EXCHANGE_BITMART)) if (this->needDecompressWebsocketMessage) { std::string decompressed; const std::string& payload = msg->get_payload(); try { ErrorCode ec = this->inflater.decompress(reinterpret_cast(&payload[0]), payload.size(), decompressed); if (ec) { CCAPI_LOGGER_FATAL(ec.message()); } CCAPI_LOGGER_DEBUG("decompressed = " + decompressed); this->onTextMessage(hdl, decompressed, now); } catch (const std::exception& e) { std::stringstream ss; ss << std::hex << std::setfill('0'); for (int i = 0; i < payload.size(); ++i) { ss << std::setw(2) << static_cast(reinterpret_cast(&payload[0])[i]); } CCAPI_LOGGER_ERROR("binaryMessage = " + ss.str()); this->onError(Event::Type::SUBSCRIPTION_STATUS, Message::Type::GENERIC_ERROR, e); } ErrorCode ec = this->inflater.inflate_reset(); if (ec) { this->onError(Event::Type::SUBSCRIPTION_STATUS, Message::Type::GENERIC_ERROR, ec, "decompress"); } } #endif } } void onPong(wspp::connection_hdl hdl, std::string payload) { auto now = UtilTime::now(); this->onPongByMethod(PingPongMethod::WEBSOCKET_PROTOCOL_LEVEL, hdl, payload, now); } void onPongByMethod(PingPongMethod method, wspp::connection_hdl hdl, const std::string& textMessage, const TimePoint& timeReceived) { CCAPI_LOGGER_FUNCTION_ENTER; WsConnection& wsConnection = this->getWsConnectionFromConnectionPtr(this->serviceContextPtr->tlsClientPtr->get_con_from_hdl(hdl)); CCAPI_LOGGER_TRACE(pingPongMethodToString(method) + ": received a pong from " + toString(wsConnection)); this->lastPongTpByMethodByConnectionIdMap[wsConnection.id][method] = timeReceived; CCAPI_LOGGER_FUNCTION_EXIT; } bool onPing(wspp::connection_hdl hdl, std::string payload) { CCAPI_LOGGER_FUNCTION_ENTER; WsConnection& wsConnection = this->getWsConnectionFromConnectionPtr(this->serviceContextPtr->tlsClientPtr->get_con_from_hdl(hdl)); CCAPI_LOGGER_TRACE("received a ping from " + toString(wsConnection)); CCAPI_LOGGER_FUNCTION_EXIT; return true; } void send(wspp::connection_hdl hdl, std::string const& payload, wspp::frame::opcode::value op, ErrorCode& ec) { this->serviceContextPtr->tlsClientPtr->send(hdl, payload, op, ec); } void ping(wspp::connection_hdl hdl, std::string const& payload, ErrorCode& ec) { this->serviceContextPtr->tlsClientPtr->ping(hdl, payload, ec); } virtual void pingOnApplicationLevel(wspp::connection_hdl hdl, ErrorCode& ec) {} void setPingPongTimer(PingPongMethod method, WsConnection& wsConnection, wspp::connection_hdl hdl, std::function pingMethod) { CCAPI_LOGGER_FUNCTION_ENTER; CCAPI_LOGGER_TRACE("method = " + pingPongMethodToString(method)); auto pingIntervalMilliseconds = this->pingIntervalMillisecondsByMethodMap[method]; auto pongTimeoutMilliseconds = this->pongTimeoutMillisecondsByMethodMap[method]; CCAPI_LOGGER_TRACE("pingIntervalMilliseconds = " + toString(pingIntervalMilliseconds)); CCAPI_LOGGER_TRACE("pongTimeoutMilliseconds = " + toString(pongTimeoutMilliseconds)); if (pingIntervalMilliseconds <= pongTimeoutMilliseconds) { return; } if (wsConnection.status == WsConnection::Status::OPEN) { if (this->pingTimerByMethodByConnectionIdMap.find(wsConnection.id) != this->pingTimerByMethodByConnectionIdMap.end() && this->pingTimerByMethodByConnectionIdMap.at(wsConnection.id).find(method) != this->pingTimerByMethodByConnectionIdMap.at(wsConnection.id).end()) { this->pingTimerByMethodByConnectionIdMap.at(wsConnection.id).at(method)->cancel(); } TimerPtr timerPtr( new net::steady_timer(*this->serviceContextPtr->ioContextPtr, std::chrono::milliseconds(pingIntervalMilliseconds - pongTimeoutMilliseconds))); timerPtr->async_wait([wsConnection, that = shared_from_this(), hdl, pingMethod, pongTimeoutMilliseconds, method](ErrorCode const& ec) { if (that->wsConnectionByIdMap.find(wsConnection.id) != that->wsConnectionByIdMap.end()) { if (ec) { CCAPI_LOGGER_ERROR("wsConnection = " + toString(wsConnection) + ", ping timer error: " + ec.message()); that->onError(Event::Type::SUBSCRIPTION_STATUS, Message::Type::GENERIC_ERROR, ec, "timer"); } else { if (that->wsConnectionByIdMap.at(wsConnection.id).status == WsConnection::Status::OPEN) { ErrorCode ec; pingMethod(hdl, ec); if (ec) { that->onError(Event::Type::SUBSCRIPTION_STATUS, Message::Type::GENERIC_ERROR, ec, "ping"); } if (pongTimeoutMilliseconds <= 0) { return; } if (that->pongTimeOutTimerByMethodByConnectionIdMap.find(wsConnection.id) != that->pongTimeOutTimerByMethodByConnectionIdMap.end() && that->pongTimeOutTimerByMethodByConnectionIdMap.at(wsConnection.id).find(method) != that->pongTimeOutTimerByMethodByConnectionIdMap.at(wsConnection.id).end()) { that->pongTimeOutTimerByMethodByConnectionIdMap.at(wsConnection.id).at(method)->cancel(); } TimerPtr timerPtr(new net::steady_timer(*that->serviceContextPtr->ioContextPtr, std::chrono::milliseconds(pongTimeoutMilliseconds))); timerPtr->async_wait([wsConnection, that, hdl, pingMethod, pongTimeoutMilliseconds, method](ErrorCode const& ec) { if (that->wsConnectionByIdMap.find(wsConnection.id) != that->wsConnectionByIdMap.end()) { if (ec) { CCAPI_LOGGER_ERROR("wsConnection = " + toString(wsConnection) + ", pong time out timer error: " + ec.message()); that->onError(Event::Type::SUBSCRIPTION_STATUS, Message::Type::GENERIC_ERROR, ec, "timer"); } else { if (that->wsConnectionByIdMap.at(wsConnection.id).status == WsConnection::Status::OPEN) { auto now = UtilTime::now(); if (that->lastPongTpByMethodByConnectionIdMap.find(wsConnection.id) != that->lastPongTpByMethodByConnectionIdMap.end() && that->lastPongTpByMethodByConnectionIdMap.at(wsConnection.id).find(method) != that->lastPongTpByMethodByConnectionIdMap.at(wsConnection.id).end() && std::chrono::duration_cast(now - that->lastPongTpByMethodByConnectionIdMap.at(wsConnection.id).at(method)) .count() >= pongTimeoutMilliseconds) { auto thisWsConnection = wsConnection; ErrorCode ec; that->close(thisWsConnection, hdl, websocketpp::close::status::normal, "pong timeout", ec); if (ec) { that->onError(Event::Type::SUBSCRIPTION_STATUS, Message::Type::GENERIC_ERROR, ec, "shutdown"); } that->shouldProcessRemainingMessageOnClosingByConnectionIdMap[thisWsConnection.id] = true; } else { auto thisWsConnection = wsConnection; that->setPingPongTimer(method, thisWsConnection, hdl, pingMethod); } } } } }); that->pongTimeOutTimerByMethodByConnectionIdMap[wsConnection.id][method] = timerPtr; } } } }); this->pingTimerByMethodByConnectionIdMap[wsConnection.id][method] = timerPtr; } CCAPI_LOGGER_FUNCTION_EXIT; } virtual void onTextMessage(wspp::connection_hdl hdl, const std::string& textMessage, const TimePoint& timeReceived) {} #else void close(std::shared_ptr wsConnectionPtr, beast::websocket::close_code const code, beast::websocket::close_reason reason, ErrorCode& ec) { WsConnection& wsConnection = *wsConnectionPtr; if (wsConnection.status == WsConnection::Status::CLOSING) { CCAPI_LOGGER_WARN("websocket connection is already in the state of closing"); return; } wsConnection.status = WsConnection::Status::CLOSING; wsConnection.remoteCloseCode = code; wsConnection.remoteCloseReason = reason; wsConnectionPtr->streamPtr->async_close(code, beast::bind_front_handler(&Service::onClose, shared_from_this(), wsConnectionPtr)); } virtual void prepareConnect(std::shared_ptr wsConnectionPtr) { this->connect(wsConnectionPtr); } virtual void connect(std::shared_ptr wsConnectionPtr) { CCAPI_LOGGER_FUNCTION_ENTER; WsConnection& wsConnection = *wsConnectionPtr; wsConnection.status = WsConnection::Status::CONNECTING; CCAPI_LOGGER_DEBUG("connection initialization on id " + wsConnection.id); std::string url = wsConnection.getUrl(); CCAPI_LOGGER_DEBUG("url = " + url); this->startResolveWs(wsConnectionPtr); CCAPI_LOGGER_FUNCTION_EXIT; } void startResolveWs(std::shared_ptr wsConnectionPtr) { std::shared_ptr newResolverPtr(new tcp::resolver(*this->serviceContextPtr->ioContextPtr)); CCAPI_LOGGER_TRACE("wsConnectionPtr = " + wsConnectionPtr->toString()); CCAPI_LOGGER_TRACE("wsConnectionPtr->host = " + wsConnectionPtr->host); CCAPI_LOGGER_TRACE("wsConnectionPtr->port = " + wsConnectionPtr->port); newResolverPtr->async_resolve(wsConnectionPtr->host, wsConnectionPtr->port, beast::bind_front_handler(&Service::onResolveWs, shared_from_this(), wsConnectionPtr, newResolverPtr)); } void onResolveWs(std::shared_ptr wsConnectionPtr, std::shared_ptr newResolverPtr, beast::error_code ec, tcp::resolver::results_type tcpNewResolverResultsWs) { if (ec) { CCAPI_LOGGER_TRACE("dns resolve fail"); this->onFail(wsConnectionPtr); return; } this->startConnectWs(wsConnectionPtr, this->sessionOptions.websocketConnectTimeoutMilliseconds, tcpNewResolverResultsWs); } void startConnectWs(std::shared_ptr wsConnectionPtr, long timeoutMilliseconds, tcp::resolver::results_type tcpResolverResults) { beast::websocket::stream>& stream = *wsConnectionPtr->streamPtr; if (timeoutMilliseconds > 0) { beast::get_lowest_layer(stream).expires_after(std::chrono::milliseconds(timeoutMilliseconds)); } // Set SNI Hostname (many hosts need this to handshake successfully) CCAPI_LOGGER_TRACE("wsConnectionPtr->host = " + wsConnectionPtr->host) if (!SSL_set_tlsext_host_name(stream.next_layer().native_handle(), wsConnectionPtr->host.c_str())) { beast::error_code ec{static_cast(::ERR_get_error()), net::error::get_ssl_category()}; CCAPI_LOGGER_DEBUG("error SSL_set_tlsext_host_name: " + ec.message()); this->onError(Event::Type::SUBSCRIPTION_STATUS, Message::Type::SUBSCRIPTION_FAILURE, ec, "set SNI Hostname", wsConnectionPtr->correlationIdList); return; } CCAPI_LOGGER_TRACE("before async_connect"); beast::get_lowest_layer(stream).async_connect(tcpResolverResults, beast::bind_front_handler(&Service::onConnectWs, shared_from_this(), wsConnectionPtr)); CCAPI_LOGGER_TRACE("after async_connect"); } void onConnectWs(std::shared_ptr wsConnectionPtr, beast::error_code ec, tcp::resolver::results_type::endpoint_type ep) { CCAPI_LOGGER_TRACE("async_connect callback start"); if (ec) { CCAPI_LOGGER_TRACE("fail"); this->onFail(wsConnectionPtr); return; } CCAPI_LOGGER_TRACE("connected"); CCAPI_LOGGER_TRACE("ep.port() = " + std::to_string(ep.port())); wsConnectionPtr->hostHttpHeaderValue = this->hostHttpHeaderValueIgnorePort ? wsConnectionPtr->host : wsConnectionPtr->host + ':' + std::to_string(ep.port()); CCAPI_LOGGER_TRACE("wsConnectionPtr->hostHttpHeaderValue = " + wsConnectionPtr->hostHttpHeaderValue); beast::websocket::stream>& stream = *wsConnectionPtr->streamPtr; beast::get_lowest_layer(stream).socket().set_option(tcp::no_delay(true)); CCAPI_LOGGER_TRACE("before ssl async_handshake"); stream.next_layer().async_handshake(ssl::stream_base::client, beast::bind_front_handler(&Service::onSslHandshakeWs, shared_from_this(), wsConnectionPtr)); CCAPI_LOGGER_TRACE("after ssl async_handshake"); } void onSslHandshakeWs(std::shared_ptr wsConnectionPtr, beast::error_code ec) { CCAPI_LOGGER_TRACE("ssl async_handshake callback start"); if (ec) { CCAPI_LOGGER_TRACE("ssl handshake fail"); this->onFail(wsConnectionPtr); return; } CCAPI_LOGGER_TRACE("ssl handshaked"); beast::websocket::stream>& stream = *wsConnectionPtr->streamPtr; beast::get_lowest_layer(stream).expires_never(); beast::websocket::stream_base::timeout opt{std::chrono::milliseconds(this->sessionOptions.websocketConnectTimeoutMilliseconds), std::chrono::milliseconds(this->sessionOptions.pongWebsocketProtocolLevelTimeoutMilliseconds), true}; stream.set_option(opt); stream.set_option(beast::websocket::stream_base::decorator([wsConnectionPtr](beast::websocket::request_type& req) { req.set(http::field::user_agent, std::string(BOOST_BEAST_VERSION_STRING)); for (const auto& kv : wsConnectionPtr->headers) { req.set(kv.first, kv.second); } })); CCAPI_LOGGER_TRACE("before ws async_handshake"); stream.async_handshake(wsConnectionPtr->hostHttpHeaderValue, wsConnectionPtr->path, beast::bind_front_handler(&Service::onWsHandshakeWs, shared_from_this(), wsConnectionPtr)); CCAPI_LOGGER_TRACE("after ws async_handshake"); } void onWsHandshakeWs(std::shared_ptr wsConnectionPtr, beast::error_code ec) { CCAPI_LOGGER_TRACE("ws async_handshake callback start"); if (ec) { CCAPI_LOGGER_TRACE("ws handshake fail"); this->onFail(wsConnectionPtr); return; } CCAPI_LOGGER_TRACE("ws handshaked"); this->onOpen(wsConnectionPtr); this->wsConnectionByIdMap.insert(std::make_pair(wsConnectionPtr->id, wsConnectionPtr)); CCAPI_LOGGER_TRACE("about to start read"); this->startReadWs(wsConnectionPtr); auto& stream = *wsConnectionPtr->streamPtr; stream.control_callback([wsConnectionPtr, that = shared_from_this()](boost::beast::websocket::frame_type kind, boost::beast::string_view payload) { that->onControlCallback(wsConnectionPtr, kind, payload); }); } void startReadWs(std::shared_ptr wsConnectionPtr) { auto& stream = *wsConnectionPtr->streamPtr; CCAPI_LOGGER_TRACE("before async_read"); auto& connectionId = wsConnectionPtr->id; auto& readMessageBuffer = this->readMessageBufferByConnectionIdMap[connectionId]; stream.async_read(readMessageBuffer, beast::bind_front_handler(&Service::onReadWs, shared_from_this(), wsConnectionPtr)); CCAPI_LOGGER_TRACE("after async_read"); } void onReadWs(std::shared_ptr wsConnectionPtr, const ErrorCode& ec, std::size_t n) { CCAPI_LOGGER_FUNCTION_ENTER; CCAPI_LOGGER_TRACE("n = " + toString(n)); auto now = UtilTime::now(); if (ec) { if (ec == beast::error::timeout) { CCAPI_LOGGER_TRACE("timeout, connection closed"); } else { CCAPI_LOGGER_TRACE("fail"); Event event; event.setType(Event::Type::SESSION_STATUS); Message message; message.setTimeReceived(now); message.setType(Message::Type::SESSION_CONNECTION_DOWN); message.setCorrelationIdList(wsConnectionPtr->correlationIdList); Element element(true); auto& connectionId = wsConnectionPtr->id; element.insert(CCAPI_CONNECTION_ID, connectionId); message.setElementList({element}); event.setMessageList({message}); this->eventHandler(event, nullptr); this->onFail(wsConnectionPtr); } return; } if (wsConnectionPtr->status != WsConnection::Status::OPEN) { CCAPI_LOGGER_WARN("should not process remaining message on closing"); return; } auto& connectionId = wsConnectionPtr->id; auto& readMessageBuffer = this->readMessageBufferByConnectionIdMap[connectionId]; this->onMessage(wsConnectionPtr, (const char*)readMessageBuffer.data().data(), readMessageBuffer.size()); readMessageBuffer.consume(readMessageBuffer.size()); this->startReadWs(wsConnectionPtr); this->onPongByMethod(PingPongMethod::WEBSOCKET_PROTOCOL_LEVEL, wsConnectionPtr, now, false); CCAPI_LOGGER_FUNCTION_EXIT; } virtual void onOpen(std::shared_ptr wsConnectionPtr) { CCAPI_LOGGER_FUNCTION_ENTER; auto now = UtilTime::now(); WsConnection& wsConnection = *wsConnectionPtr; wsConnection.status = WsConnection::Status::OPEN; CCAPI_LOGGER_INFO("connection " + toString(wsConnection) + " established"); auto urlBase = UtilString::split(wsConnection.getUrl(), "?").at(0); this->connectNumRetryOnFailByConnectionUrlMap[urlBase] = 0; Event event; event.setType(Event::Type::SESSION_STATUS); Message message; message.setTimeReceived(now); message.setType(Message::Type::SESSION_CONNECTION_UP); std::vector correlationIdList = wsConnection.correlationIdList; CCAPI_LOGGER_DEBUG("correlationIdList = " + toString(correlationIdList)); message.setCorrelationIdList(correlationIdList); Element element; element.insert(CCAPI_CONNECTION_ID, wsConnection.id); element.insert(CCAPI_CONNECTION_URL, wsConnection.getUrl()); message.setElementList({element}); event.setMessageList({message}); this->eventHandler(event, nullptr); if (this->enableCheckPingPongWebsocketProtocolLevel) { this->setPingPongTimer(PingPongMethod::WEBSOCKET_PROTOCOL_LEVEL, wsConnectionPtr, [wsConnectionPtr, that = shared_from_this()](ErrorCode& ec) { that->ping(wsConnectionPtr, "", ec); }); } if (this->enableCheckPingPongWebsocketApplicationLevel) { this->setPingPongTimer(PingPongMethod::WEBSOCKET_APPLICATION_LEVEL, wsConnectionPtr, [wsConnectionPtr, that = shared_from_this()](ErrorCode& ec) { that->pingOnApplicationLevel(wsConnectionPtr, ec); }); } } void writeMessage(std::shared_ptr wsConnectionPtr, const char* data, size_t dataSize) { if (wsConnectionPtr->status != WsConnection::Status::OPEN) { CCAPI_LOGGER_WARN("should write no more messages"); return; } auto& connectionId = wsConnectionPtr->id; auto& writeMessageBuffer = this->writeMessageBufferByConnectionIdMap[connectionId]; auto& writeMessageBufferWrittenLength = this->writeMessageBufferWrittenLengthByConnectionIdMap[connectionId]; auto& writeMessageBufferBoundary = this->writeMessageBufferBoundaryByConnectionIdMap[connectionId]; size_t n = writeMessageBufferWrittenLength; memcpy(writeMessageBuffer.data() + n, data, dataSize); writeMessageBufferBoundary.push_back(dataSize); n += dataSize; CCAPI_LOGGER_TRACE("connectionId = " + connectionId); CCAPI_LOGGER_DEBUG("about to send " + std::string(data, dataSize)); CCAPI_LOGGER_TRACE("writeMessageBufferWrittenLength = " + toString(writeMessageBufferWrittenLength)); if (writeMessageBufferWrittenLength == 0) { CCAPI_LOGGER_TRACE("about to start write"); this->startWriteWs(wsConnectionPtr, writeMessageBuffer.data(), writeMessageBufferBoundary.front()); } writeMessageBufferWrittenLength = n; CCAPI_LOGGER_TRACE("writeMessageBufferWrittenLength = " + toString(writeMessageBufferWrittenLength)); CCAPI_LOGGER_TRACE("writeMessageBufferBoundary = " + toString(writeMessageBufferBoundary)); } void startWriteWs(std::shared_ptr wsConnectionPtr, const char* data, size_t numBytesToWrite) { auto& stream = *wsConnectionPtr->streamPtr; CCAPI_LOGGER_TRACE("before async_write"); CCAPI_LOGGER_TRACE("numBytesToWrite = " + toString(numBytesToWrite)); stream.binary(false); stream.async_write(net::buffer(data, numBytesToWrite), beast::bind_front_handler(&Service::onWriteWs, shared_from_this(), wsConnectionPtr)); CCAPI_LOGGER_TRACE("after async_write"); } void onWriteWs(std::shared_ptr wsConnectionPtr, const ErrorCode& ec, std::size_t n) { CCAPI_LOGGER_FUNCTION_ENTER; auto& connectionId = wsConnectionPtr->id; auto& writeMessageBuffer = this->writeMessageBufferByConnectionIdMap[connectionId]; auto& writeMessageBufferWrittenLength = this->writeMessageBufferWrittenLengthByConnectionIdMap[connectionId]; auto& writeMessageBufferBoundary = this->writeMessageBufferBoundaryByConnectionIdMap[connectionId]; writeMessageBufferWrittenLength -= writeMessageBufferBoundary.front(); writeMessageBufferBoundary.erase(writeMessageBufferBoundary.begin()); CCAPI_LOGGER_TRACE("writeMessageBufferWrittenLength = " + toString(writeMessageBufferWrittenLength)); CCAPI_LOGGER_TRACE("writeMessageBufferBoundary = " + toString(writeMessageBufferBoundary)); if (writeMessageBufferWrittenLength > 0) { std::memmove(writeMessageBuffer.data(), writeMessageBuffer.data() + n, writeMessageBufferWrittenLength); CCAPI_LOGGER_TRACE("about to start write"); this->startWriteWs(wsConnectionPtr, writeMessageBuffer.data(), writeMessageBufferBoundary.front()); } CCAPI_LOGGER_FUNCTION_EXIT; } virtual void onFail_(std::shared_ptr wsConnectionPtr) { WsConnection& wsConnection = *wsConnectionPtr; wsConnection.status = WsConnection::Status::FAILED; this->onError(Event::Type::SUBSCRIPTION_STATUS, Message::Type::SUBSCRIPTION_FAILURE, "connection " + toString(wsConnection) + " has failed before opening"); WsConnection thisWsConnection = wsConnection; this->wsConnectionByIdMap.erase(thisWsConnection.id); auto urlBase = UtilString::split(thisWsConnection.getUrl(), "?").at(0); long seconds = std::round(UtilAlgorithm::exponentialBackoff(1, 1, 2, std::min(this->connectNumRetryOnFailByConnectionUrlMap[urlBase], 6))); CCAPI_LOGGER_INFO("about to set timer for " + toString(seconds) + " seconds"); if (this->connectRetryOnFailTimerByConnectionIdMap.find(thisWsConnection.id) != this->connectRetryOnFailTimerByConnectionIdMap.end()) { this->connectRetryOnFailTimerByConnectionIdMap.at(thisWsConnection.id)->cancel(); } TimerPtr timerPtr(new net::steady_timer(*this->serviceContextPtr->ioContextPtr, std::chrono::milliseconds(seconds * 1000))); timerPtr->async_wait([wsConnectionPtr, that = shared_from_this(), urlBase](ErrorCode const& ec) { WsConnection& thisWsConnection = *wsConnectionPtr; if (that->wsConnectionByIdMap.find(thisWsConnection.id) == that->wsConnectionByIdMap.end()) { if (ec) { CCAPI_LOGGER_ERROR("wsConnection = " + toString(thisWsConnection) + ", connect retry on fail timer error: " + ec.message()); that->onError(Event::Type::SUBSCRIPTION_STATUS, Message::Type::GENERIC_ERROR, ec, "timer"); } else { CCAPI_LOGGER_INFO("about to retry"); try { auto thatWsConnectionPtr = that->createWsConnectionPtr(wsConnectionPtr); that->prepareConnect(thatWsConnectionPtr); that->connectNumRetryOnFailByConnectionUrlMap[urlBase] += 1; } catch (const beast::error_code& ec) { CCAPI_LOGGER_TRACE("fail"); that->onError(Event::Type::SUBSCRIPTION_STATUS, Message::Type::SUBSCRIPTION_FAILURE, ec, "create stream", wsConnectionPtr->correlationIdList); return; } } } }); this->connectRetryOnFailTimerByConnectionIdMap[thisWsConnection.id] = timerPtr; } std::shared_ptr createWsConnectionPtr(std::shared_ptr wsConnectionPtr) { std::shared_ptr thatWsConnectionPtr = wsConnectionPtr; std::shared_ptr>> streamPtr(nullptr); try { streamPtr = this->createWsStream(this->serviceContextPtr->ioContextPtr, this->serviceContextPtr->sslContextPtr); } catch (const beast::error_code& ec) { throw ec; } thatWsConnectionPtr->streamPtr = streamPtr; return thatWsConnectionPtr; } virtual void onFail(std::shared_ptr wsConnectionPtr) { CCAPI_LOGGER_FUNCTION_ENTER; this->clearStates(wsConnectionPtr); this->onFail_(wsConnectionPtr); CCAPI_LOGGER_FUNCTION_EXIT; } virtual void clearStates(std::shared_ptr wsConnectionPtr) { WsConnection& wsConnection = *wsConnectionPtr; CCAPI_LOGGER_INFO("clear states for wsConnection " + toString(wsConnection)); this->shouldProcessRemainingMessageOnClosingByConnectionIdMap.erase(wsConnection.id); this->lastPongTpByMethodByConnectionIdMap.erase(wsConnection.id); this->extraPropertyByConnectionIdMap.erase(wsConnection.id); if (this->pingTimerByMethodByConnectionIdMap.find(wsConnection.id) != this->pingTimerByMethodByConnectionIdMap.end()) { for (const auto& x : this->pingTimerByMethodByConnectionIdMap.at(wsConnection.id)) { x.second->cancel(); } this->pingTimerByMethodByConnectionIdMap.erase(wsConnection.id); } if (this->pongTimeOutTimerByMethodByConnectionIdMap.find(wsConnection.id) != this->pongTimeOutTimerByMethodByConnectionIdMap.end()) { for (const auto& x : this->pongTimeOutTimerByMethodByConnectionIdMap.at(wsConnection.id)) { x.second->cancel(); } this->pongTimeOutTimerByMethodByConnectionIdMap.erase(wsConnection.id); } // auto urlBase = UtilString::split(wsConnection.url, "?").at(0); // this->connectNumRetryOnFailByConnectionUrlMap.erase(urlBase); if (this->connectRetryOnFailTimerByConnectionIdMap.find(wsConnection.id) != this->connectRetryOnFailTimerByConnectionIdMap.end()) { this->connectRetryOnFailTimerByConnectionIdMap.at(wsConnection.id)->cancel(); this->connectRetryOnFailTimerByConnectionIdMap.erase(wsConnection.id); } this->readMessageBufferByConnectionIdMap.erase(wsConnection.id); this->writeMessageBufferByConnectionIdMap.erase(wsConnection.id); this->writeMessageBufferWrittenLengthByConnectionIdMap.erase(wsConnection.id); } virtual void onClose(std::shared_ptr wsConnectionPtr, ErrorCode ec) { CCAPI_LOGGER_FUNCTION_ENTER; auto now = UtilTime::now(); WsConnection& wsConnection = *wsConnectionPtr; wsConnection.status = WsConnection::Status::CLOSED; CCAPI_LOGGER_INFO("connection " + toString(wsConnection) + " is closed"); std::stringstream s; s << "close code: " << wsConnectionPtr->remoteCloseCode << " (" << std::to_string(wsConnectionPtr->remoteCloseCode) << "), close reason: " << wsConnectionPtr->remoteCloseReason.reason; std::string reason = s.str(); CCAPI_LOGGER_INFO("reason is " + reason); Event event; event.setType(Event::Type::SESSION_STATUS); Message message; message.setTimeReceived(now); message.setType(Message::Type::SESSION_CONNECTION_DOWN); Element element; element.insert(CCAPI_CONNECTION_ID, wsConnection.id); element.insert(CCAPI_CONNECTION_URL, wsConnection.getUrl()); element.insert(CCAPI_REASON, reason); message.setElementList({element}); std::vector correlationIdList; for (const auto& subscription : wsConnection.subscriptionList) { correlationIdList.push_back(subscription.getCorrelationId()); } CCAPI_LOGGER_DEBUG("correlationIdList = " + toString(correlationIdList)); message.setCorrelationIdList(correlationIdList); event.setMessageList({message}); this->eventHandler(event, nullptr); CCAPI_LOGGER_INFO("connection " + toString(wsConnection) + " is closed"); this->clearStates(wsConnectionPtr); auto thisWsConnectionPtr = this->createWsConnectionPtr(wsConnectionPtr); this->wsConnectionByIdMap.erase(wsConnectionPtr->id); if (this->shouldContinue.load()) { this->prepareConnect(thisWsConnectionPtr); } CCAPI_LOGGER_FUNCTION_EXIT; } void onMessage(std::shared_ptr wsConnectionPtr, const char* data, size_t dataSize) { auto now = UtilTime::now(); WsConnection& wsConnection = *wsConnectionPtr; CCAPI_LOGGER_DEBUG("received a message from connection " + toString(wsConnection)); if (wsConnection.status != WsConnection::Status::OPEN && !this->shouldProcessRemainingMessageOnClosingByConnectionIdMap[wsConnection.id]) { CCAPI_LOGGER_WARN("should not process remaining message on closing"); return; } auto& stream = *wsConnectionPtr->streamPtr; if (stream.got_text()) { boost::beast::string_view textMessage(data, dataSize); CCAPI_LOGGER_DEBUG(std::string("received a text message: ") + std::string(textMessage)); try { this->onTextMessage(wsConnectionPtr, textMessage, now); } catch (const std::exception& e) { CCAPI_LOGGER_ERROR(std::string("textMessage = ") + std::string(textMessage)); this->onError(Event::Type::SUBSCRIPTION_STATUS, Message::Type::GENERIC_ERROR, e); } } else if (stream.got_binary()) { CCAPI_LOGGER_DEBUG(std::string("received a binary message: ") + UtilAlgorithm::stringToHex(std::string(data, dataSize))); #if defined(CCAPI_ENABLE_SERVICE_MARKET_DATA) && \ (defined(CCAPI_ENABLE_EXCHANGE_HUOBI) || defined(CCAPI_ENABLE_EXCHANGE_HUOBI_USDT_SWAP) || defined(CCAPI_ENABLE_EXCHANGE_HUOBI_COIN_SWAP)) || \ defined(CCAPI_ENABLE_SERVICE_EXECUTION_MANAGEMENT) && \ (defined(CCAPI_ENABLE_EXCHANGE_HUOBI_USDT_SWAP) || defined(CCAPI_ENABLE_EXCHANGE_HUOBI_COIN_SWAP) || defined(CCAPI_ENABLE_EXCHANGE_BITMART)) if (this->needDecompressWebsocketMessage) { std::string decompressed; boost::beast::string_view payload(data, dataSize); try { ErrorCode ec = this->inflater.decompress(reinterpret_cast(&payload[0]), payload.size(), decompressed); if (ec) { CCAPI_LOGGER_FATAL(ec.message()); } CCAPI_LOGGER_DEBUG("decompressed = " + decompressed); this->onTextMessage(wsConnectionPtr, decompressed, now); } catch (const std::exception& e) { std::stringstream ss; ss << std::hex << std::setfill('0'); for (int i = 0; i < payload.size(); ++i) { ss << std::setw(2) << static_cast(reinterpret_cast(&payload[0])[i]); } CCAPI_LOGGER_ERROR("binaryMessage = " + ss.str()); this->onError(Event::Type::SUBSCRIPTION_STATUS, Message::Type::GENERIC_ERROR, e); } ErrorCode ec = this->inflater.inflate_reset(); if (ec) { this->onError(Event::Type::SUBSCRIPTION_STATUS, Message::Type::GENERIC_ERROR, ec, "decompress"); } } #endif } } void onControlCallback(std::shared_ptr wsConnectionPtr, boost::beast::websocket::frame_type kind, boost::beast::string_view payload) { if (kind == boost::beast::websocket::frame_type::ping) { this->onPing(wsConnectionPtr, payload); } else if (kind == boost::beast::websocket::frame_type::pong) { this->onPong(wsConnectionPtr, payload); } else if (kind == boost::beast::websocket::frame_type::close) { this->onClose(wsConnectionPtr, ErrorCode()); } } void onPong(std::shared_ptr wsConnectionPtr, boost::beast::string_view payload) { auto now = UtilTime::now(); this->onPongByMethod(PingPongMethod::WEBSOCKET_PROTOCOL_LEVEL, wsConnectionPtr, now, true); } void onPongByMethod(PingPongMethod method, std::shared_ptr wsConnectionPtr, const TimePoint& timeReceived, bool truePong) { CCAPI_LOGGER_FUNCTION_ENTER; CCAPI_LOGGER_TRACE(pingPongMethodToString(method) + ": received a " + (truePong ? "websocket protocol pong" : "data message") + " from " + toString(*wsConnectionPtr)); this->lastPongTpByMethodByConnectionIdMap[wsConnectionPtr->id][method] = timeReceived; CCAPI_LOGGER_FUNCTION_EXIT; } void onPing(std::shared_ptr wsConnectionPtr, boost::beast::string_view payload) { CCAPI_LOGGER_FUNCTION_ENTER; auto now = UtilTime::now(); CCAPI_LOGGER_TRACE("received a ping from " + toString(*wsConnectionPtr)); this->lastPongTpByMethodByConnectionIdMap[wsConnectionPtr->id][PingPongMethod::WEBSOCKET_PROTOCOL_LEVEL] = now; CCAPI_LOGGER_FUNCTION_EXIT; } void send(std::shared_ptr wsConnectionPtr, boost::beast::string_view payload, ErrorCode& ec) { this->writeMessage(wsConnectionPtr, payload.data(), payload.length()); } void ping(std::shared_ptr wsConnectionPtr, boost::beast::string_view payload, ErrorCode& ec) { if (!this->wsConnectionPendingPingingByConnectionIdMap[wsConnectionPtr->id]) { auto& stream = *wsConnectionPtr->streamPtr; stream.async_ping( "", [that = this, wsConnectionPtr](ErrorCode const& ec) { that->wsConnectionPendingPingingByConnectionIdMap[wsConnectionPtr->id] = false; }); this->wsConnectionPendingPingingByConnectionIdMap[wsConnectionPtr->id] = true; } } virtual void pingOnApplicationLevel(std::shared_ptr wsConnectionPtr, ErrorCode& ec) {} void setPingPongTimer(PingPongMethod method, std::shared_ptr wsConnectionPtr, std::function pingMethod) { CCAPI_LOGGER_FUNCTION_ENTER; CCAPI_LOGGER_TRACE("method = " + pingPongMethodToString(method)); auto pingIntervalMilliseconds = this->pingIntervalMillisecondsByMethodMap[method]; auto pongTimeoutMilliseconds = this->pongTimeoutMillisecondsByMethodMap[method]; CCAPI_LOGGER_TRACE("pingIntervalMilliseconds = " + toString(pingIntervalMilliseconds)); CCAPI_LOGGER_TRACE("pongTimeoutMilliseconds = " + toString(pongTimeoutMilliseconds)); if (pingIntervalMilliseconds <= pongTimeoutMilliseconds) { return; } WsConnection& wsConnection = *wsConnectionPtr; if (wsConnection.status == WsConnection::Status::OPEN) { if (this->pingTimerByMethodByConnectionIdMap.find(wsConnection.id) != this->pingTimerByMethodByConnectionIdMap.end() && this->pingTimerByMethodByConnectionIdMap.at(wsConnection.id).find(method) != this->pingTimerByMethodByConnectionIdMap.at(wsConnection.id).end()) { this->pingTimerByMethodByConnectionIdMap.at(wsConnection.id).at(method)->cancel(); } TimerPtr timerPtr( new net::steady_timer(*this->serviceContextPtr->ioContextPtr, std::chrono::milliseconds(pingIntervalMilliseconds - pongTimeoutMilliseconds))); timerPtr->async_wait([wsConnectionPtr, that = shared_from_this(), pingMethod, pongTimeoutMilliseconds, method](ErrorCode const& ec) { WsConnection& wsConnection = *wsConnectionPtr; if (that->wsConnectionByIdMap.find(wsConnection.id) != that->wsConnectionByIdMap.end()) { if (ec) { CCAPI_LOGGER_ERROR("wsConnection = " + toString(wsConnection) + ", ping timer error: " + ec.message()); that->onError(Event::Type::SUBSCRIPTION_STATUS, Message::Type::GENERIC_ERROR, ec, "timer"); } else { if (that->wsConnectionByIdMap.at(wsConnection.id)->status == WsConnection::Status::OPEN) { ErrorCode ec; pingMethod(ec); if (ec) { that->onError(Event::Type::SUBSCRIPTION_STATUS, Message::Type::GENERIC_ERROR, ec, "ping"); } if (pongTimeoutMilliseconds <= 0) { return; } if (that->pongTimeOutTimerByMethodByConnectionIdMap.find(wsConnection.id) != that->pongTimeOutTimerByMethodByConnectionIdMap.end() && that->pongTimeOutTimerByMethodByConnectionIdMap.at(wsConnection.id).find(method) != that->pongTimeOutTimerByMethodByConnectionIdMap.at(wsConnection.id).end()) { that->pongTimeOutTimerByMethodByConnectionIdMap.at(wsConnection.id).at(method)->cancel(); } TimerPtr timerPtr(new net::steady_timer(*that->serviceContextPtr->ioContextPtr, std::chrono::milliseconds(pongTimeoutMilliseconds))); timerPtr->async_wait([wsConnectionPtr, that, pingMethod, pongTimeoutMilliseconds, method](ErrorCode const& ec) { WsConnection& wsConnection = *wsConnectionPtr; if (that->wsConnectionByIdMap.find(wsConnection.id) != that->wsConnectionByIdMap.end()) { if (ec) { CCAPI_LOGGER_ERROR("wsConnection = " + toString(wsConnection) + ", pong time out timer error: " + ec.message()); that->onError(Event::Type::SUBSCRIPTION_STATUS, Message::Type::GENERIC_ERROR, ec, "timer"); } else { if (that->wsConnectionByIdMap.at(wsConnection.id)->status == WsConnection::Status::OPEN) { auto now = UtilTime::now(); if (that->lastPongTpByMethodByConnectionIdMap.find(wsConnection.id) != that->lastPongTpByMethodByConnectionIdMap.end() && that->lastPongTpByMethodByConnectionIdMap.at(wsConnection.id).find(method) != that->lastPongTpByMethodByConnectionIdMap.at(wsConnection.id).end() && std::chrono::duration_cast(now - that->lastPongTpByMethodByConnectionIdMap.at(wsConnection.id).at(method)) .count() >= pongTimeoutMilliseconds) { auto thisWsConnectionPtr = wsConnectionPtr; ErrorCode ec; that->close(thisWsConnectionPtr, beast::websocket::close_code::normal, beast::websocket::close_reason(beast::websocket::close_code::normal, "pong timeout"), ec); if (ec) { that->onError(Event::Type::SUBSCRIPTION_STATUS, Message::Type::GENERIC_ERROR, ec, "shutdown"); } that->shouldProcessRemainingMessageOnClosingByConnectionIdMap[thisWsConnectionPtr->id] = true; } else { auto thisWsConnectionPtr = wsConnectionPtr; that->setPingPongTimer(method, thisWsConnectionPtr, pingMethod); } } } } }); that->pongTimeOutTimerByMethodByConnectionIdMap[wsConnection.id][method] = timerPtr; } } } }); this->pingTimerByMethodByConnectionIdMap[wsConnection.id][method] = timerPtr; } CCAPI_LOGGER_FUNCTION_EXIT; } std::string convertParamTimeSecondsToTimeMilliseconds(const std::string& input) { auto dotPosition = input.find('.'); if (dotPosition == std::string::npos) { return std::to_string(std::stoll(input) * 1000); } else { } return std::to_string(std::stoll(input.substr(0, dotPosition)) * 1000 + std::stoll(UtilString::rightPadTo(input.substr(dotPosition + 1, 3), 3, '0'))); } virtual void onTextMessage(std::shared_ptr wsConnectionPtr, boost::beast::string_view textMessage, const TimePoint& timeReceived) {} #endif bool hostHttpHeaderValueIgnorePort{}; std::string apiKeyName; std::string apiSecretName; std::string exchangeName; std::string baseUrlWs; std::string baseUrlRest; std::function* eventQueue)> eventHandler; SessionOptions sessionOptions; SessionConfigs sessionConfigs; ServiceContextPtr serviceContextPtr; tcp::resolver resolver, resolverWs; std::string hostRest; std::string portRest; std::string hostWs; std::string portWs; // tcp::resolver::results_type tcpResolverResultsRest, tcpResolverResultsWs; std::map>>> httpConnectionPool; std::map credentialDefault; std::map sendRequestDelayTimerByCorrelationIdMap; #ifdef CCAPI_LEGACY_USE_WEBSOCKETPP std::map wsConnectionByIdMap; #else std::map> wsConnectionByIdMap; // TODO(cryptochassis): for consistency, to be renamed to wsConnectionPtrByIdMap std::map readMessageBufferByConnectionIdMap; std::map> writeMessageBufferByConnectionIdMap; std::map writeMessageBufferWrittenLengthByConnectionIdMap; std::map> writeMessageBufferBoundaryByConnectionIdMap; #endif std::map wsConnectionPendingPingingByConnectionIdMap; std::map shouldProcessRemainingMessageOnClosingByConnectionIdMap; std::map connectNumRetryOnFailByConnectionUrlMap; std::map connectRetryOnFailTimerByConnectionIdMap; std::map> lastPongTpByMethodByConnectionIdMap; std::map> pingTimerByMethodByConnectionIdMap; std::map> pongTimeOutTimerByMethodByConnectionIdMap; std::map pingIntervalMillisecondsByMethodMap; std::map pongTimeoutMillisecondsByMethodMap; std::atomic shouldContinue{true}; std::map> extraPropertyByConnectionIdMap; bool enableCheckPingPongWebsocketProtocolLevel{}; bool enableCheckPingPongWebsocketApplicationLevel{}; std::map requestOperationToMessageTypeMap; // std::regex convertNumberToStringInJsonRegex{"(\\[|,|\":)\\s?(-?\\d+\\.?\\d*)"}; // std::string convertNumberToStringInJsonRewrite{"$1\"$2\""}; bool needDecompressWebsocketMessage{}; #if defined(CCAPI_ENABLE_SERVICE_MARKET_DATA) && \ (defined(CCAPI_ENABLE_EXCHANGE_HUOBI) || defined(CCAPI_ENABLE_EXCHANGE_HUOBI_USDT_SWAP) || defined(CCAPI_ENABLE_EXCHANGE_HUOBI_COIN_SWAP)) || \ defined(CCAPI_ENABLE_SERVICE_EXECUTION_MANAGEMENT) && \ (defined(CCAPI_ENABLE_EXCHANGE_HUOBI_USDT_SWAP) || defined(CCAPI_ENABLE_EXCHANGE_HUOBI_COIN_SWAP) || defined(CCAPI_ENABLE_EXCHANGE_BITMART)) #ifdef CCAPI_LEGACY_USE_WEBSOCKETPP struct monostate {}; websocketpp::extensions_workaround::permessage_deflate::enabled inflater; #else InflateStream inflater; #endif #endif }; } /* namespace ccapi */ #endif // INCLUDE_CCAPI_CPP_SERVICE_CCAPI_SERVICE_H_