http: add Resolver

include/opendht/dht_proxy_client.h

@@ -360,7 +360,6 @@ private:
     /**
      * Store listen requests.
      */
-    struct Listener;
     struct ProxySearch;
 
     size_t listenerToken_ {0};

include/opendht/http.h

@@ -28,28 +28,43 @@
 
 namespace http {
 
-using ConnectionId = unsigned int;
+class Connection;
+
+// basic types
+using Id = unsigned int;
+
+// asio handlers callbacks
+using HandlerCb = std::function<void(const asio::error_code& ec)>;
+using ConnectionCb = std::function<void(const asio::error_code& ec, std::shared_ptr<Connection>)>;
 
 class OPENDHT_PUBLIC Connection
 {
 public:
-    Connection(const ConnectionId id, asio::ip::tcp::socket socket);
+    Connection(asio::ip::tcp::socket socket, std::shared_ptr<dht::Logger> logger = {});
     ~Connection();
 
-    ConnectionId id();
+    Id id();
     bool is_open();
     bool is_v6();
+    void timeout(const std::chrono::seconds timeout, HandlerCb cb = {});
+    // TODO abort to silence error in request?
     void close();
 
 private:
     friend class Client;
 
-    ConnectionId id_;
+    Id id_;
+    static Id ids_;
+    std::shared_ptr<dht::Logger> logger_;
+
     asio::ip::tcp::socket socket_;
+    asio::ip::tcp::endpoint endpoint_;
+
     asio::streambuf request_;
     std::string response_body_;
     asio::streambuf response_chunk_;
-    asio::ip::tcp::endpoint endpoint_;
+
+    std::unique_ptr<asio::steady_timer> timeout_timer_;
 };
 
 /**
@@ -63,19 +78,6 @@ struct ListenerSession
     std::shared_ptr<restinio::response_builder_t<restinio::chunked_output_t>> response;
 };
 
-/**
- * Context of an active connection allowing it to parse responses etc.
- */
-struct ConnectionContext
-{
-    // TODO store multiple requests
-    std::string request;
-    std::unique_ptr<http_parser> parser;
-    std::unique_ptr<http_parser_settings> parser_settings;
-    std::shared_ptr<Connection> connection;
-    std::unique_ptr<asio::steady_timer> timeout_timer;
-};
-
 class ConnectionListener
 {
 public:
@@ -98,74 +100,74 @@ private:
     std::shared_ptr<std::mutex> lock_;
     std::shared_ptr<std::map<restinio::connection_id_t,
                              http::ListenerSession>> listeners_;
+
     std::shared_ptr<dht::Logger> logger_;
 };
 
-class OPENDHT_PUBLIC Client
+
+class Resolver {
+    Resolver(asio::io_context& ctx, const std::string& host, const std::string& service = "80");
+
+    using ResolvedCb = std::function<void(const asio::error_code&, const asio::ip::basic_resolver_results<asio::ip::tcp>&)>;
+    void addOnResolved(ResolvedCb cb);
+
+private:
+    std::mutex cbsMutex_;
+    asio::ip::tcp::resolver resolver_;
+    asio::error_code error_;
+    asio::ip::basic_resolver_results<asio::ip::tcp> endpoints_;
+    std::queue<ResolvedCb> cbs_;
+    bool completed_ {false};
+};
+
+class OPENDHT_PUBLIC Request
 {
 public:
-    using HandlerCb = std::function<void(const asio::error_code& ec)>;
-    using ConnectionCb = std::function<void(std::shared_ptr<Connection>)>;
-
-    Client(asio::io_context& ctx, const std::string host, const std::string service = "80",
-           std::shared_ptr<dht::Logger> logger = {}, const bool resolve = true);
-
-    asio::io_context& io_context();
+    Request(asio::io_context& ctx, const std::string host, const std::string service = "80",
+            const bool resolve = true, std::shared_ptr<dht::Logger> logger = {});
+    ~Request();
 
     void set_logger(std::shared_ptr<dht::Logger> logger);
 
-    bool active_connection(const ConnectionId conn_id);
-    void close_connection(const ConnectionId conn_id);
-    void set_connection_timeout(const ConnectionId conn_id, const std::chrono::seconds timeout,
-                                HandlerCb cb = {});
+    void terminate();
 
+    void resolve(const std::string host, const std::string service, HandlerCb cb = {});
     bool resolved();
-    void async_resolve(const std::string host, const std::string service, HandlerCb cb = {});
 
-    void async_connect(ConnectionCb cb);
+    void connect(ConnectionCb cb = {});
 
-    std::string create_request(const restinio::http_request_header_t header,
-                               const restinio::http_header_fields_t header_fields,
-                               const restinio::http_connection_header_t connection,
-                               const std::string body);
+    std::string build(const restinio::http_request_header_t header,
+                      const restinio::http_header_fields_t header_fields,
+                      const restinio::http_connection_header_t connection, const std::string body);
 
-    void async_request(std::shared_ptr<http::Connection> conn,
-                       std::string request, std::unique_ptr<http_parser> parser,
-                       std::unique_ptr<http_parser_settings> parser_s, HandlerCb cb = {});
+    void send(std::string request, std::unique_ptr<http_parser> parser,
+              std::unique_ptr<http_parser_settings> parser_s, HandlerCb cb = {});
 
 private:
-    std::shared_ptr<Connection> create_connection();
+    void handle_request(const asio::error_code& ec, HandlerCb cb = {});
 
-    void handle_resolve(const asio::error_code& ec,
-                        asio::ip::tcp::resolver::iterator endpoint_it,
-                        std::shared_ptr<Connection> conn = {}, HandlerCb cb = {});
+    void handle_response_header(const asio::error_code& ec, const size_t bytes, HandlerCb cb = {});
 
-    void handle_request(const asio::error_code& ec,
-                        std::shared_ptr<Connection> conn = {}, HandlerCb cb = {});
+    void handle_response_body(const asio::error_code& ec, const size_t bytes, HandlerCb cb = {});
 
-    void handle_response_header(const asio::error_code& ec, const size_t bytes,
-                                std::shared_ptr<Connection> conn = {}, HandlerCb cb = {});
+    void parse_request(const std::string request);
 
-    void handle_response_body(const asio::error_code& ec, const size_t bytes,
-                              std::shared_ptr<Connection> conn = {}, HandlerCb cb = {});
+    size_t get_content_length(const std::string str);
 
-    void parse_request(const std::string request, const ConnectionId conn_id = 0);
-
-    size_t get_content_length(const std::string str, const ConnectionId conn_id = 0);
-
-    std::string service_;
+    Resolver resolver_;
+    /* std::string service_;
     std::string host_;
 
-    // contains the io_context
     asio::ip::tcp::resolver resolver_;
-    // resolved endpoint
     asio::ip::basic_resolver_results<asio::ip::tcp> endpoints_;
+*/
+    Id id_;
+    static Id ids_;
+    std::shared_ptr<Connection> conn_;
 
-    ConnectionId connId_ {1};
-    /*
-     * An association between an active connection and its context.
-     */
-    std::map<ConnectionId, ConnectionContext> connections_;
+    std::string request_;
+    std::unique_ptr<http_parser> parser_;
+    std::unique_ptr<http_parser_settings> parser_s_;
 
     std::shared_ptr<dht::Logger> logger_;
 };

src/dht_proxy_client.cpp

@@ -297,7 +297,7 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb,
         if (logger_)
             context->logger = logger_;
         std::weak_ptr<GetContext> wcontext = context;
-        context->cb = [this, key, wcontext, cb](const dht::Sp<dht::Value> value) -> bool {
+        context->cb = [this, wcontext, cb](const dht::Sp<dht::Value> value) -> bool {
             auto context = wcontext.lock();
             {
                 std::lock_guard<std::mutex> lock(lockCallbacks_);
@@ -310,7 +310,7 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb,
             loopSignal_();
             return context->ok;
         };
-        context->donecb = [this, key, donecb](bool ok){
+        context->donecb = [this, donecb](bool ok){
             {
                 std::lock_guard<std::mutex> lock(lockCallbacks_);
                 callbacks_.emplace_back([=](){
@@ -663,8 +663,7 @@ DhtProxyClient::handleProxyStatus(const asio::error_code& ec, std::shared_ptr<In
                 if (logger_)
                     context->logger = logger_;
                 std::weak_ptr<StatusContext> wcontext = context;
-                context->cb = [this, wcontext](Json::Value infos)
-                {
+                context->cb = [wcontext](Json::Value infos) {
                     auto context = wcontext.lock();
                     if (context->family == AF_INET)
                         context->infoState->ipv4++;

src/http.cpp

@@ -24,8 +24,8 @@ constexpr char HTTP_HEADER_CONTENT_LENGTH[] = "Content-Length:";
 
 // connection
 
-Connection::Connection(const ConnectionId id, asio::ip::tcp::socket socket):
-    id_(id), socket_(std::move(socket))
+Connection::Connection(asio::ip::tcp::socket socket, std::shared_ptr<dht::Logger> logger):
+    id_(++ids_), socket_(std::move(socket)), logger_(logger)
 {}
 
 Connection::~Connection(){
@@ -47,6 +47,29 @@ Connection::is_v6(){
     return endpoint_.address().is_v6();
 }
 
+void
+Connection::timeout(const std::chrono::seconds timeout, HandlerCb cb)
+{
+    if (!is_open()){
+        if (logger_)
+            logger_->e("[connection:%i] closed, can't timeout", id_);
+        return;
+    }
+    if (!timeout_timer_)
+        timeout_timer_ = std::make_unique<asio::steady_timer>(io_context());
+    timeout_timer_->expires_at(std::chrono::steady_clock::now() + timeout);
+    timeout_timer_->async_wait([this, cb](const asio::error_code &ec){
+        if (ec == asio::error_code::operation_aborted)
+            return;
+        else if (ec){
+            if (logger_)
+                logger_->e("[connection:%i] timeout error: %s", id_, ec.message().c_str());
+        }
+        if (cb)
+            cb(ec);
+    });
+}
+
 void
 Connection::close(){
     socket_.close();
@@ -64,7 +87,8 @@ ConnectionListener::ConnectionListener(std::shared_ptr<dht::DhtRunner> dht,
 {}
 
 ConnectionListener::~ConnectionListener()
-{}
+{
+}
 
 void
 ConnectionListener::state_changed(const restinio::connection_state::notice_t& notice) noexcept
@@ -107,109 +131,64 @@ ConnectionListener::to_str(restinio::connection_state::cause_t cause) noexcept
     return result;
 }
 
-// client
+// request
 
-Client::Client(asio::io_context& ctx, const std::string host, const std::string service,
-               std::shared_ptr<dht::Logger> logger, const bool resolve):
-        resolver_(ctx), logger_(logger)
+Request::Request(asio::io_context& ctx, const std::string host, const std::string service,
+                 const bool resolve, std::shared_ptr<dht::Logger> logger):
+        id_(++ids_), resolver_(ctx), logger_(logger)
 {
     if (resolve)
-        async_resolve(host, service);
+        resolve(host, service);
 }
 
-asio::io_context&
-Client::io_context()
+Request::~Request()
 {
-    return resolver_.get_io_context();
+    terminate();
 }
 
 void
-Client::set_logger(std::shared_ptr<dht::Logger> logger)
+Request::set_logger(std::shared_ptr<dht::Logger> logger)
 {
     logger_ = logger;
 }
 
-bool
-Client::active_connection(const ConnectionId conn_id)
-{
-    auto conn_context = connections_.find(conn_id);
-    if (conn_context == connections_.end())
-        return false;
-    return conn_context->second.connection->is_open();
-}
-
 void
-Client::close_connection(const ConnectionId conn_id)
+Request::terminate()
 {
-    auto& conn_context = connections_[conn_id];
-    if (conn_context.parser)
-        // ensure on_message_complete is fired
-        http_parser_execute(conn_context.parser.get(),
-                            conn_context.parser_settings.get(), "", 0);
-    if (conn_context.connection){
-        // close the socket
-        if (conn_context.connection->is_open())
-            conn_context.connection->close();
-    }
-    // remove from active requests
-    connections_.erase(conn_id);
+    if (parser_)
+        // ensure on_message_complete callback is called
+        http_parser_execute(parser_.get(), parser_s_.get(), "", 0);
     if (logger_)
-        logger_->d("[http::client] [connection:%i] closed", conn_id);
-}
-
-void
-Client::set_connection_timeout(const ConnectionId conn_id,
-                               const std::chrono::seconds timeout, HandlerCb cb)
-{
-    auto& conn_context = connections_[conn_id];
-    if (logger_ && !conn_context.connection){
-        logger_->e("[http::client] [connection:%i] closed, can't timeout", conn_id);
-        return;
-    }
-    if (!conn_context.timeout_timer)
-        conn_context.timeout_timer = std::make_unique<asio::steady_timer>(io_context());
-    // define or overwrites existing
-    conn_context.timeout_timer->expires_at(std::chrono::steady_clock::now() + timeout);
-    // define timeout
-    conn_context.timeout_timer->async_wait([this, conn_id, cb](const asio::error_code &ec){
-        if (ec){
-            if (logger_)
-                logger_->e("[http::client] [connection:%i] timeout error: %s",
-                           conn_id, ec.message().c_str());
-        }
-        if (cb)
-            cb(ec);
-    });
+        logger_->d("[http:request:%i] done", id_);
 }
 
 bool
-Client::resolved()
+Request::resolved()
 {
     return !endpoints_.empty();
 }
 
 void
-Client::async_resolve(const std::string host, const std::string service, HandlerCb cb)
+Request::resolve(const std::string host, const std::string service, HandlerCb cb)
 {
     // build the query
     asio::ip::tcp::resolver::query query(host, service);
 
     // resolve the query to the server
-    resolver_.async_resolve(query, [this, host, service, cb](
-            const asio::error_code& ec,
-            asio::ip::tcp::resolver::results_type endpoints)
+    resolver_.async_resolve(query, [this, host, service, cb]
+        (const asio::error_code& ec, asio::ip::tcp::resolver::results_type endpoints)
     {
-        if (ec and logger_){
-            logger_->e("[http::client] [resolve %s:%s] error resolving: %s",
-                       host.c_str(), service.c_str(), ec.message().c_str());
+        if (ec){
+            if (logger_)
+                logger_->e("[http:request:%i] error resolving %s:%s: %s",
+                           id_, host.c_str(), service.c_str(), ec.message().c_str());
         }
         else {
             for (auto it = endpoints.begin(); it != endpoints.end(); ++it){
                 if (logger_){
                     asio::ip::tcp::endpoint endpoint = *it;
-                    logger_->d("[http::client] [resolve %s:%s] address=%s ipv%i",
-                        host.c_str(), service.c_str(),
-                        endpoint.address().to_string().c_str(),
+                    logger_->d("[http:request:%i] resolved %s:%s: address=%s ipv%i",
+                        id_, host.c_str(), service.c_str(), endpoint.address().to_string().c_str(),
                         endpoint.address().is_v6() ? 6 : 4);
                 }
             }
@@ -221,10 +200,9 @@ Client::async_resolve(const std::string host, const std::string service, Handler
 }
 
 std::string
-Client::create_request(const restinio::http_request_header_t header,
-                       const restinio::http_header_fields_t header_fields,
-                       const restinio::http_connection_header_t connection,
-                       const std::string body)
+Request::build(const restinio::http_request_header_t header,
+               const restinio::http_header_fields_t header_fields,
+               const restinio::http_connection_header_t connection, const std::string body)
 {
     std::stringstream request;
 
@@ -263,160 +241,147 @@ Client::create_request(const restinio::http_request_header_t header,
 }
 
 void
-Client::async_connect(ConnectionCb cb)
+Request::connect(ConnectionCb cb)
 {
     if (endpoints_.empty()){
         if (logger_)
-            logger_->e("[http::client] host not resolved, can't send request");
+            logger_->e("[http:request] [connect] host not resolved");
         if (cb)
-            cb(nullptr);
+            cb(asio::error::connection_aborted, nullptr);
         return;
     }
-    auto conn = std::make_shared<Connection>(connId_++,
-        std::move(asio::ip::tcp::socket{resolver_.get_io_context()}));
-    if (logger_)
-        logger_->d("[http::client] [connection:%i] created", conn->id());
+    auto socket = asio::ip::tcp::socket{resolver_.get_io_context()};
+    conn_ = std::make_shared<Connection>(std::move(socket));
 
     // try to connect to any until one works
-    asio::async_connect(conn->socket_, endpoints_, [this, conn, cb](const asio::error_code& ec,
-                                                                    const asio::ip::tcp::endpoint& endpoint){
-        // only reacts if all endpoints fail
-        if (ec){
+    asio::async_connect(conn_->socket_, endpoints_, [this, conn, cb]
+                       (const asio::error_code& ec, const asio::ip::tcp::endpoint& endpoint){
+        if (ec){ // all endpoints failed
             if (logger_)
-                logger_->e("[http::client] [connection] failed to connect to any endpoints");
-            close_connection(conn->id());
-            if (cb)
-                cb(nullptr);
-            return;
+                logger_->e("[http:request:%i] [connect] failed to connect to any endpoints", id_);
+        }
+        else {
+            if (logger_)
+                logger_->d("[http:request:%i] [connect] success", id_);
+            // save the associated endpoint
+            conn_->endpoint_ = endpoint;
         }
-        // save the associated endpoint
-        conn->endpoint_ = endpoint;
-        // make the connection context and save it
-        connections_[conn->id()].connection = conn;
-        // get back to user
         if (cb)
-            cb(conn);
+            cb(ec, conn_);
     });
 }
 
 void
-Client::async_request(std::shared_ptr<Connection> conn, std::string request,
-                      std::unique_ptr<http_parser> parser,
-                      std::unique_ptr<http_parser_settings> parser_s, HandlerCb cb)
+Request::send(std::string request, std::unique_ptr<http_parser> parser,
+              std::unique_ptr<http_parser_settings> parser_s, HandlerCb cb)
 {
     if (endpoints_.empty()){
         if (logger_)
-            logger_->e("[http::client] host not resolved, can't send request");
+            logger_->e("[http:request:%i] [send] unresolved destination", id_);
         if (cb)
             cb(asio::error::addrinfo_errors::service_not_found);
-        return;
     }
-    if (!conn){
+    if (!conn_){
         if (logger_)
-            logger_->e("[http::client] invalid connection, can't post request:\n%s", request.c_str());
+            logger_->e("[http:request:%i] [send] uninitialized connection", id_);
         if (cb)
             cb(asio::error::not_connected);
-        return;
     }
-    else if (!conn->is_open()){
+    else if (!conn_->is_open()){
         if (logger_)
-            logger_->e("[http::client] closed connection, can't post request");
-        close_connection(conn->id());
+            logger_->e("[http:request] [send] closed connection");
         if (cb)
             cb(asio::error::not_connected);
-        return;
     }
-    // save the request context
-    auto& conn_context = connections_[conn->id()];
-    conn_context.request = request;
-    conn_context.parser = std::move(parser);
-    conn_context.parser_settings = std::move(parser_s);
+    else {
+        // save the request context
+        request_ = request;
+        parser_ = std::move(parser);
+        parser_s_ = std::move(parser_s);
 
-    // write the request to buffer
-    std::ostream request_stream(&conn->request_);
-    request_stream << request;
+        // write the request to buffer
+        std::ostream request_stream(&conn_->request_);
+        request_stream << request;
 
-    // send the request
-    asio::async_write(conn->socket_, conn->request_,
-        std::bind(&Client::handle_request, this, std::placeholders::_1, conn, cb));
+        // send the request
+        asio::async_write(conn_->socket_, conn_->request_,
+            std::bind(&Request::handle_request, this, std::placeholders::_1, cb));
+    }
 }
 
 void
-Client::handle_request(const asio::error_code& ec, std::shared_ptr<Connection> conn, HandlerCb cb)
+Request::handle_request(const asio::error_code& ec, HandlerCb cb)
 {
-    if (!conn->is_open()){
+    if (!conn_->is_open()){
         if (logger_)
-            logger_->e("[http::client] closed connection, can't handle request");
-        close_connection(conn->id());
+            logger_->e("[http:request:%i] [write] closed connection", id_);
+        terminate();
         if (cb)
             cb(asio::error::not_connected);
         return;
     }
     if (ec and ec != asio::error::eof){
         if (logger_)
-            logger_->e("[http::client] [connection:%i] error handling request: %s",
-                       conn->id(), ec.message().c_str());
-        close_connection(conn->id());
+            logger_->e("[http:request:%i] [write] error: %s", id_, ec.message().c_str());
+        terminate();
         if (cb)
             cb(ec);
         return;
     }
     if (logger_)
-        logger_->d("[http::client] [connection:%i] request write", conn->id());
+        logger_->d("[http:request:%i] [write] success", id_);
 
     // read response
-    asio::async_read_until(conn->socket_, conn->response_chunk_, "\r\n\r\n",
-        std::bind(&Client::handle_response_header, this,
-            std::placeholders::_1, std::placeholders::_2, conn, cb));
+    asio::async_read_until(conn_->socket_, conn_->response_chunk_, "\r\n\r\n",
+        std::bind(&Request::handle_response_header, this,
+                  std::placeholders::_1, std::placeholders::_2, cb));
 }
 
 void
-Client::handle_response_header(const asio::error_code& ec, const size_t bytes,
-                               std::shared_ptr<Connection> conn, HandlerCb cb)
+Request::handle_response_header(const asio::error_code& ec, const size_t bytes, HandlerCb cb)
 {
-    if (!conn->is_open()){
+    if (!conn_->is_open()){
         if (logger_)
-            logger_->e("[http::client] closed connection, can't handle response header");
-        close_connection(conn->id());
+            logger_->e("[http:request:%i] [read:header] closed connection", id_);
+        terminate();
         if (cb)
             cb(asio::error::not_connected);
         return;
     }
     if (ec && ec != asio::error::eof){
         if (logger_)
-            logger_->e("[http::client] [connection:%i] error handling response header: %s",
-                       conn->id(), ec.message().c_str());
-        close_connection(conn->id());
+            logger_->e("[http:request:%i] [read:header] error: %s", id_, ec.message().c_str());
+        terminate();
         if (cb)
             cb(ec);
         return;
     }
     else if ((ec == asio::error::eof) || (ec == asio::error::connection_reset)){
-        close_connection(conn->id());
+        terminate();
         if (cb)
             cb(ec);
         return;
     }
     // read the response buffer
     std::ostringstream str_s;
-    str_s << &conn->response_chunk_;
+    str_s << &conn_->response_chunk_;
     auto chunk = str_s.str();
     auto header = chunk.substr(0, bytes);
     if (logger_)
-        logger_->d("[http::client] [connection:%i] response header read:\n%s", conn->id(), header.c_str());
+        logger_->d("[http:request:%i] [read:header]:\n%s", id_, header.c_str());
     // parse the header right away
-    parse_request(header, conn->id());
+    parse_request(header, conn_->id());
 
     unsigned int content_length = get_content_length(header);
     // has body size
     if (content_length){
         // append current body chunk
         auto body_chunk = chunk.substr(bytes, std::string::npos);
-        conn->response_body_.append(body_chunk);
+        conn_->response_body_.append(body_chunk);
         // read the rest of body
-        asio::async_read(conn->socket_, conn->response_chunk_,
+        asio::async_read(conn_->socket_, conn_->response_chunk_,
             asio::transfer_exactly(content_length - (body_chunk.size())),
-            std::bind(&Client::handle_response_body, this,
+            std::bind(&Request::handle_response_body, this,
                 std::placeholders::_1, content_length, conn, cb));
         return;
     }
@@ -424,80 +389,76 @@ Client::handle_response_header(const asio::error_code& ec, const size_t bytes,
     else if (header.find("Content-Type: application/json") != std::string::npos){
         auto body_chunk = chunk.substr(bytes, std::string::npos);
         if (!body_chunk.empty()){
-            parse_request(body_chunk, conn->id());
+            parse_request(body_chunk, conn_->id());
         }
         // keep reading
-        asio::async_read(conn->socket_, conn->response_chunk_, asio::transfer_at_least(1),
-            std::bind(&Client::handle_response_body, this,
+        asio::async_read(conn_->socket_, conn_->response_chunk_, asio::transfer_at_least(1),
+            std::bind(&Request::handle_response_body, this,
                 std::placeholders::_1, std::placeholders::_2, conn, cb));
     }
 }
 
 void
-Client::handle_response_body(const asio::error_code& ec, const size_t bytes,
+Request::handle_response_body(const asio::error_code& ec, const size_t bytes,
                              std::shared_ptr<Connection> conn, HandlerCb cb)
 {
-    if (!conn->is_open()){
+    if (!conn_->is_open()){
         if (logger_)
-            logger_->e("[http::client] closed connection, can't handle response body");
-        close_connection(conn->id());
+            logger_->e("[http:request] [read:body] closed connection");
+        terminate();
         if (cb)
             cb(asio::error::not_connected);
         return;
     }
     if (ec && ec != asio::error::eof){
         if (logger_)
-            logger_->e("[http::client] [connection:%i] error handling response body: %s",
-                       conn->id(), ec.message().c_str());
-        close_connection(conn->id());
+            logger_->e("[http:request:%i] [read:body] error: %s", id_, ec.message().c_str());
+        terminate();
         if (cb)
             cb(ec);
         return;
     }
     else if ((ec == asio::error::eof) || (ec == asio::error::connection_reset)){
-        close_connection(conn->id());
+        terminate();
         if (cb)
             cb(ec);
         return;
     }
     std::string body;
-    if (!conn->response_body_.empty()){
+    if (!conn_->response_body_.empty()){
         // append previous incomplete chunk from header
-        body.append(conn->response_body_);
+        body.append(conn_->response_body_);
     }
     // read the response buffer
     std::ostringstream str_s;
-    str_s << &conn->response_chunk_;
+    str_s << &conn_->response_chunk_;
     body.append(str_s.str().substr(0, bytes));
     if (logger_)
-        logger_->d("[http::client] [connection:%i] response body read:\n%s", conn->id(), body.c_str());
-    // parse the body
-    parse_request(body, conn->id());
-    // clear the read body
-    conn->response_body_.clear();
-    // keep reading line by line
-    asio::async_read(conn->socket_, conn->response_chunk_, asio::transfer_at_least(1),
-        std::bind(&Client::handle_response_body, this,
+        logger_->d("[http:request:%i] [read:body] success:\n%s", id_, body.c_str());
+
+    parse_request(body);
+    conn_->response_body_.clear();
+
+    asio::async_read(conn_->socket_, conn_->response_chunk_, asio::transfer_at_least(1),
+        std::bind(&Request::handle_response_body, this,
             std::placeholders::_1, std::placeholders::_2, conn, cb));
 }
 
 void
-Client::parse_request(const std::string request, const ConnectionId conn_id)
+Request::parse_request(const std::string request)
 {
-    auto& conn_context = connections_[conn_id];
-    http_parser_execute(conn_context.parser.get(), conn_context.parser_settings.get(),
-                        request.c_str(), request.size());
+    http_parser_execute(parser_.get(), parser_s_.get(), request.c_str(), request.size());
     // detect parsing errors
-    if (HPE_OK != conn_context.parser->http_errno && HPE_PAUSED != conn_context.parser->http_errno){
+    if (HPE_OK != parser_->http_errno && HPE_PAUSED != parser_->http_errno){
         if (logger_){
-            auto err = HTTP_PARSER_ERRNO(conn_context.parser.get());
-            logger_->e("[http::client] [connection:%i] error parsing: %s", conn_id, http_errno_name(err));
+            auto err = HTTP_PARSER_ERRNO(parser_.get());
+            logger_->e("[http:request:%i] [parse] error: %s", id_, http_errno_name(err));
         }
     }
 }
 
 size_t
-Client::get_content_length(const std::string str, const ConnectionId conn_id)
+Request::get_content_length(const std::string str)
 {
     size_t content_length = 0;
     auto content_length_i = str.find(HTTP_HEADER_CONTENT_LENGTH);
@@ -509,11 +470,50 @@ Client::get_content_length(const std::string str, const ConnectionId conn_id)
         }
         catch (const std::exception& e){
             if (logger_)
-                logger_->d("[http::client] [connection:%i] invalid content-length '%s': %s",
-                           conn_id, content_length_str.c_str(), e.what());
+                logger_->d("[http:request:%i] [content-length] invalid '%s': %s",
+                           id_, content_length_str.c_str(), e.what());
         }
     }
     return content_length;
 }
 
+
+Resolver::Resolver(asio::io_context& ctx, const std::string& host, const std::string& service)
+{
+    resolver_.async_resolve({host, service}, [this]
+        (const asio::error_code& ec, asio::ip::tcp::resolver::results_type endpoints)
+    {
+        std::lock_guard<std::mutex> lock(cbsMutex_);
+        while (not cbs_.empty()) {
+            cbs_.front()(ec, endpoints);
+            cbs_.pop();
+        }
+        error_ = ec;
+        endpoints_ = endpoints;
+        completed_ = true;
+    }
+}
+
+void
+Resolver::addOnResolved(ResolvedCb cb)
+{
+    std::lock_guard<std::mutex> lock(cbsMutex_);
+    if (completed_) {
+        cb(error_, endpoints_);
+    } else {
+        cbs_.push(std::move(cb));
+    }
+}
+
+/*
+private:
+    std::mutex cbsMutex_;
+    std::queue<ResolvedCb> cbs_;
+
+    asio::ip::tcp::resolver resolver_;
+    asio::error_code error_;
+    asio::ip::basic_resolver_results<asio::ip::tcp> endpoints_;
+} */
+
+
 } // namespace http

tools/dhtnode.cpp

@@ -578,15 +578,19 @@ main(int argc, char **argv)
     std::mutex m;
     bool done {false};
 
-    node->shutdown([&]()
-    {
-        done = true;
+    node->shutdown([&]() {
+        {
+            std::lock_guard<std::mutex> lk(m);
+            done = true;
+        }
         cv.notify_all();
     });
 
     // wait for shutdown
-    std::unique_lock<std::mutex> lk(m);
-    cv.wait(lk, [&](){ return done; });
+    {
+        std::unique_lock<std::mutex> lk(m);
+        cv.wait(lk, [&](){ return done; });
+    }
 
     node->join();
 #ifdef WIN32_NATIVE