From bc84655f424c8a3c75235a83a16d3f163f6ebfa3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Adrien=20B=C3=A9raud?= Date: Thu, 31 Oct 2019 15:56:08 -0400 Subject: [PATCH] proxy server: update for Restinio v.0.6.0.1 --- include/opendht/dht_proxy_server.h | 7 +-- src/dht_proxy_server.cpp | 71 +++++++++--------------------- 2 files changed, 26 insertions(+), 52 deletions(-) diff --git a/include/opendht/dht_proxy_server.h b/include/opendht/dht_proxy_server.h index 3edc7bc9..0ebab3e8 100644 --- a/include/opendht/dht_proxy_server.h +++ b/include/opendht/dht_proxy_server.h @@ -149,6 +149,8 @@ private: std::unique_ptr createRestRouter(); + void onConnectionClosed(restinio::connection_id_t); + /** * Return the PublicKey id, the node id and node stats * Method: GET "/" @@ -336,10 +338,9 @@ private: std::unique_ptr printStatsTimer_; // Thread-safe access to listeners map. - std::shared_ptr lockListener_; + std::mutex lockListener_; // Shared with connection listener. - std::shared_ptr> listeners_; + std::map listeners_; // Connection Listener observing conn state changes. std::shared_ptr connListener_; diff --git a/src/dht_proxy_server.cpp b/src/dht_proxy_server.cpp index 9acaf7a1..ccf1aec7 100644 --- a/src/dht_proxy_server.cpp +++ b/src/dht_proxy_server.cpp @@ -110,10 +110,7 @@ class DhtProxyServer::ConnectionListener { public: ConnectionListener() {}; - ConnectionListener(std::shared_ptr dht, - std::shared_ptr> listeners, - std::shared_ptr lock, std::shared_ptr logger): - dht_(dht), lock_(lock), listeners_(listeners), logger_(logger) {}; + ConnectionListener(std::function onClosed) : onClosed_(std::move(onClosed)) {}; ~ConnectionListener() {}; /** @@ -123,54 +120,31 @@ public: void state_changed(const restinio::connection_state::notice_t& notice) noexcept; private: - std::string to_str( restinio::connection_state::cause_t cause ) noexcept; - - std::shared_ptr dht_; - std::shared_ptr lock_; - std::shared_ptr> listeners_; - - std::shared_ptr logger_; + std::function onClosed_; }; void DhtProxyServer::ConnectionListener::state_changed(const restinio::connection_state::notice_t& notice) noexcept { - std::lock_guard lock(*lock_); - auto id = notice.connection_id(); - auto cause = to_str(notice.cause()); - - if (listeners_->find(id) != listeners_->end()){ - if (notice.cause() == restinio::connection_state::cause_t::closed){ - if (logger_) - logger_->d("[proxy:server] [connection:%li] cancelling listener", id); - dht_->cancelListen(listeners_->at(id).hash, - std::move(listeners_->at(id).token)); - listeners_->erase(id); - if (logger_) - logger_->d("[proxy:server] %li listeners are connected", listeners_->size()); - } + if (restinio::holds_alternative(notice.cause())) { + onClosed_(notice.connection_id()); } } -std::string -DhtProxyServer::ConnectionListener::to_str(restinio::connection_state::cause_t cause) noexcept +void +DhtProxyServer::onConnectionClosed(restinio::connection_id_t id) { - 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"; + std::lock_guard lock(lockListener_); + auto it = listeners_.find(id); + if (it != listeners_.end()) { + if (logger_) + logger_->d("[proxy:server] [connection:%li] cancelling listener", id); + dht_->cancelListen(it->second.hash, + std::move(it->second.token)); + listeners_.erase(it); + if (logger_) + logger_->d("[proxy:server] %li listeners are connected", listeners_.size()); } - return result; } struct DhtProxyServer::RestRouterTraitsTls : public restinio::default_tls_traits_t @@ -199,9 +173,8 @@ DhtProxyServer::DhtProxyServer( std::shared_ptr dht, in_port_t port, const std::string& pushServer, std::shared_ptr logger ) - : dht_(dht), logger_(logger), lockListener_(std::make_shared()), - listeners_(std::make_shared>()), - connListener_(std::make_shared(dht, listeners_, lockListener_, logger)), + : dht_(dht), logger_(logger), + connListener_(std::make_shared(std::bind(&DhtProxyServer::onConnectionClosed, this, std::placeholders::_1))), pushServer_(pushServer) { if (not dht_) @@ -365,7 +338,7 @@ DhtProxyServer::updateStats() const stats_.pushListenersCount = pushListeners_.size(); #endif stats_.putCount = puts_.size(); - stats_.listenCount = listeners_->size(); + stats_.listenCount = listeners_.size(); stats_.nodeInfo = nodeInfo_; } @@ -585,9 +558,9 @@ DhtProxyServer::listen(restinio::request_handle_t request, this->initHttpResponse(request->create_response())); response->flush(); try { - std::lock_guard lock(*lockListener_); + std::lock_guard lock(lockListener_); // save the listener to handle a disconnect - auto &session = (*listeners_)[request->connection_id()]; + auto &session = listeners_[request->connection_id()]; session.hash = infoHash; session.response = response; session.token = dht_->listen(infoHash, [this, response] @@ -829,7 +802,7 @@ DhtProxyServer::handleCancelPushListen(const asio::error_code &ec, const std::st if (logger_) logger_->d("[proxy:server] [listen:push %s] cancelled for %s", key.toString().c_str(), clientId.c_str()); - std::lock_guard lock(*lockListener_); + std::lock_guard lock(lockListener_); auto pushListener = pushListeners_.find(pushToken); if (pushListener == pushListeners_.end())