Merge pull request #547 from crypto-chassis/proxy

add subscription proxyUrl

Author: cryptochassis

README.md

@@ -32,7 +32,6 @@
       - [Multiple subscription fields](#multiple-subscription-fields)
       - [Make Session::sendRequest blocking](#make-sessionsendrequest-blocking)
       - [Provide API credentials for an exchange](#provide-api-credentials-for-an-exchange)
-      - [Override exchange urls](#override-exchange-urls)
       - [Complex request parameters](#complex-request-parameters-1)
       - [Send request by Websocket API](#send-request-by-websocket-api)
       - [Specify instrument type](#specify-instrument-type)
@@ -44,6 +43,8 @@
       - [Set timer](#set-timer)
       - [Heartbeat](#heartbeat)
       - [Use multiple sessions](#use-multiple-sessions)
+      - [Override exchange urls](#override-exchange-urls)
+      - [Connect to a proxy](#connect-to-a-proxy)
   - [Performance Tuning](#performance-tuning)
   - [Known Issues and Workarounds](#known-issues-and-workarounds)
 
@@ -701,11 +702,11 @@ Bye
 
 #### Specify correlation id
 
-Instantiate `Request` with the desired correlationId. The `correlationId` should be unique.
+Instantiate `Request` with the desired `correlationId`. The `correlationId` should be unique.
 ```
 Request request(Request::Operation::CREATE_ORDER, "okx", "BTC-USDT", "cool correlation id");
 ```
-Instantiate `Subscription` with the desired correlationId.
+Instantiate `Subscription` with the desired `correlationId`.
 ```
 Subscription subscription("okx", "BTC-USDT", "ORDER_UPDATE", "", "cool correlation id");
 ```
@@ -768,9 +769,6 @@ Subscription subscription("okx", "BTC-USDT", "ORDER_UPDATE", "", "", {
 });
 ```
 
-#### Override exchange urls
-You can override exchange urls at compile time by using macros. See section "exchange REST urls", "exchange WS urls", and "exchange FIX urls" in [`include/ccapi_cpp/ccapi_macro.h`](include/ccapi_cpp/ccapi_macro.h). You can also override exchange urls at runtime. See [this example](example/src/override_exchange_url_at_runtime/main.cpp). These can be useful if you need to connect to test accounts (e.g. https://www.okx.com/docs-v5/en/#overview-demo-trading-services) or connect to an IP address (e.g. ws://172.30.0.146:9000).
-
 #### Complex request parameters
 Please follow the exchange's API documentations: e.g. https://www.okx.com/docs-v5/en/#order-book-trading-trade-post-place-order.
 ```
@@ -1074,6 +1072,14 @@ Subscription subscription("", "", "HEARTBEAT", "HEARTBEAT_INTERVAL_MILLISECONDS=
 session.subscribe(subscription);
 ```
 
+#### Override exchange urls
+You can override exchange urls at compile time by using macros. See section "exchange REST urls", "exchange WS urls", and "exchange FIX urls" in [`include/ccapi_cpp/ccapi_macro.h`](include/ccapi_cpp/ccapi_macro.h). You can also override exchange urls at runtime. See [this example](example/src/override_exchange_url_at_runtime/main.cpp). These can be useful if you need to connect to test accounts (e.g. https://www.okx.com/docs-v5/en/#overview-demo-trading-services).
+
+#### Connect to a proxy
+Instantiate `Subscription` with the desired `proxyUrl`.
+```
+Subscription subscription("okx", "BTC-USDT", "MARKET_DEPTH", "", "", {}, "172.30.0.146:9000");
+```
 
 
 ## Performance Tuning

include/ccapi_cpp/ccapi_subscription.h

@@ -16,8 +16,8 @@ namespace ccapi {
 class Subscription {
  public:
   explicit Subscription(const std::string& exchange = "", const std::string& instrument = "", const std::string& field = "", const std::string& options = "",
-                        const std::string& correlationId = "", const std::map<std::string, std::string>& credential = {})
-      : exchange(exchange), instrument(instrument), field(field), correlationId(correlationId), credential(credential) {
+                        const std::string& correlationId = "", const std::map<std::string, std::string>& credential = {}, const std::string& proxyUrl = "")
+      : exchange(exchange), instrument(instrument), field(field), correlationId(correlationId), credential(credential), proxyUrl(proxyUrl) {
     auto originalInstrumentSet = UtilString::splitToSet(instrument, ",");
     std::copy_if(originalInstrumentSet.begin(), originalInstrumentSet.end(), std::inserter(this->instrumentSet, this->instrumentSet.end()),
                  [](const std::string& value) { return !value.empty(); });
@@ -68,8 +68,8 @@ class Subscription {
     }
     std::string output = "Subscription [exchange = " + exchange + ", marginType = " + marginType + ", instrumentType = " + instrumentType +
                          ", instrument = " + instrument + ", field = " + field + ", optionMap = " + ccapi::toString(optionMap) +
-                         ", correlationId = " + correlationId + ", credential = " + ccapi::toString(shortCredential) + ", serviceName = " + serviceName +
-                         ", timeSent = " + UtilTime::getISOTimestamp(timeSent) + "]";
+                         ", correlationId = " + correlationId + ", credential = " + ccapi::toString(shortCredential) + ", proxyUrl = " + proxyUrl +
+                         ", serviceName = " + serviceName + ", timeSent = " + UtilTime::getISOTimestamp(timeSent) + "]";
     return output;
   }
 
@@ -89,6 +89,8 @@ class Subscription {
 
   const std::map<std::string, std::string>& getCredential() const { return credential; }
 
+  const std::string& getProxyUrl() const { return proxyUrl; }
+
   const std::string& getServiceName() const { return serviceName; }
 
   const std::set<std::string>& getInstrumentSet() const { return instrumentSet; }
@@ -131,11 +133,14 @@ class Subscription {
 
   void setField(const std::string& field) { this->field = field; }
 
+  void setProxyUrl(const std::string& proxyUrl) { this->proxyUrl = proxyUrl; }
+
   void setTimeSent(TimePoint timeSent) { this->timeSent = timeSent; }
 
   void setInstrumentType(const std::string& instrumentType) { this->instrumentType = instrumentType; }
 
   void setMarginType(const std::string& marginType) { this->marginType = marginType; }
+
   enum class Status {
     UNKNOWN,
     SUBSCRIBING,
@@ -180,6 +185,7 @@ class Subscription {
   std::map<std::string, std::string> optionMap;
   std::string correlationId;
   std::map<std::string, std::string> credential;
+  std::string proxyUrl;
   std::string serviceName;
   std::set<std::string> instrumentSet;
   std::set<std::string> fieldSet;

include/ccapi_cpp/ccapi_ws_connection.h

@@ -18,8 +18,8 @@ class WsConnection {
   WsConnection& operator=(const WsConnection&) = delete;
 
   WsConnection(const std::string& url, const std::string& group, const std::vector<Subscription>& subscriptionList,
-               const std::map<std::string, std::string>& credential)
-      : url(url), group(group), subscriptionList(subscriptionList), credential(credential) {
+               const std::map<std::string, std::string>& credential, const std::string& proxyUrl = "")
+      : url(url), group(group), subscriptionList(subscriptionList), credential(credential), proxyUrl(proxyUrl) {
     std::map<std::string, std::string> shortCredential;
     for (const auto& x : credential) {
       shortCredential.insert(std::make_pair(x.first, UtilString::firstNCharacter(x.second, CCAPI_CREDENTIAL_DISPLAY_LENGTH)));
@@ -51,8 +51,9 @@ class WsConnection {
         streamPtr);
     std::string output = "WsConnection [longId = " + longId + ", id = " + id + ", url = " + url + ", group = " + group +
                          ", subscriptionList = " + ccapi::toString(subscriptionList) + ", credential = " + ccapi::toString(shortCredential) +
-                         ", status = " + statusToString(status) + ", headers = " + ccapi::toString(headers) + ", streamPtr = " + oss.str() +
-                         ", remoteCloseCode = " + std::to_string(remoteCloseCode) + ", remoteCloseReason = " + std::string(remoteCloseReason.reason.c_str()) +
+                         ", proxyUrl = " + proxyUrl + ", status = " + statusToString(status) + ", headers = " + ccapi::toString(headers) +
+                         ", streamPtr = " + oss.str() + ", remoteCloseCode = " + std::to_string(remoteCloseCode) +
+                         ", remoteCloseReason = " + std::string(remoteCloseReason.reason.c_str()) +
                          ", hostHttpHeaderValue = " + ccapi::toString(hostHttpHeaderValue) + ", path = " + ccapi::toString(path) +
                          ", host = " + ccapi::toString(host) + ", port = " + ccapi::toString(port) + ", isSecure = " + ccapi::toString(isSecure) + "]";
     return output;
@@ -145,6 +146,7 @@ class WsConnection {
   Status status{Status::UNKNOWN};
   std::map<std::string, std::string> headers;
   std::map<std::string, std::string> credential;
+  std::string proxyUrl;
   std::variant<std::shared_ptr<beast::websocket::stream<beast::ssl_stream<beast::tcp_stream>>>, std::shared_ptr<beast::websocket::stream<beast::tcp_stream>>>
       streamPtr;
   beast::websocket::close_code remoteCloseCode{};

include/ccapi_cpp/service/ccapi_execution_management_service.h

@@ -61,13 +61,15 @@ class ExecutionManagementService : public Service {
           }
 
           const auto& fieldSet = subscription.getFieldSet();
+          const auto& proxyUrl = subscription.getProxyUrl();
+
           if (fieldSet.find(CCAPI_EM_WEBSOCKET_ORDER_ENTRY) != fieldSet.end()) {
-            auto wsConnectionPtr = std::make_shared<WsConnection>(that->baseUrlWsOrderEntry, "", std::vector<Subscription>{subscription}, credential);
+            auto wsConnectionPtr = std::make_shared<WsConnection>(that->baseUrlWsOrderEntry, "", std::vector<Subscription>{subscription}, credential, proxyUrl);
             that->setWsConnectionStream(wsConnectionPtr);
             CCAPI_LOGGER_WARN("about to subscribe with new wsConnectionPtr " + toString(*wsConnectionPtr));
             that->prepareConnect(wsConnectionPtr);
           } else {
-            auto wsConnectionPtr = std::make_shared<WsConnection>(that->baseUrlWs, "", std::vector<Subscription>{subscription}, credential);
+            auto wsConnectionPtr = std::make_shared<WsConnection>(that->baseUrlWs, "", std::vector<Subscription>{subscription}, credential, proxyUrl);
             that->setWsConnectionStream(wsConnectionPtr);
             CCAPI_LOGGER_WARN("about to subscribe with new wsConnectionPtr " + toString(*wsConnectionPtr));
             that->prepareConnect(wsConnectionPtr);

include/ccapi_cpp/service/ccapi_execution_management_service_ascendex.h

@@ -362,9 +362,10 @@ class ExecutionManagementServiceAscendex : public ExecutionManagementService {
             credential = that->credentialDefault;
           }
           const auto& accountGroup = mapGetWithDefault(credential, that->apiAccountGroupName);
+          const auto& proxyUrl = subscription.getProxyUrl();
 
           auto wsConnectionPtr = std::make_shared<WsConnection>(that->baseUrlWs + "/" + accountGroup + "/api/pro/v1/stream", "",
-                                                                std::vector<Subscription>{subscription}, credential);
+                                                                std::vector<Subscription>{subscription}, credential, proxyUrl);
           that->setWsConnectionStream(wsConnectionPtr);
           CCAPI_LOGGER_WARN("about to subscribe with new wsConnectionPtr " + toString(*wsConnectionPtr));
           that->prepareConnect(wsConnectionPtr);

include/ccapi_cpp/service/ccapi_execution_management_service_gateio_perpetual_futures.h

@@ -82,31 +82,32 @@ class ExecutionManagementServiceGateioPerpetualFutures : public ExecutionManagem
     CCAPI_LOGGER_DEBUG("this->baseUrlWs = " + this->baseUrlWs);
     if (this->shouldContinue.load()) {
       for (auto& subscription : subscriptionList) {
-        boost::asio::post(
-            *this->serviceContextPtr->ioContextPtr, [that = shared_from_base<ExecutionManagementServiceGateioPerpetualFutures>(), subscription]() mutable {
-              auto now = UtilTime::now();
-              subscription.setTimeSent(now);
-              const auto& instrumentSet = subscription.getInstrumentSet();
-              auto it = instrumentSet.begin();
-              if (it != instrumentSet.end()) {
-                std::string settle;
-                std::string symbolId = *it;
-                if (UtilString::endsWith(symbolId, "_USD")) {
-                  settle = "btc";
-                } else if (UtilString::endsWith(symbolId, "_USDT")) {
-                  settle = "usdt";
-                }
-                auto credential = subscription.getCredential();
-                if (credential.empty()) {
-                  credential = that->credentialDefault;
-                }
+        boost::asio::post(*this->serviceContextPtr->ioContextPtr, [that = shared_from_base<ExecutionManagementServiceGateioPerpetualFutures>(),
+                                                                   subscription]() mutable {
+          auto now = UtilTime::now();
+          subscription.setTimeSent(now);
+          const auto& instrumentSet = subscription.getInstrumentSet();
+          auto it = instrumentSet.begin();
+          if (it != instrumentSet.end()) {
+            std::string settle;
+            std::string symbolId = *it;
+            if (UtilString::endsWith(symbolId, "_USD")) {
+              settle = "btc";
+            } else if (UtilString::endsWith(symbolId, "_USDT")) {
+              settle = "usdt";
+            }
+            auto credential = subscription.getCredential();
+            if (credential.empty()) {
+              credential = that->credentialDefault;
+            }
+            const auto& proxyUrl = subscription.getProxyUrl();
 
-                auto wsConnectionPtr = std::make_shared<WsConnection>(that->baseUrlWs + settle, "", std::vector<Subscription>{subscription}, credential);
-                that->setWsConnectionStream(wsConnectionPtr);
-                CCAPI_LOGGER_WARN("about to subscribe with new wsConnectionPtr " + toString(*wsConnectionPtr));
-                that->prepareConnect(wsConnectionPtr);
-              }
-            });
+            auto wsConnectionPtr = std::make_shared<WsConnection>(that->baseUrlWs + settle, "", std::vector<Subscription>{subscription}, credential, proxyUrl);
+            that->setWsConnectionStream(wsConnectionPtr);
+            CCAPI_LOGGER_WARN("about to subscribe with new wsConnectionPtr " + toString(*wsConnectionPtr));
+            that->prepareConnect(wsConnectionPtr);
+          }
+        });
       }
     }
     CCAPI_LOGGER_FUNCTION_EXIT;

include/ccapi_cpp/service/ccapi_market_data_service.h

@@ -59,21 +59,21 @@ class MarketDataService : public Service {
     CCAPI_LOGGER_FUNCTION_ENTER;
     if (this->shouldContinue.load()) {
       for (auto& x : this->groupSubscriptionListByInstrumentGroup(subscriptionList)) {
-        auto instrumentGroup = x.first;
-        auto subscriptionListGivenInstrumentGroup = x.second;
+        const auto& instrumentGroup = x.first;
+        auto& subscriptionListGivenInstrumentGroup = x.second;
         boost::asio::post(*this->serviceContextPtr->ioContextPtr, [that = shared_from_base<MarketDataService>(), instrumentGroup,
                                                                    subscriptionListGivenInstrumentGroup]() mutable {
-          auto now = UtilTime::now();
+          const auto& now = UtilTime::now();
           for (auto& subscription : subscriptionListGivenInstrumentGroup) {
             subscription.setTimeSent(now);
           }
           std::map<std::string, std::vector<std::string>> wsConnectionIdListByInstrumentGroupMap = invertMapMulti(that->instrumentGroupByWsConnectionIdMap);
           if (wsConnectionIdListByInstrumentGroupMap.find(instrumentGroup) != wsConnectionIdListByInstrumentGroupMap.end() &&
               that->subscriptionStatusByInstrumentGroupInstrumentMap.find(instrumentGroup) != that->subscriptionStatusByInstrumentGroupInstrumentMap.end()) {
-            auto wsConnectionId = wsConnectionIdListByInstrumentGroupMap.at(instrumentGroup).at(0);
-            auto wsConnectionPtr = that->wsConnectionPtrByIdMap.at(wsConnectionId);
+            const auto& wsConnectionId = wsConnectionIdListByInstrumentGroupMap.at(instrumentGroup).at(0);
+            const auto& wsConnectionPtr = that->wsConnectionPtrByIdMap.at(wsConnectionId);
             for (const auto& subscription : subscriptionListGivenInstrumentGroup) {
-              auto instrument = subscription.getInstrument();
+              const auto& instrument = subscription.getInstrument();
               if (that->subscriptionStatusByInstrumentGroupInstrumentMap[instrumentGroup].find(instrument) !=
                   that->subscriptionStatusByInstrumentGroupInstrumentMap[instrumentGroup].end()) {
                 that->onError(Event::Type::SUBSCRIPTION_STATUS, Message::Type::SUBSCRIPTION_FAILURE, "already subscribed: " + toString(subscription));
@@ -86,13 +86,15 @@ class MarketDataService : public Service {
             CCAPI_LOGGER_INFO("about to subscribe to exchange");
             that->subscribeToExchange(wsConnectionPtr);
           } else {
-            auto url = UtilString::split(instrumentGroup, "|").at(0);
+            const auto& splittedInstrumentGroup = UtilString::split(instrumentGroup, "|");
+            const auto& url = splittedInstrumentGroup.at(0);
             auto credential = subscriptionListGivenInstrumentGroup.at(0).getCredential();
             if (credential.empty()) {
               credential = that->credentialDefault;
             }
+            const auto& proxyUrl = splittedInstrumentGroup.at(splittedInstrumentGroup.size() - 1);
 
-            auto wsConnectionPtr = std::make_shared<WsConnection>(url, instrumentGroup, subscriptionListGivenInstrumentGroup, credential);
+            auto wsConnectionPtr = std::make_shared<WsConnection>(url, instrumentGroup, subscriptionListGivenInstrumentGroup, credential, proxyUrl);
             that->setWsConnectionStream(wsConnectionPtr);
             CCAPI_LOGGER_WARN("about to subscribe with new wsConnectionPtr " + toString(*wsConnectionPtr));
             that->prepareConnect(wsConnectionPtr);
@@ -122,9 +124,11 @@ class MarketDataService : public Service {
   virtual std::string getInstrumentGroup(const Subscription& subscription) {
     const auto& field = subscription.getField();
     if (field == CCAPI_GENERIC_PUBLIC_SUBSCRIPTION) {
-      return this->baseUrlWs + "|" + subscription.getField() + "|" + subscription.getCorrelationId() + "|" + subscription.getSerializedCredential();
+      return this->baseUrlWs + "|" + subscription.getField() + "|" + subscription.getCorrelationId() + "|" + subscription.getSerializedCredential() + "|" +
+             subscription.getProxyUrl();
     } else {
-      return this->baseUrlWs + "|" + subscription.getField() + "|" + subscription.getSerializedOptions() + "|" + subscription.getSerializedCredential();
+      return this->baseUrlWs + "|" + subscription.getField() + "|" + subscription.getSerializedOptions() + "|" + subscription.getSerializedCredential() + "|" +
+             subscription.getProxyUrl();
     }
   }
 

include/ccapi_cpp/service/ccapi_market_data_service_gateio_perpetual_futures.h

@@ -41,7 +41,7 @@ class MarketDataServiceGateioPerpetualFutures : public MarketDataServiceGateioBa
     } else if (UtilString::endsWith(instrument, "_USDT")) {
       url += "usdt";
     }
-    return url + "|" + subscription.getField() + "|" + subscription.getSerializedOptions();
+    return url + "|" + subscription.getField() + "|" + subscription.getSerializedOptions() + "|" + subscription.getProxyUrl();
   }
 
   void substituteParamSettle(std::string& target, const std::map<std::string, std::string>& param, const std::string& symbolId) {

include/ccapi_cpp/service/ccapi_market_data_service_gemini.h

@@ -216,7 +216,7 @@ class MarketDataServiceGemini : public MarketDataService {
       }
       url += parameter + "=true";
     }
-    return url;
+    return url + "|" + subscription.getProxyUrl();
   }
 
   void convertRequestForRest(http::request<http::string_body>& req, const Request& request, const TimePoint& now, const std::string& symbolId,

include/ccapi_cpp/service/ccapi_market_data_service_huobi_base.h

@@ -102,7 +102,7 @@ class MarketDataServiceHuobiBase : public MarketDataService {
         url += "/ws";
       }
     }
-    return url + "|" + field + "|" + subscription.getSerializedOptions();
+    return url + "|" + field + "|" + subscription.getSerializedOptions() + "|" + subscription.getProxyUrl();
   }
 
   void processTextMessage(std::shared_ptr<WsConnection> wsConnectionPtr, boost::beast::string_view textMessageView, const TimePoint& timeReceived, Event& event,