From 5812b24d569e893ed7277da09477dc712913ad6b Mon Sep 17 00:00:00 2001 From: Seva Date: Wed, 17 Jul 2019 11:27:51 -0400 Subject: [PATCH] http: transform client into request --- include/opendht/dht_proxy_client.h | 25 +- include/opendht/dht_proxy_server.h | 10 +- include/opendht/http.h | 207 +++++--- src/dht_proxy_client.cpp | 732 ++++++++++++--------------- src/dht_proxy_server.cpp | 117 ++--- src/http.cpp | 772 ++++++++++++++++++----------- 6 files changed, 1016 insertions(+), 847 deletions(-) diff --git a/include/opendht/dht_proxy_client.h b/include/opendht/dht_proxy_client.h index f5ac8348..ddd2343c 100644 --- a/include/opendht/dht_proxy_client.h +++ b/include/opendht/dht_proxy_client.h @@ -42,7 +42,8 @@ namespace Json { } namespace http { - class Client; + class Resolver; + class Request; } namespace dht { @@ -56,7 +57,7 @@ public: const std::string& pushClientId = "", std::shared_ptr logger = {}); - restinio::http_header_fields_t initHeaderFields(); + void setHeaderFields(std::shared_ptr request); virtual void setPushNotificationToken(const std::string& token) { #ifdef OPENDHT_PUSH_NOTIFICATIONS @@ -294,7 +295,7 @@ private: void handleExpireListener(const asio::error_code &ec, const InfoHash& key); struct Listener; - struct ListenState; + struct OperationState; enum class ListenMethod { LISTEN, SUBSCRIBE, @@ -303,15 +304,13 @@ private: /** * Send Listen with httpClient_ */ - void sendListen(const restinio::http_request_header_t header, - const ValueCallback &cb, - const Sp &state, Listener &listener, - ListenMethod method = ListenMethod::LISTEN); - void handleResubscribe(const asio::error_code &ec, const InfoHash& key, - const size_t token, std::shared_ptr state); + void sendListen(const restinio::http_request_header_t header, const ValueCallback& cb, + const Sp& state, Listener& listener, ListenMethod method = ListenMethod::LISTEN); + void handleResubscribe(const asio::error_code& ec, const InfoHash& key, + const size_t token, std::shared_ptr state); void doPut(const InfoHash&, Sp, DoneCallback, time_point created, bool permanent); - void handleRefreshPut(const asio::error_code &ec, const InfoHash& key, const Value::Id id, + void handleRefreshPut(const asio::error_code& ec, const InfoHash& key, const Value::Id id, std::shared_ptr ok); /** @@ -335,10 +334,8 @@ private: * Note: Each context is used in one thread only */ asio::io_context httpContext_; - /* - * http::Client instance used on http io_context - */ - std::unique_ptr httpClient_; + std::shared_ptr resolver_; + std::map> requests_; /* * Thread for executing the http io_context.run() blocking call */ diff --git a/include/opendht/dht_proxy_server.h b/include/opendht/dht_proxy_server.h index 0eb0e078..eebaa20f 100644 --- a/include/opendht/dht_proxy_server.h +++ b/include/opendht/dht_proxy_server.h @@ -39,7 +39,7 @@ #endif namespace http { - class Client; + class Request; class opendht_logger_t; struct ListenerSession; class ConnectionListener; @@ -297,7 +297,7 @@ private: * @param key of the device * @param json, the content to send */ - void sendPushNotification(const std::string& key, Json::Value&& json, bool isAndroid) const; + void sendPushNotification(const std::string& key, Json::Value&& json, bool isAndroid); /** * Send push notification with an expire timeout. @@ -330,9 +330,13 @@ private: std::shared_ptr logger_; Json::StreamWriterBuilder jsonBuilder_; + // http server std::thread httpServerThread_; std::unique_ptr> httpServer_; - std::unique_ptr httpClient_; + + // http client + std::pair pushHostPort_; + std::map> requests_; mutable std::mutex statsMutex_; mutable ServerStats stats_; diff --git a/include/opendht/http.h b/include/opendht/http.h index 5ff3b370..d7598a02 100644 --- a/include/opendht/http.h +++ b/include/opendht/http.h @@ -28,28 +28,40 @@ namespace http { -using ConnectionId = unsigned int; +using HandlerCb = std::function; class OPENDHT_PUBLIC Connection { public: - Connection(const ConnectionId id, asio::ip::tcp::socket socket); + Connection(asio::io_context& ctx, std::shared_ptr logger = {}); ~Connection(); - ConnectionId id(); + unsigned int id(); bool is_open(); bool is_v6(); + + void set_endpoint(const asio::ip::tcp::endpoint& endpoint); + + asio::streambuf& input(); + asio::streambuf& data(); + asio::ip::tcp::socket& socket(); + + void timeout(const std::chrono::seconds timeout, HandlerCb cb = {}); + void close(); private: - friend class Client; + unsigned int id_; + static unsigned int ids_; - ConnectionId id_; asio::ip::tcp::socket socket_; - asio::streambuf request_; - std::string response_body_; - asio::streambuf response_chunk_; asio::ip::tcp::endpoint endpoint_; + + asio::streambuf write_buf_; + asio::streambuf read_buf_; + + std::unique_ptr timeout_timer_; + std::shared_ptr logger_; }; /** @@ -63,19 +75,6 @@ struct ListenerSession std::shared_ptr> response; }; -/** - * Context of an active connection allowing it to parse responses etc. - */ -struct ConnectionContext -{ - // TODO store multiple requests - std::string request; - std::unique_ptr parser; - std::unique_ptr parser_settings; - std::shared_ptr connection; - std::unique_ptr timeout_timer; -}; - class ConnectionListener { public: @@ -96,76 +95,152 @@ private: std::shared_ptr dht_; std::shared_ptr lock_; - std::shared_ptr> listeners_; + std::shared_ptr> listeners_; + std::shared_ptr logger_; }; -class OPENDHT_PUBLIC Client +/* @class Resolver + * @brief The purpose is to only resolve once to avoid mutliple dns requests per operation. + */ +class OPENDHT_PUBLIC Resolver { public: - using HandlerCb = std::function; - using ConnectionCb = std::function)>; + using ResolverCb = std::function results)>; - Client(asio::io_context& ctx, const std::string host, const std::string service = "80", - std::shared_ptr logger = {}, const bool resolve = true); + Resolver(asio::io_context& ctx, const std::string& host, const std::string& service = "80", + std::shared_ptr logger = {}); - asio::io_context& io_context(); + // use already resolved endpoints with classes using this resolver + Resolver(asio::io_context& ctx, asio::ip::basic_resolver_results endpoints, + std::shared_ptr logger = {}); + + ~Resolver(); + + void add_callback(ResolverCb cb); + +private: + void resolve(const std::string host, const std::string service); + + std::mutex mutex_; + + asio::error_code ec_; + asio::ip::tcp::resolver resolver_; + asio::ip::basic_resolver_results endpoints_; + + bool completed_ {false}; + std::queue cbs_; + + std::shared_ptr logger_; +}; + +struct Response +{ + unsigned int status_code; + std::map headers; + std::string body; +}; + +class OPENDHT_PUBLIC Request +{ +public: + enum class State { + CREATED, + SENDING, + HEADER_RECEIVED, + RECEIVING, + DONE + }; + using OnStateChangeCb = std::function; + using OnStatusCb = std::function; + using OnDataCb = std::function; + using OnMessageCompleteCb = std::function; + + // resolves implicitly + Request(asio::io_context& ctx, const std::string& host, const std::string& service = "80", + std::shared_ptr logger = {}); + + // user defined resolver + Request(asio::io_context& ctx, std::shared_ptr resolver, std::shared_ptr logger = {}); + + // user defined resolved endpoints + Request(asio::io_context& ctx, asio::ip::basic_resolver_results&& endpoints, + std::shared_ptr logger = {}); + + ~Request(); + + unsigned int id() const; + std::shared_ptr get_connection() const; void set_logger(std::shared_ptr 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 set_header(const restinio::http_request_header_t header); + void set_header_field(const restinio::http_field_t field, const std::string& value); + void set_connection_type(const restinio::http_connection_header_t connection); + void set_body(const std::string& body); - bool resolved(); - void async_resolve(const std::string host, const std::string service, HandlerCb cb = {}); + void add_on_state_changed_callback(OnStateChangeCb cb); + void add_on_status_callback(OnStatusCb cb); + void add_on_body_callback(OnDataCb cb); + void add_on_message_complete_callback(OnMessageCompleteCb cb); - void async_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); - - void async_request(std::shared_ptr conn, - std::string request, std::unique_ptr parser, - std::unique_ptr parser_s, HandlerCb cb = {}); + void send(); + void end(); private: - std::shared_ptr create_connection(); + struct Callbacks { + Callbacks(){} - void handle_resolve(const asio::error_code& ec, - asio::ip::tcp::resolver::iterator endpoint_it, - std::shared_ptr conn = {}, HandlerCb cb = {}); + OnStatusCb on_status; + OnDataCb on_header_field; + OnDataCb on_header_value; + OnDataCb on_body; + OnMessageCompleteCb on_message_complete; - void handle_request(const asio::error_code& ec, - std::shared_ptr conn = {}, HandlerCb cb = {}); + OnStateChangeCb on_state_changed; + }; - void handle_response_header(const asio::error_code& ec, const size_t bytes, - std::shared_ptr conn = {}, HandlerCb cb = {}); + void notify_state_changed(const State state); - void handle_response_body(const asio::error_code& ec, const size_t bytes, - std::shared_ptr conn = {}, HandlerCb cb = {}); + void build(); + void init_parser(); - void parse_request(const std::string request, const ConnectionId conn_id = 0); + void connect(asio::ip::basic_resolver_results&& endpoints, HandlerCb cb = {}); - size_t get_content_length(const std::string str, const ConnectionId conn_id = 0); + void terminate(const asio::error_code& ec); + + void post(); + + void handle_request(const asio::error_code& ec); + + void handle_response_header(const asio::error_code& ec, const size_t bytes); + + void handle_response_body(const asio::error_code& ec, const size_t bytes, const std::string chunk); + + void parse_request(const std::string request); + + restinio::http_request_header_t header_; + std::map headers_; + restinio::http_connection_header_t connection_type_; + std::string body_; + + std::mutex cbs_mutex_; + std::unique_ptr cbs_; std::string service_; std::string host_; - // contains the io_context - asio::ip::tcp::resolver resolver_; - // resolved endpoint - asio::ip::basic_resolver_results endpoints_; + unsigned int id_; + static unsigned int ids_; + asio::io_context& ctx_; + std::shared_ptr conn_; + std::shared_ptr resolver_; - ConnectionId connId_ {1}; - /* - * An association between an active connection and its context. - */ - std::map connections_; + Response response_ {}; + std::string request_; + std::unique_ptr parser_; + std::unique_ptr parser_s_; std::shared_ptr logger_; }; diff --git a/src/dht_proxy_client.cpp b/src/dht_proxy_client.cpp index b96ce111..1afaeb9a 100644 --- a/src/dht_proxy_client.cpp +++ b/src/dht_proxy_client.cpp @@ -30,9 +30,9 @@ struct DhtProxyClient::InfoState { std::atomic_bool cancel {false}; }; -struct DhtProxyClient::ListenState { +struct DhtProxyClient::OperationState { std::atomic_bool ok {true}; - std::atomic_bool cancel {false}; + std::atomic_bool stop {false}; }; struct DhtProxyClient::Listener @@ -41,11 +41,11 @@ struct DhtProxyClient::Listener cache(std::move(c)) {} - uint16_t connId {0}; unsigned callbackId; OpValueCache cache; ValueCallback cb; - Sp state; + Sp state; + std::shared_ptr request; std::unique_ptr refreshSubscriberTimer; }; @@ -70,16 +70,16 @@ struct DhtProxyClient::ProxySearch { DhtProxyClient::DhtProxyClient() {} DhtProxyClient::DhtProxyClient(std::function signal, const std::string& serverHost, - const std::string& pushClientId, std::shared_ptr logger): - pushClientId_(pushClientId), loopSignal_(signal), - logger_(logger) + const std::string& pushClientId, std::shared_ptr logger) + : pushClientId_(pushClientId), loopSignal_(signal), logger_(logger) { // build http client serverHostService_ = splitPort(serverHost); serverHostService_.second = serverHostService_.second.empty() ? "80" : serverHostService_.second; - httpClient_ = std::make_unique(httpContext_, - serverHostService_.first, serverHostService_.second); + // resolve once + resolver_ = std::make_shared(httpContext_, serverHostService_.first, + serverHostService_.second, logger_); // run http client httpClientThread_ = std::thread([this](){ try { @@ -188,15 +188,14 @@ DhtProxyClient::cancelAllListeners() auto l = s.second.listeners.find(token); if (l == s.second.listeners.end()) return; - if (httpClient_->active_connection(l->second.connId)){ - l->second.state->cancel = true; + if (l->second.request->get_connection()->is_open()){ + l->second.state->stop.store(true); try { - httpClient_->close_connection(l->second.connId); + l->second.request.reset(); } catch (const std::exception& e) { if (logger_) logger_->e("[proxy:client] [listeners:cancel:all] error closing socket: %s", e.what()); } - l->second.connId = 0; } s.second.listeners.erase(token); }); @@ -256,86 +255,41 @@ DhtProxyClient::periodic(const uint8_t*, size_t, SockAddr) return time_point::max(); } -restinio::http_header_fields_t -DhtProxyClient::initHeaderFields(){ - restinio::http_header_fields_t header_fields; - header_fields.append_field(restinio::http_field_t::host, - (serverHostService_.first + ":" + serverHostService_.second).c_str()); - header_fields.append_field(restinio::http_field_t::user_agent, "RESTinio client"); - header_fields.append_field(restinio::http_field_t::accept, "*/*"); - header_fields.append_field(restinio::http_field_t::content_type, "application/json"); - return header_fields; +void +DhtProxyClient::setHeaderFields(std::shared_ptr request){ + request->set_header_field(restinio::http_field_t::host, + (serverHostService_.first + ":" + serverHostService_.second).c_str()); + request->set_header_field(restinio::http_field_t::user_agent, "RESTinio client"); + request->set_header_field(restinio::http_field_t::accept, "*/*"); + request->set_header_field(restinio::http_field_t::content_type, "application/json"); } void -DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, - Value::Filter&& f, Where&& w) +DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Value::Filter&& f, Where&& w) { if (logger_) logger_->d("[proxy:client] [get] [search %s]", key.to_c_str()); - Value::Filter filter = w.empty() ? f : f.chain(w.getFilter()); - httpClient_->async_connect([this, key, cb, donecb, filter](std::shared_ptr conn) - { + auto request = std::make_shared(httpContext_, resolver_, logger_); + auto reqid = request->id(); + try { restinio::http_request_header_t header; header.request_target("/" + key.toString()); header.method(restinio::http_method_get()); - auto header_fields = this->initHeaderFields(); - auto request = httpClient_->create_request(header, header_fields, - restinio::http_connection_header_t::keep_alive, ""/*body*/); + request->set_header(header); + setHeaderFields(request); - struct GetContext { - std::function)> cb; // wrapper - DoneCallbackSimple donecb; // wrapper - Value::Filter filter; - std::atomic_bool ok {true}; - std::atomic_bool stop {false}; - std::shared_ptr logger; - }; - auto context = std::make_shared(); - context->filter = filter; - if (logger_) - context->logger = logger_; - std::weak_ptr wcontext = context; - context->cb = [this, key, wcontext, cb](const dht::Sp value) -> bool { - auto context = wcontext.lock(); - { - std::lock_guard lock(lockCallbacks_); - callbacks_.emplace_back([context, cb, value](){ - if (not context->stop and not cb({value})){ - context->stop = true; - } - }); - } - loopSignal_(); - return context->ok; - }; - context->donecb = [this, key, donecb](bool ok){ - { - std::lock_guard lock(lockCallbacks_); - callbacks_.emplace_back([=](){ - donecb(ok, {}); - }); - } - loopSignal_(); - }; - auto parser = std::make_unique(); - http_parser_init(parser.get(), HTTP_RESPONSE); - parser->data = static_cast(context.get()); + auto state = std::make_shared(); + Value::Filter filter = w.empty() ? f : f.chain(w.getFilter()); - auto parser_s = std::make_unique(); - http_parser_settings_init(parser_s.get()); - parser_s->on_status = [](http_parser* parser, const char* /*at*/, size_t /*length*/) -> int { - auto context = static_cast(parser->data); - if (parser->status_code != 200){ - if (context->logger) - context->logger->e("[proxy:client] [get] status error: %i", parser->status_code); - context->ok = true; + request->add_on_status_callback([this, key, state](unsigned int status_code){ + if (status_code != 200){ + if (logger_) + logger_->e("[proxy:client] [get %s] status error: %i", key.to_c_str(), status_code); + state->ok.store(false); } - return 0; - }; - parser_s->on_body = [](http_parser* parser, const char* at, size_t length) -> int { - auto context = static_cast(parser->data); + }); + request->add_on_body_callback([this, key, state, filter, cb](const char* at, size_t length){ try { Json::CharReaderBuilder rbuilder; auto body = std::string(at, length); @@ -343,64 +297,84 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, // one value per body line std::string data_line; std::stringstream body_stream(body); - while (std::getline(body_stream, data_line, '\n') and !context->stop){ + while (std::getline(body_stream, data_line, '\n') and !(state->stop.load())){ std::string err; Json::Value json; auto* char_data = static_cast(&data_line[0]); auto reader = std::unique_ptr(rbuilder.newCharReader()); if (!reader->parse(char_data, char_data + data_line.size(), &json, &err)){ - context->ok = false; - return 1; + state->ok.store(false); + return; } auto value = std::make_shared(json); values.push_back(value); - if ((not context->filter or context->filter(*value)) and context->cb){ - context->cb(value); + if ((not filter or filter(*value)) and cb){ + { + std::lock_guard lock(lockCallbacks_); + callbacks_.emplace_back([state, cb, value](){ + if (not state->stop.load() and not cb({value})){ + state->stop.store(true); + } + }); + } + loopSignal_(); } } } catch(const std::exception& e) { - if (context->logger) - context->logger->e("[proxy:client] [get] body parsing error: %s", e.what()); - context->ok = false; - return 1; + if (logger_) + logger_->e("[proxy:client] [get %s] body parsing error: %s", key.to_c_str(), e.what()); + state->ok.store(false); } - return 0; - }; - parser_s->on_message_complete = [](http_parser* parser) -> int { - auto context = static_cast(parser->data); - try { - if (context->donecb) - context->donecb(context->ok); - } catch(const std::exception& e) { - if (context->logger) - context->logger->e("[proxy:client] [get] message complete parsing error: %i", - parser->status_code); - return 1; - } - return 0; - }; - // keeping context data alive - httpClient_->async_request(conn, request, std::move(parser), std::move(parser_s), - [this, key, context](const asio::error_code& ec){ - if (ec && ec != asio::error::eof && logger_) - logger_->e("[proxy:client] [get %s] failed: %s", key.to_c_str(), ec.message().c_str()); }); - }); + request->add_on_message_complete_callback([this, key, state, donecb, reqid](const http::Response response){ + if (response.status_code != 200){ + logger_->e("[proxy:client] [get %s] failed with code=%i", key.to_c_str(), response.status_code); + state->ok.store(false); + } + else { + try { + if (donecb){ + { + std::lock_guard lock(lockCallbacks_); + callbacks_.emplace_back([donecb, state](){ + donecb(state->ok, {}); + }); + } + loopSignal_(); + } + } catch(const std::exception& e) { + if (logger_) + logger_->e("[proxy:client] [get %s] message complete parsing error", key.to_c_str()); + } + } + }); + request->add_on_state_changed_callback([this, reqid](const http::Request::State state){ + if (state == http::Request::State::DONE) + requests_.erase(reqid); + }); + request->send(); + requests_[reqid] = request; + } + catch (const std::exception &e){ + if (logger_) + logger_->e("[proxy:client] [get %s] error: %s", key.to_c_str(), e.what()); + requests_.erase(reqid); + } } void -DhtProxyClient::put(const InfoHash& key, Sp val, DoneCallback cb, - time_point created, bool permanent) +DhtProxyClient::put(const InfoHash& key, Sp val, DoneCallback cb, time_point created, bool permanent) { - if (logger_) - logger_->d("[proxy:client] [put] [search %s]", key.to_c_str()); if (not val){ if (cb) cb(false, {}); return; } + if (logger_) + logger_->d("[proxy:client] [put] [search %s]", key.to_c_str()); + if (val->id == Value::INVALID_ID) { - crypto::random_device rdev; + crypto::random_device rdev; std::uniform_int_distribution rand_id {}; val->id = rand_id(rdev); } @@ -410,7 +384,8 @@ DhtProxyClient::put(const InfoHash& key, Sp val, DoneCallback cb, auto& search = searches_[key]; auto refreshPutTimer = std::make_unique(httpContext_, std::chrono::steady_clock::now() + proxy::OP_TIMEOUT - proxy::OP_MARGIN); - auto ok = std::make_shared(false); + auto ok = std::make_shared(); + ok->store(true); refreshPutTimer->async_wait(std::bind(&DhtProxyClient::handleRefreshPut, this, std::placeholders::_1, key, id, ok)); search.puts.erase(id); @@ -453,12 +428,16 @@ DhtProxyClient::doPut(const InfoHash& key, Sp val, DoneCallback cb, time_ { if (logger_) logger_->d("[proxy:client] [put] [search %s] executing for %s", key.to_c_str(), val->toString().c_str()); - httpClient_->async_connect([this, key, val, cb, permanent](std::shared_ptr conn) - { + + auto request = std::make_shared(httpContext_, resolver_, logger_); + auto reqid = request->id(); + try { restinio::http_request_header_t header; header.request_target("/" + key.toString()); header.method(restinio::http_method_post()); - auto header_fields = this->initHeaderFields(); + request->set_header(header); + setHeaderFields(request); + auto json = val->toJson(); if (permanent) { if (deviceKey_.empty()) { @@ -477,66 +456,52 @@ DhtProxyClient::doPut(const InfoHash& key, Sp val, DoneCallback cb, time_ wbuilder["commentStyle"] = "None"; wbuilder["indentation"] = ""; auto body = Json::writeString(wbuilder, json); - auto request = httpClient_->create_request(header, header_fields, - restinio::http_connection_header_t::close, body); + request->set_body(body); - struct PutContext { - DoneCallbackSimple donecb; - std::atomic_bool ok; - std::shared_ptr logger; - }; - auto context = std::make_shared(); - context->ok = false; - if (logger_) - context->logger = logger_; - std::weak_ptr wcontext = context; - context->donecb = [this, wcontext, cb](bool ok){ - auto context = wcontext.lock(); - { - std::lock_guard lock(lockCallbacks_); - callbacks_.emplace_back([=](){ - cb(ok, {}); - }); - } - loopSignal_(); - }; - // define http parser - auto parser = std::make_unique(); - http_parser_init(parser.get(), HTTP_RESPONSE); - parser->data = static_cast(context.get()); - // define http parser callbacks - auto parser_s = std::make_unique(); - http_parser_settings_init(parser_s.get()); - parser_s->on_status = [](http_parser* parser, const char* /*at*/, size_t /*length*/) -> int { - auto context = static_cast(parser->data); - if (parser->status_code == 200){ - context->ok = true; - } else { - if (context->logger) - context->logger->e("[proxy:client] [put] status error: %i", parser->status_code); + auto ok = std::make_shared(); + ok->store(true); + + request->add_on_status_callback([this, key, ok](unsigned int status_code){ + if (status_code != 200){ + if (logger_) + logger_->e("[proxy:client] [put %s] status error: %i", key.to_c_str(), status_code); + ok->store(false); } return 0; - }; - parser_s->on_message_complete = [](http_parser* parser) -> int { - auto context = static_cast(parser->data); - try { - if (context->donecb) - context->donecb(context->ok); - } catch(const std::exception& e) { - if (context->logger) - context->logger->e("[proxy:client] [put] message complete error: %s", e.what()); - return 1; - } - return 0; - }; - // keeping context data alive - httpClient_->async_request(conn, request, std::move(parser), std::move(parser_s), - [this, key, context](const asio::error_code& ec){ - if (ec && ec != asio::error::eof && logger_) - logger_->e("[proxy:client] [put %s] failed: %s", key.to_c_str(), ec.message().c_str()); }); - - }); + request->add_on_message_complete_callback([this, key, ok, cb, reqid](const http::Response response){ + if (response.status_code != 200){ + if (logger_) + logger_->e("[proxy:client] [put %s] message complete error: %i", + key.to_c_str(), response.status_code); + } + else { + try { + if (cb){ + { + std::lock_guard lock(lockCallbacks_); + callbacks_.emplace_back([cb, ok](){ + cb(ok->load(), {}); + }); + } + loopSignal_(); + } + } catch(const std::exception& e) { + } + } + }); + request->add_on_state_changed_callback([this, reqid](const http::Request::State state){ + if (state == http::Request::State::DONE) + requests_.erase(reqid); + }); + request->send(); + requests_[reqid] = request; + } + catch (const std::exception &e){ + if (logger_) + logger_->e("[proxy:client] [put %s] error: %s", key.to_c_str(), e.what()); + requests_.erase(reqid); + } } /** @@ -626,108 +591,80 @@ DhtProxyClient::handleProxyStatus(const asio::error_code& ec, std::shared_ptre("[proxy:client] [status] handling error: %s", ec.message().c_str()); return; } - // A node can have a Ipv4 and a Ipv6. So, we need to retrieve all public ips - httpClient_->async_resolve(serverHostService_.first, serverHostService_.second, - [this, infoState](const asio::error_code& ec){ - if (ec){ - if (logger_) - logger_->e("[proxy:client] [status] error resolving: %s", ec.message().c_str()); - return; - } - // in all cases attempt to connect - httpClient_->async_connect([this, infoState](std::shared_ptr conn){ - try { - // make an http header - restinio::http_request_header_t header; - header.request_target("/"); - header.method(restinio::http_method_get()); - auto header_fields = this->initHeaderFields(); - auto request = httpClient_->create_request(header, header_fields, - restinio::http_connection_header_t::keep_alive, ""/*body*/); - if (logger_) - logger_->d("[proxy:client] [status] sending request"); - // initalise the parser callback data - struct StatusContext { - unsigned int family; - std::function cb; // wrapper - std::atomic_bool ok {true}; - std::shared_ptr infoState; - std::function proxyInfo; - std::shared_ptr logger; - }; - auto context = std::make_shared(); - context->infoState = infoState; - context->family = conn->is_v6() ? AF_INET6 : AF_INET; - context->proxyInfo = std::bind(&DhtProxyClient::onProxyInfos, this, - std::placeholders::_1, std::placeholders::_2); - if (logger_) - context->logger = logger_; - std::weak_ptr wcontext = context; - context->cb = [this, wcontext](Json::Value infos) - { - auto context = wcontext.lock(); - if (context->family == AF_INET) - context->infoState->ipv4++; - else if (context->family == AF_INET6) - context->infoState->ipv6++; - if (not context->infoState->cancel) - context->proxyInfo(infos, context->family); - }; - // initialize the parser - auto parser = std::make_unique(); - http_parser_init(parser.get(), HTTP_RESPONSE); - parser->data = static_cast(context.get()); - // init the parser callbacks - auto parser_s = std::make_unique(); - http_parser_settings_init(parser_s.get()); - parser_s->on_status = [](http_parser* parser, const char* /*at*/, size_t /*length*/) -> int { - auto context = static_cast(parser->data); - if (parser->status_code != 200){ - if (context->logger) - context->logger->e("[proxy:client] [status] error: %i", parser->status_code); - context->ok = true; - } - return 0; - }; - parser_s->on_body = [](http_parser* parser, const char* at, size_t length) -> int { - auto context = static_cast(parser->data); - try{ - std::string err; - Json::Value proxyInfos; - Json::CharReaderBuilder rbuilder; - auto body = std::string(at, length); - auto* char_data = static_cast(&body[0]); - auto reader = std::unique_ptr(rbuilder.newCharReader()); - if (!reader->parse(char_data, char_data + body.size(), &proxyInfos, &err)){ - context->ok = false; - return 1; - } - context->cb(proxyInfos); - } - catch (const std::exception& e) { - if (context->logger) - context->logger->e("[proxy:client] [status] body error: %s", e.what()); - context->ok = false; - return 1; - } - return 0; - }; - if (context->infoState->cancel) - return; - // keeping context data alive - httpClient_->async_request(conn, request, std::move(parser), std::move(parser_s), - [this, context](const asio::error_code& ec){ - if (ec && ec != asio::error::eof && logger_) - logger_->e("[proxy:client] [status] failed: %s", ec.message().c_str()); - }); + if (logger_) + logger_->d("[proxy:client] [status] sending request"); + auto request = std::make_shared(httpContext_, resolver_, logger_); + auto reqid = request->id(); + try { + // make an http header + restinio::http_request_header_t header; + header.request_target("/"); + header.method(restinio::http_method_get()); + request->set_header(header); + setHeaderFields(request); + + std::weak_ptr wreq = request; + auto ok = std::make_shared(); + ok->store(true); + + request->add_on_status_callback([this, ok](unsigned int status_code){ + if (status_code != 200){ + if (logger_) + logger_->e("[proxy:client] [status] error: %i", status_code); + ok->store(false); + } + }); + request->add_on_body_callback([this, ok, infoState, wreq](const char* at, size_t length){ + try{ + std::string err; + Json::Value proxyInfos; + Json::CharReaderBuilder rbuilder; + auto body = std::string(at, length); + auto* char_data = static_cast(&body[0]); + auto reader = std::unique_ptr(rbuilder.newCharReader()); + if (!reader->parse(char_data, char_data + body.size(), &proxyInfos, &err)){ + ok->store(false); + return; + } + auto req = wreq.lock(); + unsigned int family = req->get_connection()->is_v6() ? AF_INET6 : AF_INET; + if (family == AF_INET){ + infoState->ipv4++; + } + else if (family == AF_INET6){ + infoState->ipv6++; + } + if (not infoState->cancel) + onProxyInfos(proxyInfos, family); } catch (const std::exception& e) { if (logger_) - logger_->e("[proxy:client] [info] error sending request: %s", e.what()); + logger_->e("[proxy:client] [status] body error: %s", e.what()); + ok->store(false); } }); - }); + request->add_on_message_complete_callback([this](const http::Response response){ + if (response.status_code != 200) + if (logger_) + logger_->e("[proxy:client] [status] failed with code=%i", response.status_code); + }); + request->add_on_state_changed_callback([this, reqid](const http::Request::State state){ + if (state == http::Request::State::DONE) + requests_.erase(reqid); + }); + + if (infoState->cancel.load()) + return; + + request->send(); + requests_[reqid] = request; + } + catch (const std::exception &e){ + if (logger_) + logger_->e("[proxy:client] [status] error sending request: %s", e.what()); + requests_.erase(reqid); + } } void @@ -834,13 +771,13 @@ DhtProxyClient::listen(const InfoHash& key, ValueCallback cb, Value::Filter filt std::forward_as_tuple(std::move(cb))).first; } else { if (l->second.state) - l->second.state->cancel = true; + l->second.state->stop = true; } // Add cache callback - auto state = std::make_shared(); + auto state = std::make_shared(); l->second.state = state; l->second.cb = [this,key,token,state](const std::vector>& values, bool expired){ - if (state->cancel) + if (state->stop) return false; std::lock_guard lock(searchLock_); auto s = searches_.find(key); @@ -883,7 +820,7 @@ DhtProxyClient::listen(const InfoHash& key, ValueCallback cb, Value::Filter filt void DhtProxyClient::handleResubscribe(const asio::error_code &ec, const InfoHash& key, - const size_t token, std::shared_ptr state) + const size_t token, std::shared_ptr state) { if (ec == asio::error::operation_aborted) return; @@ -892,7 +829,7 @@ DhtProxyClient::handleResubscribe(const asio::error_code &ec, const InfoHash& ke logger_->e("[proxy:client] [resubscribe %s] %s", key.toString().c_str(), ec.message().c_str()); return; } - if (state->cancel) + if (state->stop) return; std::lock_guard lock(searchLock_); auto s = searches_.find(key); @@ -956,16 +893,19 @@ DhtProxyClient::handleExpireListener(const asio::error_code &ec, const InfoHash& return; auto& listener = it->second; - listener.state->cancel = true; + listener.state->stop = true; + if (not deviceKey_.empty()) { // UNSUBSCRIBE - httpClient_->async_connect([this, key](std::shared_ptr conn) - { + auto request = std::make_shared(httpContext_, resolver_, logger_); + auto reqid = request->id(); + try { restinio::http_request_header_t header; header.request_target("/" + key.toString()); header.method(restinio::http_method_unsubscribe()); - auto header_fields = this->initHeaderFields(); - // fill request body + request->set_header(header); + setHeaderFields(request); + Json::Value body; body["key"] = deviceKey_; body["client_id"] = pushClientId_; @@ -974,51 +914,35 @@ DhtProxyClient::handleExpireListener(const asio::error_code &ec, const InfoHash& wbuilder["indentation"] = ""; auto content = Json::writeString(wbuilder, body) + "\n"; std::replace(content.begin(), content.end(), '\n', ' '); - // build the request - auto request = httpClient_->create_request(header, header_fields, - restinio::http_connection_header_t::keep_alive, content); - // define context - struct UnsubscribeContext { - InfoHash key; - std::shared_ptr logger; - }; - auto context = std::make_shared(); - context->key = key; - if (logger_) - context->logger = logger_; - // define parser - auto parser = std::make_unique(); - http_parser_init(parser.get(), HTTP_RESPONSE); - parser->data = static_cast(context.get()); - // define callbacks - auto parser_s = std::make_unique(); - http_parser_settings_init(parser_s.get()); - parser_s->on_status = [](http_parser* parser, const char* /*at*/, size_t /*length*/) -> int { - auto context = static_cast(parser->data); - if (parser->status_code != 200){ - if (context->logger) - context->logger->e("[proxy:client] [search %s] cancel listen failed: %i", - context->key.to_c_str(), parser->status_code); - } - return 0; - }; - httpClient_->async_request(conn, request, std::move(parser), std::move(parser_s), - [this, key](const asio::error_code& ec){ - if (ec && ec != asio::error::eof && logger_) - logger_->e("[proxy:client] [unsubscribe %s] failed: %s", key.to_c_str(), ec.message().c_str()); + request->set_body(content); + + request->add_on_status_callback([this, key](unsigned int status_code){ + if (status_code != 200) + if (logger_) + logger_->e("[proxy:client] [unsubscribe %s] failed with code=%i", + key.to_c_str(), status_code); }); - }); - } else { - // Just stop the request - if (httpClient_->active_connection(listener.connId)){ - try { - httpClient_->close_connection(listener.connId); - } - catch (const std::exception& e){ - if (logger_) - logger_->e("[proxy:client] [listen:cancel] error closing socket: %s", e.what()); - } + request->add_on_message_complete_callback([this, key, reqid](const http::Response response){ + if (response.status_code != 200) + if (logger_) + logger_->e("[proxy:client] [unsubscribe %s] failed with code=%i", + key.to_c_str(), response.status_code); + }); + request->add_on_state_changed_callback([this, reqid](const http::Request::State state){ + if (state == http::Request::State::DONE) + requests_.erase(reqid); + }); + request->send(); + requests_[reqid] = request; } + catch (const std::exception &e){ + if (logger_) + logger_->e("[proxy:client] [unsubscribe %s] failed: %s", key.to_c_str(), e.what()); + requests_.erase(reqid); + } + } else { + // stop the request + listener.request.reset(); } search->second.listeners.erase(it); if (logger_) @@ -1038,65 +962,32 @@ DhtProxyClient::handleExpireListener(const asio::error_code &ec, const InfoHash& void DhtProxyClient::sendListen(const restinio::http_request_header_t header, const ValueCallback& cb, - const Sp& state, + const Sp& state, Listener& listener, ListenMethod method) { - httpClient_->async_connect([this, &listener, header, cb, state, method](std::shared_ptr conn) - { - auto connId = conn->id(); - listener.connId = connId; - // define request header fields - auto headers = this->initHeaderFields(); - auto connType = restinio::http_connection_header_t::close; + auto request = std::make_shared(httpContext_, resolver_, logger_); + listener.request = request; + auto reqid = request->id(); + try { + request->set_header(header); + setHeaderFields(request); if (method == ListenMethod::LISTEN) - connType = restinio::http_connection_header_t::keep_alive; + request->set_connection_type(restinio::http_connection_header_t::keep_alive); std::string body; #ifdef OPENDHT_PUSH_NOTIFICATIONS if (method != ListenMethod::LISTEN) body = fillBody(method == ListenMethod::RESUBSCRIBE); #endif - auto request = httpClient_->create_request(header, headers, connType, body); + request->set_body(body); - struct ListenContext { - std::shared_ptr logger; - std::function, bool)> cb; // wrapper - Value::Filter filter; - std::shared_ptr state; - }; - auto context = std::make_shared(); - if (logger_) - context->logger = logger_; - context->state = state; - std::weak_ptr wcontext = context; - context->cb = [this, cb, wcontext, state](const std::shared_ptr value, bool expired){ - auto context = wcontext.lock(); - { - std::lock_guard lock(lockCallbacks_); - callbacks_.emplace_back([cb, value, state, expired]() { - if (not state->cancel and not cb({value}, expired)) - state->cancel = true; - }); + request->add_on_status_callback([this, reqid, state](unsigned int status_code){ + if (status_code != 200){ + if (logger_) + logger_->e("[proxy:client] [listen:send] %i status error: %i", reqid, status_code); + state->ok.store(false); } - loopSignal_(); - }; - - auto parser = std::make_unique(); - http_parser_init(parser.get(), HTTP_RESPONSE); - parser->data = static_cast(context.get()); - - auto parser_s = std::make_unique(); - http_parser_settings_init(parser_s.get()); - parser_s->on_status = [](http_parser* parser, const char* /*at*/, size_t /*length*/) -> int { - auto context = static_cast(parser->data); - if (parser->status_code != 200){ - if (context->logger) - context->logger->e("[proxy:client] [listen] status error: %i", parser->status_code); - context->state->ok = false; - } - return 0; - }; - parser_s->on_body = [](http_parser* parser, const char* at, size_t length) -> int { - auto context = static_cast(parser->data); + }); + request->add_on_body_callback([this, reqid, state, cb](const char* at, size_t length){ try { Json::CharReaderBuilder rbuilder; auto body = std::string(at, length); @@ -1104,40 +995,55 @@ DhtProxyClient::sendListen(const restinio::http_request_header_t header, // one value per body line std::string data_line; std::stringstream body_stream(body); - while (std::getline(body_stream, data_line, '\n') and !context->state->cancel){ + while (std::getline(body_stream, data_line, '\n') and !state->stop){ std::string err; Json::Value json; auto* char_data = static_cast(&data_line[0]); auto reader = std::unique_ptr(rbuilder.newCharReader()); if (!reader->parse(char_data, char_data + data_line.size(), &json, &err)){ - context->state->ok = false; - return 1; + state->ok.store(false); + return; } if (json.size() == 0){ // it's the end - context->state->cancel = true; + state->stop.store(true); } auto value = std::make_shared(json); - values.push_back(value); - if (context->cb){ + if (cb){ auto expired = json.get("expired", Json::Value(false)).asBool(); - context->cb(value, expired); + { + std::lock_guard lock(lockCallbacks_); + callbacks_.emplace_back([cb, value, state, expired]() { + if (not state->stop.load() and not cb({value}, expired)) + state->stop.store(true); + }); + } + loopSignal_(); } } } catch(const std::exception& e) { - if (context->logger) - context->logger->e("[proxy:client] [listen] error in parsing: %s", e.what()); - context->state->ok = false; - return 1; + if (logger_) + logger_->e("[proxy:client] [listen:send] %i error in parsing: %s", reqid, e.what()); + state->ok.store(false); } - return 0; - }; - // keeping context data alive - httpClient_->async_request(conn, request, std::move(parser), std::move(parser_s), - [this, context, connId](const asio::error_code& ec){ - if (logger_ and ec && ec != asio::error::eof) - logger_->e("[proxy:client] [listen:send] [connection %i] failed: %s", connId, ec.message().c_str()); }); - }); + request->add_on_message_complete_callback([this, reqid](const http::Response response){ + if (response.status_code != 200) + if (logger_) + logger_->e("[proxy:client] [listen:send] %i failed with code=%i", + reqid, response.status_code); + }); + request->add_on_state_changed_callback([this, reqid](const http::Request::State state){ + if (state == http::Request::State::DONE) + requests_.erase(reqid); + }); + request->send(); + requests_[reqid] = request; + } + catch (const std::exception &e){ + if (logger_) + logger_->e("[proxy:client] [listen:send] %i failed: %s", reqid, e.what()); + requests_.erase(reqid); + } } void @@ -1184,9 +1090,9 @@ DhtProxyClient::restartListeners() put.second.refreshPutTimer = std::make_unique(httpContext_); } put.second.refreshPutTimer->expires_at(std::chrono::steady_clock::now() + - proxy::OP_TIMEOUT - proxy::OP_MARGIN); + proxy::OP_TIMEOUT - proxy::OP_MARGIN); put.second.refreshPutTimer->async_wait(std::bind(&DhtProxyClient::handleRefreshPut, this, - std::placeholders::_1, key, put.first, put.second.ok)); + std::placeholders::_1, key, put.first, put.second.ok)); } } } @@ -1206,16 +1112,8 @@ DhtProxyClient::restartListeners() for (auto& l: search.second.listeners) { auto& listener = l.second; if (auto state = listener.state) - state->cancel = true; - if (httpClient_->active_connection(listener.connId)){ - try { - httpClient_->close_connection(listener.connId); - } catch (const std::exception& e) { - if (logger_) - logger_->e("[proxy:client] [listeners:restart] error closing socket: %s", e.what()); - } - l.second.connId = 0; - } + state->stop = true; + listener.request->end(); } } for (auto& search: searches_) { @@ -1223,8 +1121,8 @@ DhtProxyClient::restartListeners() auto& listener = l.second; auto state = listener.state; // Redo listen - state->cancel = false; - state->ok = true; + state->stop.store(false); + state->ok.store(true); auto cb = listener.cb; // define header restinio::http_request_header_t header; @@ -1271,7 +1169,7 @@ DhtProxyClient::pushNotificationReceived(const std::mapcancel) + if (list.second.state->stop) continue; if (logger_) logger_->d("[proxy:client] [push:received] [search %s] handling", key.to_c_str()); @@ -1299,7 +1197,7 @@ DhtProxyClient::pushNotificationReceived(const std::map lock(lockCallbacks_); callbacks_.emplace_back([this, key, token, state, ids]() { - if (state->cancel) + if (state->stop) return; std::lock_guard lock(searchLock_); auto s = searches_.find(key); @@ -1308,8 +1206,8 @@ DhtProxyClient::pushNotificationReceived(const std::mapsecond.listeners.find(token); if (l == s->second.listeners.end()) return; - if (not state->cancel and not l->second.cache.onValuesExpired(ids)) - state->cancel = true; + if (not state->stop and not l->second.cache.onValuesExpired(ids)) + state->stop = true; }); } loopSignal_(); @@ -1335,17 +1233,11 @@ DhtProxyClient::resubscribe(const InfoHash& key, const size_t token, Listener& l logger_->d("[proxy:client] [resubscribe] [search %s] resubscribe push listener", key.to_c_str()); auto state = listener.state; - state->cancel = true; - if (listener.connId) { - try { - httpClient_->close_connection(listener.connId); - } catch (const std::exception& e) { - if (logger_) - logger_->e("[proxy:client] [resubscribe] error closing socket: %s", e.what()); - } + state->stop = true; + if (listener.request) { + listener.request->end(); } - state->cancel = false; - state->ok = true; + state.reset(new OperationState()); restinio::http_request_header_t header; header.method(restinio::http_method_subscribe()); diff --git a/src/dht_proxy_server.cpp b/src/dht_proxy_server.cpp index 01596340..7dda6a10 100644 --- a/src/dht_proxy_server.cpp +++ b/src/dht_proxy_server.cpp @@ -81,10 +81,8 @@ DhtProxyServer::DhtProxyServer(std::shared_ptr dht, in_port_t port, restinio::own_io_context(), std::forward(settings) )); - // build http client - auto pushHostPort = splitPort(pushServer_); - httpClient_.reset(new http::Client(httpServer_->io_context(), - pushHostPort.first, pushHostPort.second)); + // define http request destination + pushHostPort_ = splitPort(pushServer_); // run http server httpServerThread_ = std::thread([this](){ try { @@ -659,80 +657,65 @@ DhtProxyServer::handleCancelPushListen(const asio::error_code &ec, const std::st } void -DhtProxyServer::sendPushNotification(const std::string& token, Json::Value&& json, bool isAndroid) const +DhtProxyServer::sendPushNotification(const std::string& token, Json::Value&& json, bool isAndroid) { if (pushServer_.empty()) return; - httpClient_->async_connect([this, token, json=std::move(json), isAndroid](std::shared_ptr conn) - { - if (!conn){ - logger_->e("[proxy:server] [notification] invalid connection"); - return; - } - try { - restinio::http_request_header_t header; - header.request_target("/api/push"); - header.method(restinio::http_method_post()); + auto request = std::make_shared(httpServer_->io_context(), pushHostPort_.first, + pushHostPort_.second, logger_); + auto reqid = request->id(); + try { + restinio::http_request_header_t header; + header.request_target("/api/push"); + header.method(restinio::http_method_post()); + request->set_header(header); - restinio::http_header_fields_t header_fields; - header_fields.append_field(restinio::http_field_t::host, pushServer_.c_str()); - header_fields.append_field(restinio::http_field_t::user_agent, "RESTinio client"); - header_fields.append_field(restinio::http_field_t::accept, "*/*"); - header_fields.append_field(restinio::http_field_t::content_type, "application/json"); + request->set_header_field(restinio::http_field_t::host, pushServer_.c_str()); + request->set_header_field(restinio::http_field_t::user_agent, "RESTinio client"); + request->set_header_field(restinio::http_field_t::accept, "*/*"); + request->set_header_field(restinio::http_field_t::content_type, "application/json"); - // NOTE: see https://github.com/appleboy/gorush - Json::Value notification(Json::objectValue); - Json::Value tokens(Json::arrayValue); - tokens[0] = token; - notification["tokens"] = std::move(tokens); - notification["platform"] = isAndroid ? 2 : 1; - notification["data"] = std::move(json); - notification["priority"] = "high"; - notification["time_to_live"] = 600; + // NOTE: see https://github.com/appleboy/gorush + Json::Value notification(Json::objectValue); + Json::Value tokens(Json::arrayValue); + tokens[0] = token; + notification["tokens"] = std::move(tokens); + notification["platform"] = isAndroid ? 2 : 1; + notification["data"] = std::move(json); + notification["priority"] = "high"; + notification["time_to_live"] = 600; - Json::Value notifications(Json::arrayValue); - notifications[0] = notification; + Json::Value notifications(Json::arrayValue); + notifications[0] = notification; - Json::Value content; - content["notifications"] = std::move(notifications); + Json::Value content; + content["notifications"] = std::move(notifications); - Json::StreamWriterBuilder wbuilder; - wbuilder["commentStyle"] = "None"; - wbuilder["indentation"] = ""; - auto body = Json::writeString(wbuilder, content); + Json::StreamWriterBuilder wbuilder; + wbuilder["commentStyle"] = "None"; + wbuilder["indentation"] = ""; - auto parser = std::make_unique(); - http_parser_init(parser.get(), HTTP_RESPONSE); + auto body = Json::writeString(wbuilder, content); + request->set_body(body); - struct PushContext { - std::shared_ptr logger; - }; - auto context = std::make_shared(); - if (logger_) - context->logger = logger_; - parser->data = static_cast(context.get()); - - auto parser_s = std::make_unique(); - http_parser_settings_init(parser_s.get()); - parser_s->on_status = [](http_parser* parser, const char* /*at*/, size_t /*length*/) -> int { - auto context = static_cast(parser->data); - if (parser->status_code == 200) - return 0; - if (context->logger) - context->logger->e("[proxy:server] [notification] error send push: %i", parser->status_code); - return 1; - }; - auto request = httpClient_->create_request(header, header_fields, - restinio::http_connection_header_t::close, body); - - httpClient_->async_request(conn, request, std::move(parser), std::move(parser_s)); - } - catch (const std::exception &e){ - if (logger_) - logger_->e("[proxy:server] [notification] error send push: %i", e.what()); - } - }); + request->add_on_message_complete_callback([this, reqid](const http::Response response){ + if (response.status_code != 200) + if (logger_) + logger_->e("[proxy:server] [notification] push failed: %i", response.status_code); + }); + request->add_on_state_changed_callback([this, reqid](const http::Request::State state){ + if (state == http::Request::State::DONE) + requests_.erase(reqid); + }); + request->send(); + requests_[reqid] = request; + } + catch (const std::exception &e){ + if (logger_) + logger_->e("[proxy:server] [notification] error send push: %i", e.what()); + requests_.erase(reqid); + } } #endif //OPENDHT_PUSH_NOTIFICATIONS diff --git a/src/http.cpp b/src/http.cpp index c92bb3b9..799d38c8 100644 --- a/src/http.cpp +++ b/src/http.cpp @@ -20,35 +20,91 @@ namespace http { -constexpr char HTTP_HEADER_CONTENT_LENGTH[] = "Content-Length:"; +constexpr char HTTP_HEADER_CONTENT_LENGTH[] = "Content-Length"; +constexpr char HTTP_HEADER_CONTENT_TYPE[] = "Content-Type"; +constexpr char HTTP_HEADER_CONTENT_TYPE_JSON[] = "application/json"; // connection -Connection::Connection(const ConnectionId id, asio::ip::tcp::socket socket): - id_(id), socket_(std::move(socket)) +unsigned int Connection::ids_ = 1; + +Connection::Connection(asio::io_context& ctx, std::shared_ptr logger): + id_(Connection::ids_++), socket_(ctx), logger_(logger) {} -Connection::~Connection(){ +Connection::~Connection() +{ close(); } -ConnectionId -Connection::id(){ +unsigned int +Connection::id() +{ return id_; } bool -Connection::is_open(){ +Connection::is_open() +{ return socket_.is_open(); } bool -Connection::is_v6(){ +Connection::is_v6() +{ return endpoint_.address().is_v6(); } void -Connection::close(){ +Connection::set_endpoint(const asio::ip::tcp::endpoint& endpoint) +{ + endpoint_ = endpoint; +} + +asio::streambuf& +Connection::input() +{ + return write_buf_; +} + +asio::streambuf& +Connection::data() +{ + return read_buf_; +} + +asio::ip::tcp::socket& +Connection::socket() +{ + return socket_; +} + +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(socket_.get_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::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 +120,8 @@ ConnectionListener::ConnectionListener(std::shared_ptr dht, {} ConnectionListener::~ConnectionListener() -{} +{ +} void ConnectionListener::state_changed(const restinio::connection_state::notice_t& notice) noexcept @@ -107,413 +164,574 @@ ConnectionListener::to_str(restinio::connection_state::cause_t cause) noexcept return result; } -// client +// Resolver -Client::Client(asio::io_context& ctx, const std::string host, const std::string service, - std::shared_ptr logger, const bool resolve): - resolver_(ctx), logger_(logger) +Resolver::Resolver(asio::io_context& ctx, const std::string& host, const std::string& service, + std::shared_ptr logger) + : resolver_(ctx), logger_(logger) { - if (resolve) - async_resolve(host, service); + resolve(host, service); } -asio::io_context& -Client::io_context() +Resolver::Resolver(asio::io_context& ctx, asio::ip::basic_resolver_results endpoints, + std::shared_ptr logger) + : resolver_(ctx), logger_(logger) { - return resolver_.get_io_context(); + endpoints_ = endpoints; + completed_ = true; +} + +Resolver::~Resolver() +{ + decltype(cbs_) cbs; + { + std::lock_guard lock(mutex_); + cbs = std::move(cbs_); + } + while (not cbs.empty()){ + auto cb = cbs.front(); + if (cb) + cb(asio::error::operation_aborted, {}); + cbs.pop(); + } } void -Client::set_logger(std::shared_ptr logger) +Resolver::add_callback(ResolverCb cb) +{ + std::lock_guard lock(mutex_); + if (!completed_) + cbs_.push(std::move(cb)); + else + cb(ec_, endpoints_); +} + +void +Resolver::resolve(const std::string host, const std::string service) +{ + asio::ip::tcp::resolver::query query_(host, service); + + resolver_.async_resolve(query_, [this, host, service] + (const asio::error_code& ec, asio::ip::tcp::resolver::results_type endpoints) + { + if (logger_){ + if (ec) + logger_->e("[http:resolver] error for %s:%s: %s", + host.c_str(), service.c_str(), ec.message().c_str()); + else { + for (auto it = endpoints.begin(); it != endpoints.end(); ++it){ + asio::ip::tcp::endpoint endpoint = *it; + logger_->d("[http:resolver] resolved %s:%s: address=%s ipv%i", + host.c_str(), service.c_str(), endpoint.address().to_string().c_str(), + endpoint.address().is_v6() ? 6 : 4); + } + } + } + decltype(cbs_) cbs; + { + std::lock_guard lock(mutex_); + ec_ = ec; + endpoints_ = endpoints; + completed_ = true; + cbs = std::move(cbs_); + } + while (not cbs.empty()){ + auto cb = cbs.front(); + if (cb) + cb(ec, endpoints); + cbs.pop(); + } + }); +} + +// Request + +unsigned int Request::ids_ = 1; + +Request::Request(asio::io_context& ctx, const std::string& host, const std::string& service, + std::shared_ptr logger) + : id_(Request::ids_++), ctx_(ctx), logger_(logger) +{ + cbs_ = std::make_unique(); + resolver_ = std::make_shared(ctx, host, service, logger_); +} + +Request::Request(asio::io_context& ctx, std::shared_ptr resolver, std::shared_ptr logger) + : id_(Request::ids_++), ctx_(ctx), logger_(logger) +{ + cbs_ = std::make_unique(); + resolver_ = resolver; +} + +// user defined resolved endpoints +Request::Request(asio::io_context& ctx, asio::ip::basic_resolver_results&& endpoints, + std::shared_ptr logger) + : id_(Request::ids_++), ctx_(ctx), logger_(logger) +{ + cbs_ = std::make_unique(); + resolver_ = std::make_shared(ctx, std::move(endpoints), logger_); +} + +Request::~Request() +{ +} + +void +Request::end() +{ + conn_.reset(); +} + +unsigned int +Request::id() const +{ + return id_; +} + +std::shared_ptr +Request::get_connection() const +{ + return conn_; +} + +void +Request::set_logger(std::shared_ptr logger) { logger_ = logger; } -bool -Client::active_connection(const ConnectionId conn_id) +void +Request::set_header(const restinio::http_request_header_t header) { - auto conn_context = connections_.find(conn_id); - if (conn_context == connections_.end()) - return false; - return conn_context->second.connection->is_open(); + header_ = header; } void -Client::close_connection(const ConnectionId conn_id) +Request::set_header_field(const restinio::http_field_t field, const std::string& value) { - 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 (logger_) - logger_->d("[http::client] [connection:%i] closed", conn_id); + headers_[field] = value; } void -Client::set_connection_timeout(const ConnectionId conn_id, - const std::chrono::seconds timeout, HandlerCb cb) +Request::set_connection_type(const restinio::http_connection_header_t connection) { - 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(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); - }); -} - -bool -Client::resolved() -{ - return !endpoints_.empty(); + connection_type_ = connection; } void -Client::async_resolve(const std::string host, const std::string service, HandlerCb cb) +Request::set_body(const std::string& body) { - // 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) - { - if (ec and logger_){ - logger_->e("[http::client] [resolve %s:%s] error resolving: %s", - 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(), - endpoint.address().is_v6() ? 6 : 4); - } - } - endpoints_ = endpoints; - } - if (cb) - cb(ec); - }); + body_ = body; } -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) +void +Request::build() { std::stringstream request; // first header - request << header.method().c_str() << " " << header.request_target() << " " << - "HTTP/" << header.http_major() << "." << header.http_minor() << "\r\n"; + request << header_.method().c_str() << " " << header_.request_target() << " " << + "HTTP/" << header_.http_major() << "." << header_.http_minor() << "\r\n"; // other headers - for (auto header_field: header_fields) - request << header_field.name() << ": " << header_field.value() << "\r\n"; + for (auto header: headers_) + request << restinio::field_to_string(header.first) << ": " << header.second << "\r\n"; // last connection header - std::string conn_str; - switch (connection){ - case restinio::http_connection_header_t::keep_alive: - conn_str = "keep-alive"; - break; + std::string conn_str = ""; + switch (connection_type_){ case restinio::http_connection_header_t::close: conn_str = "close"; break; + case restinio::http_connection_header_t::keep_alive: + conn_str = "keep-alive"; + break; case restinio::http_connection_header_t::upgrade: throw std::invalid_argument("upgrade"); break; + default: + conn_str = "close"; } request << "Connection: " << conn_str << "\r\n"; // body & content-length - if (!body.empty()){ - request << "Content-Length: " << body.size() << "\r\n\r\n"; - request << body; + if (!body_.empty()){ + request << "Content-Length: " << body_.size() << "\r\n\r\n"; + request << body_; } // last delim request << "\r\n"; - return request.str(); + request_ = request.str(); } void -Client::async_connect(ConnectionCb cb) +Request::add_on_state_changed_callback(OnStateChangeCb cb) { - if (endpoints_.empty()){ + std::lock_guard lock(cbs_mutex_); + cbs_->on_state_changed = std::move(cb); +} + +void +Request::add_on_status_callback(OnStatusCb cb) +{ + std::lock_guard lock(cbs_mutex_); + cbs_->on_status = std::move(cb); +} + +void +Request::add_on_body_callback(OnDataCb cb) +{ + std::lock_guard lock(cbs_mutex_); + cbs_->on_body = std::move(cb); +} + +void +Request::add_on_message_complete_callback(OnMessageCompleteCb cb) +{ + std::lock_guard lock(cbs_mutex_); + cbs_->on_message_complete = std::move(cb); +} + +void +Request::notify_state_changed(const State state) +{ + + std::lock_guard lock(cbs_mutex_); + if (cbs_->on_state_changed) + cbs_->on_state_changed(state); +} + +void +Request::init_parser() +{ + if (!parser_) + parser_ = std::make_unique(); + http_parser_init(parser_.get(), HTTP_RESPONSE); + parser_->data = static_cast(cbs_.get()); + + if (!parser_s_) + parser_s_ = std::make_unique(); + http_parser_settings_init(parser_s_.get()); + + // user registered callbacks wrappers to store its data in the response + auto on_status_cb = cbs_->on_status; + cbs_->on_status = [this, on_status_cb](unsigned int status_code){ + response_.status_code = status_code; + if (on_status_cb) + on_status_cb(status_code); + }; + auto header_field = std::make_shared(""); + auto on_header_field_cb = cbs_->on_header_field; + cbs_->on_header_field = [this, header_field, on_header_field_cb](const char* at, size_t length){ + header_field->erase(); + auto field = std::string(at, length); + header_field->append(field); + if (on_header_field_cb) + on_header_field_cb(at, length); + }; + auto on_header_value_cb = cbs_->on_header_value; + cbs_->on_header_value = [this, header_field, on_header_value_cb](const char* at, size_t length){ + response_.headers[*header_field] = std::string(at, length); + if (on_header_value_cb) + on_header_value_cb(at, length); + }; + auto on_body_cb = cbs_->on_body; + cbs_->on_body = [this, on_body_cb](const char* at, size_t length){ + auto content = std::string(at, length); + response_.body.append(content); + if (on_body_cb) + on_body_cb(at, length); + }; + auto on_message_complete_cb = cbs_->on_message_complete; + cbs_->on_message_complete = [this, on_message_complete_cb](const Response){ + if (on_message_complete_cb) + on_message_complete_cb(response_); + }; + + // http_parser raw c callback (note: no context can be passed into them) + parser_s_->on_status = [](http_parser* parser, const char* /*at*/, size_t /*length*/) -> int { + auto cbs = static_cast(parser->data); + if (cbs->on_status) + cbs->on_status(parser->status_code); + return 0; + }; + parser_s_->on_header_field = [](http_parser* parser, const char* at, size_t length) -> int { + auto cbs = static_cast(parser->data); + if (cbs->on_header_field) + cbs->on_header_field(at, length); + return 0; + }; + parser_s_->on_header_value = [](http_parser* parser, const char* at, size_t length) -> int { + auto cbs = static_cast(parser->data); + if (cbs->on_header_value) + cbs->on_header_value(at, length); + return 0; + }; + parser_s_->on_body = [](http_parser* parser, const char* at, size_t length) -> int { + auto cbs = static_cast(parser->data); + if (cbs->on_body) + cbs->on_body(at, length); + return 0; + }; + parser_s_->on_message_complete = [](http_parser* parser) -> int { + auto cbs = static_cast(parser->data); + if (cbs->on_message_complete) + cbs->on_message_complete(Response{}); + return 0; + }; +} + +void +Request::connect(asio::ip::basic_resolver_results&& endpoints, HandlerCb cb) +{ + if (endpoints.empty()){ if (logger_) - logger_->e("[http::client] host not resolved, can't send request"); + logger_->e("[http:request:%i] [connect] no endpoints provided", id_); if (cb) - cb(nullptr); + cb(asio::error::connection_aborted); return; } - auto conn = std::make_shared(connId_++, - std::move(asio::ip::tcp::socket{resolver_.get_io_context()})); if (logger_) - logger_->d("[http::client] [connection:%i] created", conn->id()); + logger_->d("[http:request:%i] [connect] begin", id_); + + conn_ = std::make_shared(ctx_); // 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 + asio::async_connect(conn_->socket(), std::move(endpoints), [this, cb] + (const asio::error_code& ec, const asio::ip::tcp::endpoint& endpoint){ if (ec){ 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 with all endpoints", id_); + } + else { + if (logger_) + logger_->d("[http:request:%i] [connect] success", id_); + // save the associated endpoint + conn_->set_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); }); } void -Client::async_request(std::shared_ptr conn, std::string request, - std::unique_ptr parser, - std::unique_ptr parser_s, HandlerCb cb) +Request::send() { - if (endpoints_.empty()){ - if (logger_) - logger_->e("[http::client] host not resolved, can't send request"); - if (cb) - cb(asio::error::addrinfo_errors::service_not_found); - return; - } - if (!conn){ - if (logger_) - logger_->e("[http::client] invalid connection, can't post request:\n%s", request.c_str()); - if (cb) - cb(asio::error::not_connected); - return; - } - else if (!conn->is_open()){ - if (logger_) - logger_->e("[http::client] closed connection, can't post request"); - close_connection(conn->id()); - 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); + notify_state_changed(State::CREATED); - // 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)); + resolver_->add_callback([this](const asio::error_code& ec, + asio::ip::tcp::resolver::results_type endpoints){ + if (ec){ + if (logger_) + logger_->e("[http:request:%i] [send] resolve error: %s", id_, ec.message().c_str()); + terminate(asio::error::connection_aborted); + } + else if (!conn_ or !conn_->is_open()){ + connect(std::move(endpoints), [this](const asio::error_code &ec){ + if (!ec) + post(); + else + terminate(asio::error::not_connected); + }); + } + else { + post(); + } + }); } void -Client::handle_request(const asio::error_code& ec, std::shared_ptr conn, HandlerCb cb) +Request::post() { - if (!conn->is_open()){ + if (!conn_ or !conn_->is_open()){ if (logger_) - logger_->e("[http::client] closed connection, can't handle request"); - close_connection(conn->id()); - if (cb) - cb(asio::error::not_connected); + logger_->e("[http:request:%i] [post] closed connection", id_); + terminate(asio::error::not_connected); + return; + } + build(); + init_parser(); + + if (logger_) + logger_->d("[http:request:%i] [post]\n%s", id_, request_.c_str()); + + // write the request to buffer + std::ostream request_stream(&conn_->input()); + request_stream << request_; + + // send the request + notify_state_changed(State::SENDING); + asio::async_write(conn_->socket(), conn_->input(), + std::bind(&Request::handle_request, this, std::placeholders::_1)); +} + +void +Request::terminate(const asio::error_code& ec) +{ + // set response outcome, ignore enf of file and abort + if (!ec or ec == asio::error::eof or ec == asio::error::operation_aborted) + response_.status_code = 200; + else + response_.status_code = 0; + + // ensure on_message_complete callback is called + if (parser_){ + std::lock_guard lock(cbs_mutex_); + http_parser_execute(parser_.get(), parser_s_.get(), "", 0); + } + if (logger_) + logger_->d("[http:request:%i] done", id_); + + notify_state_changed(State::DONE); +} + +void +Request::handle_request(const asio::error_code& ec) +{ + if (!conn_->is_open()){ + if (logger_) + logger_->e("[http:request:%i] [write] closed connection", id_); + terminate(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()); - if (cb) - cb(ec); + logger_->e("[http:request:%i] [write] error: %s", id_, ec.message().c_str()); + terminate(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)); + notify_state_changed(State::RECEIVING); + asio::async_read_until(conn_->socket(), conn_->data(), "\r\n\r\n", + std::bind(&Request::handle_response_header, this, + std::placeholders::_1, std::placeholders::_2)); } void -Client::handle_response_header(const asio::error_code& ec, const size_t bytes, - std::shared_ptr conn, HandlerCb cb) +Request::handle_response_header(const asio::error_code& ec, const size_t bytes) { - 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()); - if (cb) - cb(asio::error::not_connected); + logger_->e("[http:request:%i] [read:header] closed connection", id_); + terminate(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()); - if (cb) - cb(ec); + logger_->e("[http:request:%i] [read:header] error: %s", id_, ec.message().c_str()); + terminate(ec); return; } else if ((ec == asio::error::eof) || (ec == asio::error::connection_reset)){ - close_connection(conn->id()); - if (cb) - cb(ec); + terminate(ec); return; } // read the response buffer std::ostringstream str_s; - str_s << &conn->response_chunk_; + str_s << &conn_->data(); auto chunk = str_s.str(); - auto header = chunk.substr(0, bytes); + auto headers = 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_, headers.c_str()); // parse the header right away - parse_request(header, conn->id()); + parse_request(headers); + notify_state_changed(State::HEADER_RECEIVED); + + auto content_length_it = response_.headers.find(HTTP_HEADER_CONTENT_LENGTH); + auto content_type_it = response_.headers.find(HTTP_HEADER_CONTENT_TYPE); - unsigned int content_length = get_content_length(header); // has body size - if (content_length){ - // append current body chunk + if (content_length_it != response_.headers.end()){ + unsigned int content_length = atoi(content_length_it->second.c_str()); auto body_chunk = chunk.substr(bytes, std::string::npos); - conn->response_body_.append(body_chunk); - // read the rest of body - asio::async_read(conn->socket_, conn->response_chunk_, - asio::transfer_exactly(content_length - (body_chunk.size())), - std::bind(&Client::handle_response_body, this, - std::placeholders::_1, content_length, conn, cb)); + // append current body chunk + if (content_length > body_chunk.size()){ + // read the rest of body + notify_state_changed(State::RECEIVING); + asio::async_read(conn_->socket(), conn_->data(), + asio::transfer_exactly(content_length - (body_chunk.size())), + std::bind(&Request::handle_response_body, this, + std::placeholders::_1, content_length, body_chunk)); + } + else + terminate(asio::error::eof); return; } // has potential body but no content-length (current proxy behavior on key get/listen) - else if (header.find("Content-Type: application/json") != std::string::npos){ + else if (content_type_it != response_.headers.end() and + content_type_it->second == HTTP_HEADER_CONTENT_TYPE_JSON) + { auto body_chunk = chunk.substr(bytes, std::string::npos); - if (!body_chunk.empty()){ - parse_request(body_chunk, conn->id()); - } + if (!body_chunk.empty()) + parse_request(body_chunk); // keep reading - asio::async_read(conn->socket_, conn->response_chunk_, asio::transfer_at_least(1), - std::bind(&Client::handle_response_body, this, - std::placeholders::_1, std::placeholders::_2, conn, cb)); + notify_state_changed(State::RECEIVING); + asio::async_read(conn_->socket(), conn_->data(), asio::transfer_at_least(1), + std::bind(&Request::handle_response_body, this, + std::placeholders::_1, std::placeholders::_2, "")); } + /* NOTE: + * We ignore keep-alive from server since the current dhtproxy sets it on every operation + * thus, it wouldn't make sense to continue reading from a socket even if the dhtproxy + * terminated its operation. + */ } void -Client::handle_response_body(const asio::error_code& ec, const size_t bytes, - std::shared_ptr conn, HandlerCb cb) +Request::handle_response_body(const asio::error_code& ec, const size_t bytes, const std::string chunk) { - 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()); - if (cb) - cb(asio::error::not_connected); + logger_->e("[http:request] [read:body] closed connection"); + terminate(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()); - if (cb) - cb(ec); + logger_->e("[http:request:%i] [read:body] error: %s", id_, ec.message().c_str()); + terminate(ec); return; } else if ((ec == asio::error::eof) || (ec == asio::error::connection_reset)){ - close_connection(conn->id()); - if (cb) - cb(ec); + terminate(ec); return; } std::string body; - if (!conn->response_body_.empty()){ + if (!chunk.empty()){ // append previous incomplete chunk from header - body.append(conn->response_body_); + body.append(chunk); } // read the response buffer std::ostringstream str_s; - str_s << &conn->response_chunk_; + str_s << &conn_->data(); body.append(str_s.str().substr(0, bytes)); + response_.body = body; 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, - std::placeholders::_1, std::placeholders::_2, conn, cb)); + logger_->d("[http:request:%i] [read:body] success:\n%s", id_, body.c_str()); + + parse_request(body); + + asio::async_read(conn_->socket(), conn_->data(), asio::transfer_at_least(1), + std::bind(&Request::handle_response_body, this, std::placeholders::_1, std::placeholders::_2, "")); } 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()); - // detect parsing errors - if (HPE_OK != conn_context.parser->http_errno && HPE_PAUSED != conn_context.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)); - } - } -} + std::lock_guard lock(cbs_mutex_); + http_parser_execute(parser_.get(), parser_s_.get(), request.c_str(), request.size()); -size_t -Client::get_content_length(const std::string str, const ConnectionId conn_id) -{ - size_t content_length = 0; - auto content_length_i = str.find(HTTP_HEADER_CONTENT_LENGTH); - if (content_length_i != std::string::npos){ - auto content_length_str = str.substr(content_length_i + std::strlen(HTTP_HEADER_CONTENT_LENGTH) + 1, - str.find("\r\n", content_length_i)); - try { - content_length = atoi(content_length_str.c_str()); - } - 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()); + // detect parsing errors + if (HPE_OK != parser_->http_errno && HPE_PAUSED != parser_->http_errno){ + if (logger_){ + auto err = HTTP_PARSER_ERRNO(parser_.get()); + logger_->e("[http:request:%i] [parse] error: %s", id_, http_errno_name(err)); } } - return content_length; } } // namespace http