diff --git a/include/opendht/dht_proxy_client.h b/include/opendht/dht_proxy_client.h index fc74391e..fe0f093a 100644 --- a/include/opendht/dht_proxy_client.h +++ b/include/opendht/dht_proxy_client.h @@ -43,6 +43,10 @@ namespace Json { class Value; } +namespace http { + class Client; +} + namespace dht { class OPENDHT_PUBLIC DhtProxyClient final : public DhtInterface { @@ -323,7 +327,7 @@ private: /* * ASIO I/O Context for sockets used in requests */ - http::Client httpClient_; + std::unique_ptr httpClient_; mutable std::mutex lockCurrentProxyInfos_; NodeStatus statusIpv4_ {NodeStatus::Disconnected}; diff --git a/include/opendht/dht_proxy_server.h b/include/opendht/dht_proxy_server.h index c411eaec..c40690e6 100644 --- a/include/opendht/dht_proxy_server.h +++ b/include/opendht/dht_proxy_server.h @@ -75,6 +75,12 @@ private: std::shared_ptr m_logger; }; +namespace http { + class ConnectionListener; + struct SessionToHashToken; + struct custom_http_methods_t; +} + using RestRouter = restinio::router::express_router_t<>; struct RestRouterTraits : public restinio::default_traits_t { @@ -82,6 +88,7 @@ struct RestRouterTraits : public restinio::default_traits_t using http_methods_mapper_t = http::custom_http_methods_t; using logger_t = opendht_logger_t; using request_handler_t = RestRouter; + using connection_state_listener_t = http::ConnectionListener; }; using ServerSettings = restinio::run_on_thread_pool_settings_t; using RequestStatus = restinio::request_handling_status_t; @@ -111,7 +118,9 @@ public: * @note if the server fails to start (if port is already used or reserved), * it will fails silently */ - DhtProxyServer(std::shared_ptr dht, in_port_t port = 8000, const std::string& pushServer = "", std::shared_ptr logger = nullptr); + DhtProxyServer(std::shared_ptr dht, in_port_t port = 8000, + const std::string& pushServer = "", + std::shared_ptr logger = nullptr); virtual ~DhtProxyServer(); DhtProxyServer(const DhtProxyServer& other) = delete; @@ -342,7 +351,7 @@ private: std::shared_ptr logger_; Json::StreamWriterBuilder jsonBuilder_; - restinio::http_server_t httpServer_; + std::unique_ptr> httpServer_; std::thread httpServerThread_ {}; std::mutex schedulerLock_; @@ -357,14 +366,10 @@ private: // Handle client quit for listen. std::thread listenThread_; - struct SessionToHashToken { - InfoHash hash; - restinio::connection_id_t connId; - std::future token; - }; - std::vector currentListeners_; + std::vector currentListeners_; std::mutex lockListener_; std::atomic_bool stopListeners {false}; + std::shared_ptr connListener_; struct PermanentPut; struct SearchPuts; diff --git a/include/opendht/http.h b/include/opendht/http.h index 49221fe1..69698f55 100644 --- a/include/opendht/http.h +++ b/include/opendht/http.h @@ -22,6 +22,7 @@ #include #include #include +#include #include namespace http { @@ -45,7 +46,33 @@ private: asio::ip::tcp::socket socket_; }; -using ResponseCallback = std::function; +struct SessionToHashToken { + dht::InfoHash hash; + restinio::connection_id_t connId; + std::future token; +}; + +class ConnectionListener +{ +public: + ConnectionListener(); + ConnectionListener(std::vector *listeners, + std::mutex *lock, std::shared_ptr logger); + ~ConnectionListener(); + + /** + * Connection state change used to handle Listeners disconnects. + * RESTinio >= 0.5.1 https://github.com/Stiffstream/restinio/issues/28 + */ + void state_changed(const restinio::connection_state::notice_t ¬ice) noexcept; + +private: + std::string to_str( restinio::connection_state::cause_t cause ) noexcept; + + std::mutex *lock_; + std::vector *listeners_; + std::shared_ptr logger_; +}; class Client { diff --git a/src/dht_proxy_client.cpp b/src/dht_proxy_client.cpp index fc933c7b..39fa67a9 100644 --- a/src/dht_proxy_client.cpp +++ b/src/dht_proxy_client.cpp @@ -73,8 +73,8 @@ DhtProxyClient::DhtProxyClient(std::function signal, const std::string& serverHostIp_ = hostAndPort.first; serverHostPort_ = std::atoi(hostAndPort.second.c_str()); - httpClient_.set_query_address(serverHostIp_, serverHostPort_); - httpClient_.set_logger(logger); + httpClient_->set_query_address(serverHostIp_, serverHostPort_); + httpClient_->set_logger(logger); if (serverHost_.find("://") == std::string::npos) serverHost_ = proxy::HTTP_PROTO + serverHost_; @@ -85,7 +85,7 @@ DhtProxyClient::DhtProxyClient(std::function signal, const std::string& asio::io_context& DhtProxyClient::httpIOContext(){ - return httpClient_.io_context(); + return httpClient_->io_context(); } void @@ -251,7 +251,7 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Va header.request_target("/" + key.toString()); header.method(restinio::http_method_get()); auto header_fields = this->initHeaderFields(); - auto request = httpClient_.create_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()); @@ -318,7 +318,7 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Va } return 0; }; - httpClient_.post_request(request, parser, parser_s); + httpClient_->post_request(request, parser, parser_s); } void @@ -389,7 +389,7 @@ 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, + auto request = httpClient_->create_request(header, header_fields, restinio::http_connection_header_t::close, body); DHT_LOG.d("%s", request.c_str()); @@ -429,7 +429,7 @@ DhtProxyClient::doPut(const InfoHash& key, Sp val, DoneCallback cb, time_ } return 0; }; - httpClient_.post_request(request, parser, parser_s); + httpClient_->post_request(request, parser, parser_s); } /** @@ -760,7 +760,7 @@ void DhtProxyClient::sendListen(const restinio::http_request_header_t header, if (method != ListenMethod::LISTEN) body = fillBody(method == ListenMethod::RESUBSCRIBE); #endif - auto request = httpClient_.create_request(header, headers, conn, body); + auto request = httpClient_->create_request(header, headers, conn, body); DHT_LOG.d(request.c_str()); struct ListenContext { @@ -821,7 +821,7 @@ void DhtProxyClient::sendListen(const restinio::http_request_header_t header, } return 0; }; - httpClient_.post_request(request, parser, parser_s); + httpClient_->post_request(request, parser, parser_s); } bool diff --git a/src/dht_proxy_server.cpp b/src/dht_proxy_server.cpp index ce05638e..194bd244 100644 --- a/src/dht_proxy_server.cpp +++ b/src/dht_proxy_server.cpp @@ -50,12 +50,13 @@ struct DhtProxyServer::SearchPuts { 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 , +DhtProxyServer::DhtProxyServer(std::shared_ptr dht, in_port_t port, const std::string& pushServer, std::shared_ptr logger): dht_(dht), logger_(logger), threadPool_(new ThreadPool(IO_THREADS_MAX)), - pushServer_(pushServer), - httpServer_(restinio::own_io_context(), []( auto & settings ){}) + connListener_(std::make_shared( + ¤tListeners_, &lockListener_, logger)), + pushServer_(pushServer) { if (not dht_) throw std::invalid_argument("A DHT instance must be provided"); @@ -90,15 +91,16 @@ DhtProxyServer::DhtProxyServer(std::shared_ptr dht, in_port_t port , settings.socket_options_setter([](auto & options){ options.set_option(asio::ip::tcp::no_delay{true}); }); - httpServer_ = restinio::http_server_t( + settings.connection_state_listener(connListener_); + httpServer_.reset(new restinio::http_server_t( restinio::own_io_context(), // requirement: each thread has its own std::forward(settings) - ); - restinio::asio_ns::post(httpServer_.io_context(), [&]{ - httpServer_.open_sync(); + )); + restinio::asio_ns::post(httpServer_->io_context(), [&]{ + httpServer_->open_sync(); }); try { - httpServer_.io_context().run(); + httpServer_->io_context().run(); } catch(const std::exception &ex){ std::cerr << "Error starting RESTinio: " << ex.what() << std::endl; @@ -164,7 +166,7 @@ DhtProxyServer::stop() schedulerCv_.notify_all(); // stop its own io_context loop - httpServer_.close_sync(); + httpServer_->close_sync(); // listenThreads_ will stop because there is no more sessions if (listenThread_.joinable()) @@ -356,7 +358,7 @@ DhtProxyServer::listen(restinio::request_handle_t request, this->initHttpResponse(request->create_response())); response->flush(); try { - SessionToHashToken listener; + http::SessionToHashToken listener; listener.hash = infoHash; listener.connId = request->connection_id(); listener.token = dht_->listen(infoHash, [this, response] diff --git a/src/http.cpp b/src/http.cpp index c52095ff..72707f4c 100644 --- a/src/http.cpp +++ b/src/http.cpp @@ -67,6 +67,54 @@ Connection::close(){ socket_.close(); } +// connection listener + +ConnectionListener::ConnectionListener() +{} + +ConnectionListener::ConnectionListener(std::vector *listeners, + std::mutex *lock, std::shared_ptr logger): + listeners_(listeners), lock_(lock), logger_(logger) +{} + +ConnectionListener::~ConnectionListener() +{} + +void +ConnectionListener::state_changed(const restinio::connection_state::notice_t ¬ice) noexcept +{ + printf("[restinio] [connection:%li] listener is %s\n", + notice.connection_id(), to_str(notice.cause()).c_str()); + + std::lock_guard lock(*lock_); + printf("[restinio] [connection:%li] searching listeners\n", notice.connection_id()); + + //auto listener = listeners_->begin(); + //while (listener != currentListeners_->end()){ + //} +} + +std::string +ConnectionListener::to_str(restinio::connection_state::cause_t cause) noexcept +{ + std::string result; + switch(cause) + { + case restinio::connection_state::cause_t::accepted: + result = "accepted"; + break; + case restinio::connection_state::cause_t::closed: + result = "closed"; + break; + case restinio::connection_state::cause_t::upgraded_to_websocket: + result = "upgraded"; + break; + default: + result = "unknown"; + } + return result; +} + // client Client::Client(const std::string ip, const uint16_t port){