diff --git a/CMakeLists.txt b/CMakeLists.txt index d55a6363..e810da75 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -163,6 +163,7 @@ list (APPEND opendht_SOURCES src/network_utils.h src/network_utils.cpp src/thread_pool.cpp + src/http.cpp ) list (APPEND opendht_HEADERS @@ -188,6 +189,7 @@ list (APPEND opendht_HEADERS include/opendht/log_enable.h include/opendht/peer_discovery.h include/opendht/thread_pool.h + include/opendht/http.h include/opendht.h ) diff --git a/include/opendht/dht_proxy_client.h b/include/opendht/dht_proxy_client.h index c48b6913..c6767005 100644 --- a/include/opendht/dht_proxy_client.h +++ b/include/opendht/dht_proxy_client.h @@ -34,29 +34,12 @@ #include #include #include +#include "http.h" #include #include #include -// nested declaration only in c++17 -namespace restinio { namespace client { - - template - void do_with_socket(LAMBDA && lambda, const std::string & addr, std::uint16_t port, - restinio::asio_ns::io_context &io_context); - - void do_request(const std::string& request, const std::string& addr, std::uint16_t port, - http_parser &parser, http_parser_settings &settings, - std::shared_ptr io_context = nullptr, - const std::function conn_cb = nullptr); - - std::string create_http_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); -}} - namespace restbed { class Request; } @@ -86,6 +69,11 @@ public: virtual ~DhtProxyClient(); + /** + * Get Asio I/O Context. + */ + std::shared_ptr context(); + /** * Get the ID of the node. */ @@ -336,6 +324,12 @@ private: uint16_t serverHostPort_; std::string pushClientId_; + /* + * ASIO I/O Context for sockets used in requests + */ + std::shared_ptr ctx_; + http::Client httpClient_ {serverHostIp_, serverHostPort_}; + mutable std::mutex lockCurrentProxyInfos_; NodeStatus statusIpv4_ {NodeStatus::Disconnected}; NodeStatus statusIpv6_ {NodeStatus::Disconnected}; @@ -369,15 +363,6 @@ private: std::thread statusThread_; mutable std::mutex statusLock_; - /* - * ASIO I/O Context for sockets used in requests - * https://www.boost.org/doc/libs/develop/doc/html/boost_asio/reference/io_context.html - */ - std::shared_ptr io_context; - std::thread io_context_thread_; - - std::vector listeners; - Scheduler scheduler; /** * Retrieve if we can connect to the proxy (update statusIpvX_) diff --git a/include/opendht/dht_proxy_server.h b/include/opendht/dht_proxy_server.h index c462a34c..618bffa5 100644 --- a/include/opendht/dht_proxy_server.h +++ b/include/opendht/dht_proxy_server.h @@ -33,6 +33,7 @@ #include #include #include +#include "http.h" #ifdef OPENDHT_JSONCPP #include @@ -328,7 +329,6 @@ private: */ void cancelPushListen(const std::string& pushToken, const InfoHash& key, const std::string& clientId); - #endif //OPENDHT_PUSH_NOTIFICATIONS using clock = std::chrono::steady_clock; diff --git a/include/opendht/http.h b/include/opendht/http.h new file mode 100644 index 00000000..fb0c0696 --- /dev/null +++ b/include/opendht/http.h @@ -0,0 +1,77 @@ +/* + * Copyright (C) 2016-2019 Savoir-faire Linux Inc. + * Author: Vsevolod Ivanov + * + * This program is free software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation; either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + */ + +#pragma once + +#include +#include +#include +#include + +namespace http { + +class Connection +{ + public: + Connection(const uint16_t id, asio::ip::tcp::socket socket); + ~Connection(); + + uint16_t id(); + void start(asio::ip::tcp::resolver::iterator &r_iter); + bool is_open(); + std::string read(std::error_code& ec); + void write(std::string request, std::error_code& ec); + void close(); + + private: + uint16_t id_; + asio::ip::tcp::socket socket_; +}; + +using ResponseCallback = std::function; + +class Client +{ + public: + Client(std::string ip, std::uint16_t port); + + asio::io_context& context(); + + asio::ip::tcp::resolver::query query(); + + 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 post_request(std::string request, http_parser_settings parser_s, + const ResponseCallback respcb = nullptr); + + private: + void async_request(std::string request, http_parser_settings parser_s, + const ResponseCallback respcb = nullptr); + + std::uint16_t port_; + asio::ip::address addr_; + asio::io_context ctx_; + asio::ip::tcp::resolver resolver_; + uint16_t connId_ {1}; +}; + +} diff --git a/src/dht_proxy_client.cpp b/src/dht_proxy_client.cpp index ca7d3e8e..57482182 100644 --- a/src/dht_proxy_client.cpp +++ b/src/dht_proxy_client.cpp @@ -23,96 +23,6 @@ #include "op_cache.h" #include "utils.h" -namespace restinio { namespace client { - -template -void do_with_socket(LAMBDA && lambda, const std::string &ip, std::uint16_t port, - restinio::asio_ns::io_context &io_context){ - restinio::asio_ns::ip::tcp::socket socket{io_context}; - restinio::asio_ns::ip::tcp::resolver resolver{io_context}; - - auto addr = restinio::asio_ns::ip::address::from_string(ip); - auto addrType = addr.is_v4() ? restinio::asio_ns::ip::tcp::v4() : restinio::asio_ns::ip::tcp::v6(); - restinio::asio_ns::ip::tcp::resolver::query query{addrType, addr.to_string(), std::to_string(port)}; - restinio::asio_ns::ip::tcp::resolver::iterator iterator = resolver.resolve(query); - - restinio::asio_ns::connect(socket, iterator); - lambda(socket); - socket.close(); -} - -void -do_request(const std::string & request, const std::string &addr, std::uint16_t port, - http_parser &parser, http_parser_settings &settings, - std::shared_ptr io_context, - const std::function conn_cb){ - if (!io_context){ - printf("Client request without io_context, making one for the request\n"); - io_context = std::make_shared(); - } - do_with_socket([&](auto & socket){ - // callback with the active connection - if (conn_cb) - conn_cb(socket); - // write request - restinio::asio_ns::streambuf b; - std::ostream req_stream(&b); - req_stream << request; - restinio::asio_ns::write(socket, b); - // read response - http_parser_init(&parser, HTTP_RESPONSE); - restinio::asio_ns::error_code error; - restinio::asio_ns::streambuf response_stream; - while(restinio::asio_ns::read(socket, response_stream, - restinio::asio_ns::transfer_at_least(1), error)){ - std::ostringstream sout; - sout << &response_stream; - //printf("Got server response:\n%s\n", sout.str().c_str());// DHT_LOG.w ? - http_parser_execute(&parser, &settings, sout.str().c_str(), sout.str().size()); - if (HPE_OK != parser.http_errno && HPE_PAUSED != parser.http_errno){ - auto err = HTTP_PARSER_ERRNO(&parser); - std::cerr << "Couldn't parse the response: " << http_errno_name(err) << std::endl; - } - } - if (!restinio::error_is_eof(error)) - throw std::runtime_error{fmt::format("Client read error: {}", error)}; - }, addr, port, *io_context); -} - -std::string -create_http_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::stringstream request; - request << restinio::method_to_string(header.method()) << " " << - header.request_target() << " " << - "HTTP/" << header.http_major() << "." << header.http_minor() << "\r\n"; - for (auto header_field: header_fields) - request << header_field.name() << ": " << header_field.value() << "\r\n"; - std::string conn_str; - switch (connection){ - case restinio::http_connection_header_t::keep_alive: - conn_str = "keep-alive"; - break; - case restinio::http_connection_header_t::close: - conn_str = "close"; - break; - case restinio::http_connection_header_t::upgrade: - throw std::invalid_argument("upgrade"); - break; - } - request << "Connection: " << conn_str << "\r\n"; - if (!body.empty()){ - request << "Content-Length: " << body.size() << "\r\n\r\n"; - request << body; - } - request << "\r\n"; - return request.str(); -} - -}} - namespace dht { struct DhtProxyClient::InfoState { @@ -157,7 +67,7 @@ struct DhtProxyClient::ProxySearch { DhtProxyClient::DhtProxyClient() {} DhtProxyClient::DhtProxyClient(std::function signal, const std::string& serverHost, const std::string& pushClientId, const Logger& l): - serverHost_(serverHost), pushClientId_(pushClientId), loopSignal_(signal), io_context(new restinio::asio_ns::io_context()) + serverHost_(serverHost), pushClientId_(pushClientId), loopSignal_(signal), ctx_(new restinio::asio_ns::io_context()) { auto hostAndPort = splitPort(serverHost_); this->serverHostIp_ = hostAndPort.first; @@ -167,8 +77,11 @@ DhtProxyClient::DhtProxyClient(std::function signal, const std::string& serverHost_ = proxy::HTTP_PROTO + serverHost_; if (!serverHost_.empty()) startProxy(); +} - io_context_thread_ = std::thread([this](){io_context->run();}); +std::shared_ptr +DhtProxyClient::context(){ + return ctx_; } void @@ -197,8 +110,6 @@ DhtProxyClient::~DhtProxyClient() infoState_->cancel = true; if (statusThread_.joinable()) statusThread_.join(); - if (io_context_thread_.joinable()) - io_context_thread_.join(); } std::vector> @@ -223,7 +134,7 @@ void DhtProxyClient::cancelAllOperations() { // for graceful finish, call reset() before - this->io_context->stop(); + this->ctx_->stop(); } void @@ -328,7 +239,7 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Va header.request_target("/" + key.toString()); header.method(restinio::http_method_t::http_get); auto header_fields = this->initHeaderFields(); - auto request = restinio::client::create_http_request(header, header_fields, + auto request = httpClient_.create_request(header, header_fields, restinio::http_connection_header_t::keep_alive, ""/*body*/); DHT_LOG.w(request.c_str()); @@ -389,8 +300,8 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Va } return 0; }; - restinio::client::do_request(request, serverHostIp_, serverHostPort_, - parser, settings, this->io_context); + //httpClient_.do_request(request, serverHostIp_, serverHostPort_, + // parser, settings, this->ctx_); } void @@ -461,7 +372,7 @@ DhtProxyClient::doPut(const InfoHash& key, Sp val, DoneCallback cb, time_ wbuilder["commentStyle"] = "None"; wbuilder["indentation"] = ""; auto body = Json::writeString(wbuilder, json); - auto request = restinio::client::create_http_request(header, header_fields, + auto request = httpClient_.create_request(header, header_fields, restinio::http_connection_header_t::close, body); // TODO dhtlog.w printf("%s\n", request.c_str()); @@ -497,8 +408,8 @@ DhtProxyClient::doPut(const InfoHash& key, Sp val, DoneCallback cb, time_ } return 0; }; - restinio::client::do_request(request, serverHostIp_, serverHostPort_, - parser, settings, this->io_context); + //do_request(request, serverHostIp_, serverHostPort_, + // parser, settings, this->ctx_); } /** @@ -833,7 +744,7 @@ void DhtProxyClient::sendListen(const restinio::http_request_header_t header, if (method != ListenMethod::LISTEN) body = fillBody(method == ListenMethod::RESUBSCRIBE); #endif - auto request = restinio::client::create_http_request(header, headers, conn, body); + auto request = httpClient_.create_request(header, headers, conn, body); /*DHT_LOG.w*/printf(request.c_str()); struct ListenContext { @@ -896,8 +807,8 @@ void DhtProxyClient::sendListen(const restinio::http_request_header_t header, } return 0; }; - restinio::client::do_request(request, serverHostIp_, serverHostPort_, - parser, settings, this->io_context); + //do_request(request, serverHostIp_, serverHostPort_, + // parser, settings, this->ctx_); } bool diff --git a/src/dht_proxy_server.cpp b/src/dht_proxy_server.cpp index 7696584a..4b63d024 100644 --- a/src/dht_proxy_server.cpp +++ b/src/dht_proxy_server.cpp @@ -51,7 +51,7 @@ constexpr const std::chrono::minutes PRINT_STATS_PERIOD {2}; constexpr const size_t IO_THREADS_MAX {64}; -DhtProxyServer::DhtProxyServer(std::shared_ptr dht, in_port_t port , const std::string& pushServer, std::shared_ptr logger) +DhtProxyServer::DhtProxyServer(std::shared_ptr dht, in_port_t port ,const std::string& pushServer, std::shared_ptr logger) : dht_(dht), threadPool_(new ThreadPool(IO_THREADS_MAX)), pushServer_(pushServer) { if (not dht_) @@ -576,14 +576,15 @@ DhtProxyServer::sendPushNotification(const std::string& token, Json::Value&& jso { if (pushServer_.empty()) return; - http_parser_settings settings; - http_parser_settings_init(&settings); - settings.on_status = []( http_parser * parser, const char * at, size_t length ) -> int { + http_parser_settings parser_s; + http_parser_settings_init(&parser_s); + parser_s.on_status = []( http_parser * parser, const char * at, size_t length ) -> int { if (parser->status_code == 200) return 0; std::cerr << "Error in SendPushNotification status_code=" << parser->status_code << std::endl; return 1; }; + restinio::http_request_header_t header; header.request_target("/api/push"); header.method(restinio::http_method_t::http_post); @@ -615,16 +616,14 @@ DhtProxyServer::sendPushNotification(const std::string& token, Json::Value&& jso wbuilder["indentation"] = ""; auto body = Json::writeString(wbuilder, content); - auto request = restinio::client::create_http_request( - header, header_fields, restinio::http_connection_header_t::close, body); - - http_parser parser; auto hostAndPort = splitPort(pushServer_); uint16_t port = std::atoi(hostAndPort.second.c_str()); - // if other functions use the client move into private member to share the same - //auto io_context = std::make_shared(); - //std::thread([io_context](){io_context->run();}); - restinio::client::do_request(request, hostAndPort.first, port, parser, settings); + http::Client httpClient {hostAndPort.first, port}; + + auto request = httpClient.create_request(header, header_fields, + restinio::http_connection_header_t::close, body); + httpClient.post_request(request, parser_s); + httpClient.context().run(); } #endif //OPENDHT_PUSH_NOTIFICATIONS diff --git a/src/http.cpp b/src/http.cpp new file mode 100644 index 00000000..87716671 --- /dev/null +++ b/src/http.cpp @@ -0,0 +1,186 @@ +/* + * Copyright (C) 2016-2019 Savoir-faire Linux Inc. + * Author: Vsevolod Ivanov + * + * This program is free software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation; either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + */ + +#include "http.h" + +namespace http { + +// connection + +Connection::Connection(const uint16_t id, asio::ip::tcp::socket socket): + id_(id), socket_(std::move(socket)) +{} + +Connection::~Connection(){ + close(); +} + +uint16_t +Connection::id(){ + return id_; +} + +void +Connection::start(asio::ip::tcp::resolver::iterator &r_iter){ + asio::connect(socket_, r_iter); +} + +bool +Connection::is_open(){ + return socket_.is_open(); +} + +std::string +Connection::read(std::error_code& ec){ + std::string response; + asio::read(socket_, asio::dynamic_buffer(response), ec); + return response; +} + +void +Connection::write(std::string request, std::error_code& ec){ + asio::write(socket_, asio::dynamic_buffer(request), ec); +} + +void +Connection::close(){ + socket_.close(); +} + +// client + +Client::Client(std::string ip, std::uint16_t port): resolver_(ctx_){ + addr_ = asio::ip::address::from_string(ip); + port_ = port; +} + +asio::io_context& +Client::context(){ + return ctx_; +} + +asio::ip::tcp::resolver::query +Client::query(){ + return asio::ip::tcp::resolver::query { + addr_.is_v4() ? asio::ip::tcp::v4() : asio::ip::tcp::v6(), + addr_.to_string(), std::to_string(port_)}; +} + +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){ + std::stringstream request; + request << restinio::method_to_string(header.method()) << " " << + header.request_target() << " " << + "HTTP/" << header.http_major() << "." << header.http_minor() << "\r\n"; + for (auto header_field: header_fields) + request << header_field.name() << ": " << header_field.value() << "\r\n"; + std::string conn_str; + switch (connection){ + case restinio::http_connection_header_t::keep_alive: + conn_str = "keep-alive"; + break; + case restinio::http_connection_header_t::close: + conn_str = "close"; + break; + case restinio::http_connection_header_t::upgrade: + throw std::invalid_argument("upgrade"); + break; + } + request << "Connection: " << conn_str << "\r\n"; + if (!body.empty()){ + request << "Content-Length: " << body.size() << "\r\n\r\n"; + request << body; + } + request << "\r\n"; + return request.str(); +} + +void +Client::post_request(std::string request, http_parser_settings parser_s, + const ResponseCallback respcb){ + // invoke the given handler and return immediately + asio::post(ctx_, [this, request, parser_s, respcb](){ + this->async_request(request, parser_s, respcb); + // execute at most one handler, it ensures that same func call + // with different callback gets the priority on the io_context + ctx_.run_one(); + }); +} + +void +Client::async_request(std::string request, http_parser_settings parser_s, + const ResponseCallback respcb){ + using namespace asio::ip; + + auto conn = std::make_shared(connId_, std::move(tcp::socket{ctx_})); + printf("[connection:%i] created\n", conn->id()); + connId_++; + + // resolve sometime in future + resolver_.async_resolve(query(), [=](std::error_code ec, tcp::resolver::results_type res){ + if (ec or res.empty()){ + printf("[connection:%i] error resolving\n", conn->id()); + conn->close(); + return; + } + for (auto da = res.begin(); da != res.end(); ++da){ + printf("[connection:%i] resolved host=%s service=%s\n", + conn->id(), da->host_name().c_str(), da->service_name().c_str()); + conn->start(da); + break; + } + if (!conn->is_open()){ + printf("[connection:%i] error closed connection\n", conn->id()); + return; + } + // send request + printf("[connection:%i] request write\n", conn->id()); + conn->write(request, ec); + if (ec and ec != asio::error::eof){ + printf("[connection:%i] error: %s\n", conn->id(), ec.message().c_str()); + return; + } + // read response + printf("[connection:%i] response read\n", conn->id()); + auto data = conn->read(ec); + if (ec and ec != asio::error::eof){ + printf("[connection:%i] error: %s\n", conn->id(), ec.message().c_str()); + return; + } + if (respcb) + respcb(data.c_str()); + + // parse the request + http_parser parser; + + http_parser_settings_init(const_cast(&parser_s)); + http_parser_init(&parser, HTTP_RESPONSE); + + http_parser_execute(&parser, &parser_s, data.c_str(), data.size()); + + if (HPE_OK != parser.http_errno && HPE_PAUSED != parser.http_errno){ + auto err = HTTP_PARSER_ERRNO(&parser); + printf("[connection:%i] error parsing: %s\n", conn->id(), http_errno_name(err)); + } + }); +} + +}