dhtproxy: refactor client to use external io_context

include/opendht/dht_proxy_client.h

@@ -71,9 +71,9 @@ public:
     virtual ~DhtProxyClient();
 
     /**
-     * Get Asio I/O Context from the http::Client instance.
+     * Get Asio I/O Context.
      */
-    asio::io_context& httpIOContext();
+    asio::io_context& io_context();
 
     /**
      * Get the ID of the node.
@@ -327,6 +327,7 @@ private:
     /*
      * ASIO I/O Context for sockets used in requests
      */
+    asio::io_context ctx_;
     std::unique_ptr<http::Client> httpClient_;
 
     mutable std::mutex lockCurrentProxyInfos_;

include/opendht/dht_proxy_server.h

@@ -40,9 +40,10 @@
 #endif
 
 namespace http {
-    class ConnectionListener;
-    struct ListenerSession;
+    class Client;
     class opendht_logger_t;
+    struct ListenerSession;
+    class ConnectionListener;
 }
 
 namespace restinio {
@@ -323,6 +324,7 @@ private:
 
     std::unique_ptr<IOContextThreadPool> httpServerThreadPool_;
     std::unique_ptr<restinio::http_server_t<RestRouterTraits>> httpServer_;
+    std::unique_ptr<http::Client> httpClient_;
 
     mutable std::mutex statsMutex_;
     mutable ServerStats stats_;

include/opendht/http.h

@@ -85,8 +85,8 @@ private:
 class Client
 {
 public:
-    Client() = default;
-    Client(std::string ip, uint16_t port);
+    Client(asio::io_context &ctx, std::string ip, uint16_t port,
+           std::shared_ptr<dht::Logger> logger = nullptr);
 
     asio::io_context& io_context();
     void set_logger(std::shared_ptr<dht::Logger> logger);
@@ -114,8 +114,7 @@ private:
 
     uint16_t port_;
     asio::ip::address addr_;
-    asio::io_context ctx_;
-    asio::ip::tcp::resolver resolver_ {ctx_};
+    asio::ip::tcp::resolver resolver_;
     uint16_t connId_ {1};
     std::shared_ptr<dht::Logger> logger_;
 };

src/dht_proxy_client.cpp

@@ -67,14 +67,13 @@ DhtProxyClient::DhtProxyClient() {}
 
 DhtProxyClient::DhtProxyClient(std::function<void()> signal, const std::string& serverHost,
     const std::string &pushClientId, std::shared_ptr<dht::Logger> logger):
-        serverHost_(serverHost), pushClientId_(pushClientId), loopSignal_(signal), logger_(logger)
+        serverHost_(serverHost), pushClientId_(pushClientId), loopSignal_(signal),
+        logger_(logger)
 {
     auto hostAndPort = splitPort(serverHost_);
     serverHostIp_ = hostAndPort.first;
     serverHostPort_ = std::atoi(hostAndPort.second.c_str());
-
-    httpClient_->set_query_address(serverHostIp_, serverHostPort_);
-    httpClient_->set_logger(logger);
+    httpClient_ = std::make_unique<http::Client>(ctx_, serverHostIp_, serverHostPort_, logger);
 
     if (serverHost_.find("://") == std::string::npos)
         serverHost_ = proxy::HTTP_PROTO + serverHost_;
@@ -84,8 +83,8 @@ DhtProxyClient::DhtProxyClient(std::function<void()> signal, const std::string&
 }
 
 asio::io_context&
-DhtProxyClient::httpIOContext(){
-    return httpClient_->io_context();
+DhtProxyClient::io_context(){
+    return ctx_;
 }
 
 void
@@ -141,9 +140,8 @@ DhtProxyClient::getLocalById(const InfoHash& k, Value::Id id) const {
 void
 DhtProxyClient::cancelAllOperations()
 {
-    auto &io_context = this->httpIOContext();
-    if (io_context.stopped())
-        io_context.stop();
+    if (!ctx_.stopped())
+        ctx_.stop();
 }
 
 void

src/dht_proxy_server.cpp

@@ -73,6 +73,11 @@ DhtProxyServer::DhtProxyServer(std::shared_ptr<DhtRunner> dht, in_port_t port,
         restinio::external_io_context(httpServerThreadPool_->io_context()),
         std::forward<ServerSettings>(settings)
     ));
+    // build http client
+    auto pushHostPort = splitPort(pushServer_);
+    uint16_t pushPort = std::atoi(pushHostPort.second.c_str());
+    httpClient_.reset(new http::Client(httpServer_->io_context(),
+                                       pushHostPort.first, pushPort, logger_));
     // run http server
     try {
         httpServer_->open_async([]{/*Ok.*/}, [](std::exception_ptr ex){
@@ -623,11 +628,6 @@ DhtProxyServer::sendPushNotification(const std::string& token, Json::Value&& jso
     wbuilder["indentation"] = "";
     auto body = Json::writeString(wbuilder, content);
 
-    auto hostAndPort = splitPort(pushServer_);
-    uint16_t port = std::atoi(hostAndPort.second.c_str());
-    http::Client httpClient {hostAndPort.first, port};
-    httpClient.set_logger(logger_);
-
     auto parser = std::make_shared<http_parser>();
     http_parser_init(parser.get(), HTTP_RESPONSE);
 
@@ -639,11 +639,9 @@ DhtProxyServer::sendPushNotification(const std::string& token, Json::Value&& jso
         std::cerr << "Error in SendPushNotification status_code=" << parser->status_code << std::endl;
         return 1;
     };
-
-    auto request = httpClient.create_request(header, header_fields,
+    auto request = httpClient_->create_request(header, header_fields,
         restinio::http_connection_header_t::close, body);
-    httpClient.post_request(request, parser, parser_s);
-    httpClient.io_context().run();
+    httpClient_->post_request(request, parser, parser_s);
 }
 
 #endif //OPENDHT_PUSH_NOTIFICATIONS

src/http.cpp

@@ -122,13 +122,16 @@ ConnectionListener::to_str(restinio::connection_state::cause_t cause) noexcept
 
 // client
 
-Client::Client(const std::string ip, const uint16_t port){
+Client::Client(asio::io_context &ctx, const std::string ip, const uint16_t port,
+               std::shared_ptr<dht::Logger> logger):
+        resolver_(ctx), logger_(logger)
+{
     set_query_address(ip, port);
 }
 
 asio::io_context&
 Client::io_context(){
-    return ctx_;
+    return resolver_.get_io_context();
 }
 
 void
@@ -155,7 +158,7 @@ Client::open_conn(){
     using namespace asio::ip;
     auto conn = std::make_shared<Connection>(
         connId_,
-        std::move(asio::ip::tcp::socket{ctx_})
+        std::move(asio::ip::tcp::socket{resolver_.get_io_context()})
     );
     logger_->d("[connection:%i] created", conn->id());
     connId_++;
@@ -204,11 +207,8 @@ Client::post_request(std::string request,
                      std::shared_ptr<http_parser_settings> parser_s,
                      std::shared_ptr<Connection> conn){
     // invoke the given handler and return immediately
-    asio::post(ctx_, [this, request, parser, parser_s, conn](){
+    asio::post(resolver_.get_io_context(), [this, request, parser, parser_s, conn](){
         this->async_request(request, parser, parser_s, conn);
-        // execute at most one handler, it ensures that same func call
-        // with different callback gets the priority on the io_context
-        ctx_.run_one();
     });
 }