diff --git a/include/opendht/dht_proxy_client.h b/include/opendht/dht_proxy_client.h index fe0f093a..515a96fa 100644 --- a/include/opendht/dht_proxy_client.h +++ b/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 httpClient_; mutable std::mutex lockCurrentProxyInfos_; diff --git a/include/opendht/dht_proxy_server.h b/include/opendht/dht_proxy_server.h index 1db5eb68..2803d8bf 100644 --- a/include/opendht/dht_proxy_server.h +++ b/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 httpServerThreadPool_; std::unique_ptr> httpServer_; + std::unique_ptr httpClient_; mutable std::mutex statsMutex_; mutable ServerStats stats_; diff --git a/include/opendht/http.h b/include/opendht/http.h index 8621dcf1..da1f217c 100644 --- a/include/opendht/http.h +++ b/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 logger = nullptr); asio::io_context& io_context(); void set_logger(std::shared_ptr 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 logger_; }; diff --git a/src/dht_proxy_client.cpp b/src/dht_proxy_client.cpp index 39fa67a9..9571c5f7 100644 --- a/src/dht_proxy_client.cpp +++ b/src/dht_proxy_client.cpp @@ -67,14 +67,13 @@ DhtProxyClient::DhtProxyClient() {} DhtProxyClient::DhtProxyClient(std::function signal, const std::string& serverHost, const std::string &pushClientId, std::shared_ptr 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(ctx_, serverHostIp_, serverHostPort_, logger); if (serverHost_.find("://") == std::string::npos) serverHost_ = proxy::HTTP_PROTO + serverHost_; @@ -84,8 +83,8 @@ DhtProxyClient::DhtProxyClient(std::function 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 diff --git a/src/dht_proxy_server.cpp b/src/dht_proxy_server.cpp index bf909273..d80d55e0 100644 --- a/src/dht_proxy_server.cpp +++ b/src/dht_proxy_server.cpp @@ -73,6 +73,11 @@ DhtProxyServer::DhtProxyServer(std::shared_ptr dht, in_port_t port, restinio::external_io_context(httpServerThreadPool_->io_context()), std::forward(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_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 diff --git a/src/http.cpp b/src/http.cpp index c7575814..84599f90 100644 --- a/src/http.cpp +++ b/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 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( 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 parser_s, std::shared_ptr 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(); }); }