diff --git a/CMakeLists.txt b/CMakeLists.txt
index 8abc3610..d55a6363 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -16,10 +16,10 @@ option (OPENDHT_SYSTEMD "Install systemd module" OFF)
option (OPENDHT_ARGON2 "Use included argon2 sources" OFF)
option (OPENDHT_LTO "Build with LTO" OFF)
option (OPENDHT_SANITIZE "Build with address sanitizer and stack protector" OFF)
-option (OPENDHT_PROXY_SERVER "Enable DHT proxy server, use Restbed and jsoncpp" OFF)
+option (OPENDHT_PROXY_SERVER "Enable DHT proxy server, use Restinio and jsoncpp" OFF)
option (OPENDHT_PUSH_NOTIFICATIONS "Enable push notifications support" OFF)
option (OPENDHT_PROXY_SERVER_IDENTITY "Allow clients to use the node identity" OFF)
-option (OPENDHT_PROXY_CLIENT "Enable DHT proxy client, use Restbed and jsoncpp" OFF)
+option (OPENDHT_PROXY_CLIENT "Enable DHT proxy client, use Restinio and jsoncpp" OFF)
option (OPENDHT_INDEX "Build DHT indexation feature" OFF)
option (OPENDHT_TESTS "Add unit tests executable" OFF)
@@ -58,16 +58,22 @@ if (Jsoncpp_FOUND)
endif()
if (OPENDHT_PROXY_SERVER OR OPENDHT_PROXY_CLIENT)
- find_package(Restbed REQUIRED)
+ find_package(Restinio REQUIRED)
+ if (Restinio_FOUND)
+ find_library(FMT_LIBRARY fmt)
+ add_library(fmt SHARED IMPORTED)
+ find_library(HTTP_PARSER_LIBRARY http_parser)
+ add_library(http_parser SHARED IMPORTED)
+ endif()
if (NOT Jsoncpp_FOUND)
message(SEND_ERROR "Jsoncpp is required for DHT proxy support")
endif()
endif()
# Build flags
-set (CMAKE_CXX_STANDARD 11)
+set (CMAKE_CXX_STANDARD 14)
set (CMAKE_CXX_STANDARD_REQUIRED on)
-set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wno-return-type -Wall -Wextra -Wnon-virtual-dtor -pedantic-errors -fvisibility=hidden")
+set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -lpthread -Wno-return-type -Wall -Wextra -Wnon-virtual-dtor -pedantic-errors -fvisibility=hidden")
if (OPENDHT_SANITIZE)
set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -fsanitize=address -fstack-protector-strong")
endif ()
@@ -100,6 +106,9 @@ endif ()
if (Nettle_INCLUDE_DIRS)
include_directories (SYSTEM "${Nettle_INCLUDE_DIRS}")
endif ()
+if (Restinio_INCLUDE_DIR)
+ include_directories (SYSTEM "${Restinio_INCLUDE_DIR}")
+endif ()
if (Restbed_INCLUDE_DIR)
include_directories (SYSTEM "${Restbed_INCLUDE_DIR}")
endif ()
@@ -255,8 +264,9 @@ if (OPENDHT_STATIC)
target_include_directories(opendht-static SYSTEM PRIVATE ${argon2_INCLUDE_DIRS})
endif ()
target_link_libraries(opendht-static
- PRIVATE ${Restbed_LIBRARY} ${argon2_LIBRARIES}
- PUBLIC ${CMAKE_THREAD_LIBS_INIT} ${GNUTLS_LIBRARIES} ${Nettle_LIBRARIES} ${Jsoncpp_LIBRARIES})
+ PRIVATE ${Restbed_LIBRARY} ${argon2_LIBRARIES}
+ PUBLIC ${CMAKE_THREAD_LIBS_INIT} ${GNUTLS_LIBRARIES} ${Nettle_LIBRARIES}
+ ${Jsoncpp_LIBRARIES} ${FMT_LIBRARY} ${HTTP_PARSER_LIBRARY})
install (TARGETS opendht-static DESTINATION ${CMAKE_INSTALL_LIBDIR} EXPORT opendht)
endif ()
@@ -274,7 +284,11 @@ if (OPENDHT_SHARED)
target_link_libraries(opendht PRIVATE ${argon2_LIBRARIES})
target_include_directories(opendht SYSTEM PRIVATE ${argon2_INCLUDE_DIRS})
endif ()
- target_link_libraries(opendht PRIVATE ${CMAKE_THREAD_LIBS_INIT} ${GNUTLS_LIBRARIES} ${Nettle_LIBRARIES} ${Restbed_LIBRARY} ${Jsoncpp_LIBRARIES})
+ target_link_libraries(opendht
+ PRIVATE ${CMAKE_THREAD_LIBS_INIT} ${GNUTLS_LIBRARIES} ${Nettle_LIBRARIES}
+ ${Restbed_LIBRARY} ${Jsoncpp_LIBRARIES}
+ ${FMT_LIBRARY} ${HTTP_PARSER_LIBRARY})
+
install (TARGETS opendht DESTINATION ${CMAKE_INSTALL_LIBDIR} EXPORT opendht)
endif ()
diff --git a/README.md b/README.md
index fb51e8f7..7c3fc1d6 100644
--- a/README.md
+++ b/README.md
@@ -4,7 +4,7 @@
OpenDHT
-A lightweight C++11 Distributed Hash Table implementation.
+A lightweight C++14 Distributed Hash Table implementation.
OpenDHT provides an easy to use distributed in-memory data store.
Every node in the network can read and write values to the store.
@@ -14,7 +14,7 @@ Values are distributed over the network, with redundancy.
* High resilience to network disruption
* Public key cryptography layer providing optional data signature and encryption (using GnuTLS)
* IPv4 and IPv6 support
- * Clean and powerful C++11 map API
+ * Clean and powerful C++14 map API
* Python 3 bindings
* REST API
@@ -27,7 +27,7 @@ Build instructions:
* Adrien Béraud
+ * 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
@@ -30,12 +31,21 @@
#include
#include
#include
-#include
+#include
#ifdef OPENDHT_JSONCPP
#include
#endif
+using RestRouter = restinio::router::express_router_t<>;
+using RestRouterTraits = restinio::traits_t<
+ restinio::asio_timer_manager_t,
+ restinio::single_threaded_ostream_logger_t,
+ RestRouter>;
+using RequestStatus = restinio::request_handling_status_t;
+using ResponseByParts = restinio::chunked_output_t;
+using ResponseByPartsBuilder = restinio::response_builder_t;
+
namespace Json {
class Value;
}
@@ -123,6 +133,11 @@ public:
void stop();
private:
+ template
+ HttpResponse initHttpResponse(HttpResponse response) const;
+
+ std::unique_ptr createRestRouter();
+
/**
* Return the PublicKey id, the node id and node stats
* Method: GET "/"
@@ -130,7 +145,8 @@ private:
* On error: HTTP 503, body: {"err":"xxxx"}
* @param session
*/
- void getNodeInfo(const std::shared_ptr& session) const;
+ RequestStatus getNodeInfo(restinio::request_handle_t request,
+ restinio::router::route_params_t params) const;
/**
* Return ServerStats in JSON format
@@ -138,7 +154,7 @@ private:
* Result: HTTP 200, body: Node infos in JSON format
* @param session
*/
- void getStats(const std::shared_ptr& session) const;
+ //void getStats(const std::shared_ptr& session) const;
/**
* Return Values of an infoHash
@@ -150,7 +166,8 @@ private:
* On error: HTTP 503, body: {"err":"xxxx"}
* @param session
*/
- void get(const std::shared_ptr& session) const;
+ RequestStatus get(restinio::request_handle_t request,
+ restinio::router::route_params_t params);
/**
* Listen incoming Values of an infoHash.
@@ -162,7 +179,7 @@ private:
* On error: HTTP 503, body: {"err":"xxxx"}
* @param session
*/
- void listen(const std::shared_ptr& session);
+ //void listen(const std::shared_ptr& session);
/**
* Put a value on the DHT
@@ -173,7 +190,8 @@ private:
* HTTP 400, body: {"err":"xxxx"} if bad json or HTTP 502 if put fails
* @param session
*/
- void put(const std::shared_ptr& session);
+ RequestStatus put(restinio::request_handle_t request,
+ restinio::router::route_params_t params);
void cancelPut(const InfoHash& key, Value::Id vid);
@@ -187,7 +205,7 @@ private:
* HTTP 400, body: {"err":"xxxx"} if bad json
* @param session
*/
- void putSigned(const std::shared_ptr& session) const;
+ //void putSigned(const std::shared_ptr& session) const;
/**
* Put a value to encrypt by the proxy on the DHT
@@ -198,7 +216,7 @@ private:
* HTTP 400, body: {"err":"xxxx"} if bad json
* @param session
*/
- void putEncrypted(const std::shared_ptr& session) const;
+ //void putEncrypted(const std::shared_ptr& session) const;
#endif // OPENDHT_PROXY_SERVER_IDENTITY
/**
@@ -211,7 +229,7 @@ private:
* On error: HTTP 503, body: {"err":"xxxx"}
* @param session
*/
- void getFiltered(const std::shared_ptr& session) const;
+ //void getFiltered(const std::shared_ptr& session) const;
/**
* Respond allowed Methods
@@ -220,7 +238,8 @@ private:
* See https://developer.mozilla.org/en-US/docs/Web/HTTP/Methods/OPTIONS
* @param session
*/
- void handleOptionsMethod(const std::shared_ptr& session) const;
+ RequestStatus options(restinio::request_handle_t request,
+ restinio::router::route_params_t params);
/**
* Remove finished listeners
@@ -238,7 +257,7 @@ private:
* so you need to refresh the operation each six hours.
* @param session
*/
- void subscribe(const std::shared_ptr& session);
+ //void subscribe(const std::shared_ptr& session);
/**
* Unsubscribe to push notifications for an iOS or Android device.
* Method: UNSUBSCRIBE "/{InfoHash: .*}"
@@ -246,7 +265,7 @@ private:
* Return: nothing
* @param session
*/
- void unsubscribe(const std::shared_ptr& session);
+ //void unsubscribe(const std::shared_ptr& session);
/**
* Send a push notification via a gorush push gateway
* @param key of the device
@@ -269,7 +288,7 @@ private:
using time_point = clock::time_point;
std::thread server_thread {};
- std::unique_ptr service_;
+ //std::unique_ptr service_;
std::shared_ptr dht_;
Json::StreamWriterBuilder jsonBuilder_;
@@ -287,7 +306,7 @@ private:
// NOTE: can be simplified when we will supports restbed 5.0
std::thread listenThread_;
struct SessionToHashToken {
- std::shared_ptr session;
+ //std::shared_ptr session;
InfoHash hash;
std::future token;
};
@@ -313,6 +332,12 @@ private:
std::map pushListeners_;
proxy::ListenToken tokenPushNotif_ {0};
#endif //OPENDHT_PUSH_NOTIFICATIONS
+
+ const std::string RESP_MSG_INCORRECT_JSON = "{\"err:\":\"Incorrect JSON\"}";
+ const std::string RESP_MSG_SERVICE_UNAVAILABLE = "{\"err\":\"Incorrect DhtRunner\"}";
+ const std::string RESP_MSG_INTERNAL_SERVER_ERRROR = "{\"err\":\"Internal server error\"}";
+ const std::string RESP_MSG_MISSING_PARAMS = "{\"err\":\"Missing parameters\"}";
+ const std::string RESP_MSG_PUT_FAILED = "{\"err\":\"Put failed\"}";
};
}
diff --git a/python/opendht.pyx b/python/opendht.pyx
index 1fa3c239..538276b8 100644
--- a/python/opendht.pyx
+++ b/python/opendht.pyx
@@ -1,5 +1,5 @@
# distutils: language = c++
-# distutils: extra_compile_args = -std=c++11
+# distutils: extra_compile_args = -std=c++14
# distutils: include_dirs = ../../include
# distutils: library_dirs = ../../src
# distutils: libraries = opendht gnutls
diff --git a/python/setup.py.in b/python/setup.py.in
index 2f67164a..09ff43d7 100644
--- a/python/setup.py.in
+++ b/python/setup.py.in
@@ -50,8 +50,8 @@ setup(name="opendht",
["@CURRENT_SOURCE_DIR@/opendht.pyx"],
include_dirs = ['@PROJECT_SOURCE_DIR@/include'],
language="c++",
- extra_compile_args=["-std=c++11"],
- extra_link_args=["-std=c++11"],
+ extra_compile_args=["-std=c++14"],
+ extra_link_args=["-std=c++14"],
libraries=["opendht"],
library_dirs = ['@CURRENT_BINARY_DIR@', '@PROJECT_BINARY_DIR@']
))
diff --git a/src/dht_proxy_server.cpp b/src/dht_proxy_server.cpp
index 744ba986..d9c73f9a 100644
--- a/src/dht_proxy_server.cpp
+++ b/src/dht_proxy_server.cpp
@@ -2,6 +2,7 @@
* Copyright (C) 2017-2019 Savoir-faire Linux Inc.
* Author: Sébastien Blin
* Adrien Béraud
+ * 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
@@ -55,8 +56,8 @@ DhtProxyServer::DhtProxyServer(std::shared_ptr dht, in_port_t port ,
{
if (not dht_)
throw std::invalid_argument("A DHT instance must be provided");
- // NOTE in c++14, use make_unique
- service_ = std::unique_ptr(new restbed::Service());
+
+ //service_ = std::unique_ptr(new restbed::Service());
std::cout << "Running DHT proxy server on port " << port << std::endl;
if (not pushServer.empty()) {
@@ -71,50 +72,31 @@ DhtProxyServer::DhtProxyServer(std::shared_ptr dht, in_port_t port ,
jsonBuilder_["indentation"] = "";
server_thread = std::thread([this, port]() {
- // Create endpoints
- auto resource = std::make_shared();
- resource->set_path("/");
- resource->set_method_handler("GET", std::bind(&DhtProxyServer::getNodeInfo, this, _1));
- resource->set_method_handler("STATS", std::bind(&DhtProxyServer::getStats, this, _1));
- service_->publish(resource);
- resource = std::make_shared();
- resource->set_path("/{hash: .*}");
- resource->set_method_handler("GET", std::bind(&DhtProxyServer::get, this, _1));
- resource->set_method_handler("LISTEN", [this](const Sp& session) mutable { listen(session); } );
-#ifdef OPENDHT_PUSH_NOTIFICATIONS
- resource->set_method_handler("SUBSCRIBE", [this](const Sp& session) mutable { subscribe(session); } );
- resource->set_method_handler("UNSUBSCRIBE", [this](const Sp& session) mutable { unsubscribe(session); } );
-#endif //OPENDHT_PUSH_NOTIFICATIONS
- resource->set_method_handler("POST", [this](const Sp& session) mutable { put(session); });
-#ifdef OPENDHT_PROXY_SERVER_IDENTITY
- resource->set_method_handler("SIGN", std::bind(&DhtProxyServer::putSigned, this, _1));
- resource->set_method_handler("ENCRYPT", std::bind(&DhtProxyServer::putEncrypted, this, _1));
-#endif // OPENDHT_PROXY_SERVER_IDENTITY
- resource->set_method_handler("OPTIONS", std::bind(&DhtProxyServer::handleOptionsMethod, this, _1));
- service_->publish(resource);
- resource = std::make_shared();
- resource->set_path("/{hash: .*}/{value: .*}");
- resource->set_method_handler("GET", std::bind(&DhtProxyServer::getFiltered, this, _1));
- service_->publish(resource);
-
- // Start server
- auto settings = std::make_shared();
- settings->set_default_header("Content-Type", "application/json");
- settings->set_default_header("Connection", "keep-alive");
- settings->set_default_header("Access-Control-Allow-Origin", "*");
- std::chrono::milliseconds timeout(std::numeric_limits::max());
- settings->set_connection_timeout(timeout); // there is a timeout, but really huge
- settings->set_port(port);
+ using namespace std::chrono;
auto maxThreads = std::thread::hardware_concurrency() - 1;
- settings->set_worker_limit(maxThreads > 1 ? maxThreads : 1);
- lastStatsReset_ = clock::now();
+ auto restThreads = maxThreads > 1 ? maxThreads : 1;
+ auto settings = restinio::on_thread_pool(restThreads);
+ settings.address("0.0.0.0");
+ settings.port(port);
+ settings.request_handler(this->createRestRouter());
+ settings.read_next_http_message_timelimit(10s);
+ settings.write_http_response_timelimit(10s);
+ settings.handle_request_timeout(10s);
+ settings.socket_options_setter([](auto & options){
+ options.set_option(asio::ip::tcp::no_delay{true});
+ });
try {
- service_->start(settings);
- } catch(std::system_error& e) {
- std::cerr << "Error running server on port " << port << ": " << e.what() << std::endl;
+ restinio::run(std::move(settings));
+ }
+ catch(const restinio::exception_t &ex) {
+ std::cerr << "Error: " << ex.what() << std::endl;
+ }
+ catch(const std::exception &ex) {
+ std::cerr << "Error: " << ex.what() << std::endl;
}
});
+ /*
listenThread_ = std::thread([this]() {
while (not service_->is_up() and not stopListeners) {
std::this_thread::sleep_for(std::chrono::seconds(1));
@@ -139,11 +121,12 @@ DhtProxyServer::DhtProxyServer(std::shared_ptr dht, in_port_t port ,
schedulerCv_.wait_until(lock, next);
}
});
+ */
dht->forwardAllMessages(true);
printStatsJob_ = scheduler_.add(scheduler_.time() + PRINT_STATS_PERIOD, [this] {
if (stopListeners) return;
- if (service_->is_up())
- updateStats();
+ //if (service_->is_up())
+ // updateStats();
// Refresh stats cache
auto newInfo = dht_->getNodeInfo();
{
@@ -164,15 +147,17 @@ DhtProxyServer::stop()
{
if (printStatsJob_)
printStatsJob_->cancel();
+ /*
service_->stop();
{
std::lock_guard lock(lockListener_);
auto listener = currentListeners_.begin();
while (listener != currentListeners_.end()) {
- listener->session->close();
+ //listener->session->close();
++listener;
}
}
+ */
stopListeners = true;
schedulerCv_.notify_all();
// listenThreads_ will stop because there is no more sessions
@@ -201,166 +186,85 @@ DhtProxyServer::updateStats() const
stats_.nodeInfo = nodeInfo_;
}
-void
-DhtProxyServer::getNodeInfo(const Sp& session) const
+template
+HttpResponse DhtProxyServer::initHttpResponse(HttpResponse response) const
{
- requestNum_++;
- const auto request = session->get_request();
- int content_length = std::stoi(request->get_header("Content-Length", "0"));
- session->fetch(content_length,
- [this](const Sp& s, const restbed::Bytes& /*b*/) mutable
- {
- try {
- if (dht_) {
- Json::Value result;
- {
- std::lock_guard lck(statsMutex_);
- if (nodeInfo_.ipv4.good_nodes == 0 && nodeInfo_.ipv6.good_nodes == 0) {
- // NOTE: we want to avoid the disconnected state as much as possible
- // So, if the node is disconnected, we should force the update of the cache
- // and reconnect as soon as possible
- // This should not happen much
- nodeInfo_ = dht_->getNodeInfo();
- }
- result = nodeInfo_.toJson();
- }
- result["public_ip"] = s->get_origin(); // [ipv6:ipv4]:port or ipv4:port
- auto output = Json::writeString(jsonBuilder_, result) + "\n";
- s->close(restbed::OK, output);
- }
- else
- s->close(restbed::SERVICE_UNAVAILABLE, "{\"err\":\"Incorrect DhtRunner\"}");
- } catch (...) {
- s->close(restbed::INTERNAL_SERVER_ERROR, "{\"err\":\"Internal server error\"}");
- }
- }
- );
+ response.append_header("Server", "RESTinio");
+ response.append_header(restinio::http_field::content_type, "application/json");
+ response.append_header(restinio::http_field::access_control_allow_origin, "*");
+ response.connection_keep_alive();
+ return response;
}
-void
-DhtProxyServer::getStats(const Sp& session) const
+std::unique_ptr
+DhtProxyServer::createRestRouter()
{
- requestNum_++;
- const auto request = session->get_request();
- int content_length = std::stoi(request->get_header("Content-Length", "0"));
- session->fetch(content_length,
- [this](const Sp& s, const restbed::Bytes& /*b*/) mutable
- {
- try {
- if (dht_) {
-#ifdef OPENDHT_JSONCPP
- auto output = Json::writeString(jsonBuilder_, stats_.toJson()) + "\n";
- s->close(restbed::OK, output);
-#else
- s->close(restbed::NotFound, "{\"err\":\"JSON not enabled on this instance\"}");
-#endif
- }
- else
- s->close(restbed::SERVICE_UNAVAILABLE, "{\"err\":\"Incorrect DhtRunner\"}");
- } catch (...) {
- s->close(restbed::INTERNAL_SERVER_ERROR, "{\"err\":\"Internal server error\"}");
- }
- }
- );
+ using namespace std::placeholders;
+ auto router = std::make_unique();
+ router->add_handler(restinio::http_method_t::http_options,
+ "/:hash", std::bind(&DhtProxyServer::options, this, _1, _2));
+ router->http_get("/", std::bind(&DhtProxyServer::getNodeInfo, this, _1, _2));
+ router->http_get("/:hash", std::bind(&DhtProxyServer::get, this, _1, _2));
+ router->http_post("/:hash", std::bind(&DhtProxyServer::put, this, _1, _2));
+ return router;
}
-void
-DhtProxyServer::get(const Sp& session) const
+RequestStatus
+DhtProxyServer::getNodeInfo(restinio::request_handle_t request,
+ restinio::router::route_params_t params) const
{
- requestNum_++;
- const auto request = session->get_request();
- int content_length = std::stoi(request->get_header("Content-Length", "0"));
- auto hash = request->get_path_parameter("hash");
- session->fetch(content_length,
- [=](const Sp& s, const restbed::Bytes& /*b* */)
- {
- try {
- if (dht_) {
- InfoHash infoHash(hash);
- if (!infoHash) {
- infoHash = InfoHash::get(hash);
- }
- s->yield(restbed::OK, "", [=](const Sp&) {});
- dht_->get(infoHash, [this,s](const Sp& value) {
- if (s->is_closed()) return false;
- // Send values as soon as we get them
- auto output = Json::writeString(jsonBuilder_, value->toJson()) + "\n";
- s->yield(output, [](const Sp& /*session*/){ });
- return true;
- }, [s](bool /*ok* */) {
- // Communication is finished
- if (not s->is_closed()) {
- s->close();
- }
- });
- } else {
- s->close(restbed::SERVICE_UNAVAILABLE, "{\"err\":\"Incorrect DhtRunner\"}");
- }
- } catch (...) {
- s->close(restbed::INTERNAL_SERVER_ERROR, "{\"err\":\"Internal server error\"}");
- }
- }
- );
+ Json::Value result;
+ std::lock_guard lck(statsMutex_);
+ if (nodeInfo_.ipv4.good_nodes == 0 &&
+ nodeInfo_.ipv6.good_nodes == 0){
+ nodeInfo_ = this->dht_->getNodeInfo();
+ }
+ result = nodeInfo_.toJson();
+ // [ipv6:ipv4]:port or ipv4:port
+ result["public_ip"] = request->remote_endpoint().address().to_string();
+ auto output = Json::writeString(jsonBuilder_, result) + "\n";
+
+ auto response = this->initHttpResponse(request->create_response());
+ response.append_body(output);
+ return response.done();
}
-void
-DhtProxyServer::listen(const Sp& session)
+//DhtProxyServer::getStats()
+
+RequestStatus
+DhtProxyServer::get(restinio::request_handle_t request,
+ restinio::router::route_params_t params)
{
requestNum_++;
- const auto request = session->get_request();
- int content_length = std::stoi(request->get_header("Content-Length", "0"));
- auto hash = request->get_path_parameter("hash");
- InfoHash infoHash(hash);
+ dht::InfoHash infoHash(params["hash"].to_string());
if (!infoHash)
- infoHash = InfoHash::get(hash);
- session->fetch(content_length,
- [=](const Sp& s, const restbed::Bytes& /*b* */)
- {
- try {
- if (dht_) {
- InfoHash infoHash(hash);
- if (!infoHash) {
- infoHash = InfoHash::get(hash);
- }
- s->yield(restbed::OK);
- // Handle client deconnection
- // NOTE: for now, there is no handler, so we test the session in a thread
- // will be the case in restbed 5.0
- SessionToHashToken listener;
- listener.session = session;
- listener.hash = infoHash;
- // cache the session to avoid an incrementation of the shared_ptr's counter
- // else, the session->close() will not close the socket.
- auto cacheSession = std::weak_ptr(s);
- listener.token = dht_->listen(infoHash, [this,cacheSession](const std::vector>& values, bool expired) {
- auto s = cacheSession.lock();
- if (!s) return false;
- // Send values as soon as we get them
- if (!s->is_closed()) {
- for (const auto& value : values) {
- auto val = value->toJson();
- if (expired)
- val["expired"] = true;
- auto output = Json::writeString(jsonBuilder_, val) + "\n";
- s->yield(output, [](const Sp&){ });
- }
- }
- return !s->is_closed();
- });
- {
- std::lock_guard lock(lockListener_);
- currentListeners_.emplace_back(std::move(listener));
- }
- } else {
- session->close(restbed::SERVICE_UNAVAILABLE, "{\"err\":\"Incorrect DhtRunner\"}");
- }
- } catch (...) {
- s->close(restbed::INTERNAL_SERVER_ERROR, "{\"err\":\"Internal server error\"}");
- }
- }
- );
+ infoHash = dht::InfoHash::get(params["hash"].to_string());
+
+ if (!dht_){
+ auto response = this->initHttpResponse(
+ request->create_response(restinio::status_service_unavailable()));
+ response.set_body(this->RESP_MSG_SERVICE_UNAVAILABLE);
+ return response.done();
+ }
+
+ auto response = std::make_shared(
+ this->initHttpResponse(request->create_response()));
+ response->flush();
+
+ dht_->get(infoHash, [this, response](const dht::Sp& value){
+ auto output = Json::writeString(jsonBuilder_, value->toJson()) + "\n";
+ response->append_chunk(output);
+ response->flush();
+ return true;
+ },
+ [response] (bool /*ok*/){
+ response->done();
+ });
+ return restinio::request_handling_status_t::accepted;
}
+//DhtProxyServer::listen()
+
#ifdef OPENDHT_PUSH_NOTIFICATIONS
struct DhtProxyServer::Listener {
@@ -374,171 +278,8 @@ struct DhtProxyServer::PushListener {
bool isAndroid;
};
-void
-DhtProxyServer::subscribe(const std::shared_ptr& session)
-{
- requestNum_++;
- const auto request = session->get_request();
- int content_length = std::stoi(request->get_header("Content-Length", "0"));
- auto hash = request->get_path_parameter("hash");
- InfoHash infoHash(hash);
- if (!infoHash)
- infoHash = InfoHash::get(hash);
- session->fetch(content_length,
- [=](const std::shared_ptr s, const restbed::Bytes& b) mutable
- {
- try {
- std::string err;
- Json::Value root;
- Json::CharReaderBuilder rbuilder;
- auto* char_data = reinterpret_cast(b.data());
- auto reader = std::unique_ptr(rbuilder.newCharReader());
- if (!reader->parse(char_data, char_data + b.size(), &root, &err)) {
- s->close(restbed::BAD_REQUEST, "{\"err\":\"Incorrect JSON\"}");
- return;
- }
- auto pushToken = root["key"].asString();
- if (pushToken.empty()) {
- s->close(restbed::BAD_REQUEST, "{\"err\":\"No token\"}");
- return;
- }
- auto platform = root["platform"].asString();
- auto isAndroid = platform == "android";
- auto clientId = root.isMember("client_id") ? root["client_id"].asString() : std::string();
-
- std::cout << "Subscribe " << infoHash << " client:" << clientId << std::endl;
-
- {
- std::lock(schedulerLock_, lockListener_);
- std::lock_guard lk1(lockListener_, std::adopt_lock);
- std::lock_guard lk2(schedulerLock_, std::adopt_lock);
- scheduler_.syncTime();
- auto timeout = scheduler_.time() + proxy::OP_TIMEOUT;
- // Check if listener is already present and refresh timeout if launched
- // One push listener per pushToken.infoHash.clientId
- auto pushListener = pushListeners_.emplace(pushToken, PushListener{}).first;
- auto listeners = pushListener->second.listeners.emplace(infoHash, std::vector{}).first;
- for (auto& listener: listeners->second) {
- if (listener.clientId == clientId) {
- scheduler_.edit(listener.expireJob, timeout);
- scheduler_.edit(listener.expireNotifyJob, timeout - proxy::OP_MARGIN);
- s->yield(restbed::OK);
-
- if (!root.isMember("refresh") or !root["refresh"].asBool()) {
- dht_->get(
- infoHash,
- [this, s](const Sp &value) {
- if (s->is_closed())
- return false;
- // Send values as soon as we get them
- auto output = Json::writeString(jsonBuilder_, value->toJson()) + "\n";
- s->yield(output, [](const Sp
- & /*session*/) {});
- return true;
- },
- [s](bool /*ok* */) {
- // Communication is finished
- if (not s->is_closed()) {
- s->close("{}\n");
- }
- });
- } else {
- // Communication is finished
- if (not s->is_closed()) {
- s->close("{}\n");
- }
- }
- schedulerCv_.notify_one();
- return;
- }
- }
- listeners->second.emplace_back(Listener{});
- auto& listener = listeners->second.back();
- listener.clientId = clientId;
-
- // New listener
- pushListener->second.isAndroid = isAndroid;
-
- // The listener is not found, so add it.
- listener.internalToken = dht_->listen(infoHash,
- [this, infoHash, pushToken, isAndroid, clientId](const std::vector>& values, bool expired) {
- threadPool_->run([this, infoHash, pushToken, isAndroid, clientId, values, expired]() {
- // Build message content
- Json::Value json;
- json["key"] = infoHash.toString();
- json["to"] = clientId;
- if (expired and values.size() < 3) {
- std::stringstream ss;
- for(size_t i = 0; i < values.size(); ++i) {
- if(i != 0) ss << ",";
- ss << values[i]->id;
- }
- json["exp"] = ss.str();
- }
- sendPushNotification(pushToken, std::move(json), isAndroid);
- });
- return true;
- }
- );
- listener.expireJob = scheduler_.add(timeout,
- [this, clientId, infoHash, pushToken] {
- cancelPushListen(pushToken, infoHash, clientId);
- }
- );
- listener.expireNotifyJob = scheduler_.add(timeout - proxy::OP_MARGIN,
- [this, infoHash, pushToken, isAndroid, clientId] {
- std::cout << "Listener: sending refresh " << infoHash << std::endl;
- Json::Value json;
- json["timeout"] = infoHash.toString();
- json["to"] = clientId;
- sendPushNotification(pushToken, std::move(json), isAndroid);
- }
- );
- }
- schedulerCv_.notify_one();
- s->close(restbed::OK, "{}\n");
- } catch (...) {
- s->close(restbed::INTERNAL_SERVER_ERROR, "{\"err\":\"Internal server error\"}");
- }
- }
- );
-}
-
-void
-DhtProxyServer::unsubscribe(const std::shared_ptr& session)
-{
- requestNum_++;
- const auto request = session->get_request();
- int content_length = std::stoi(request->get_header("Content-Length", "0"));
- auto hash = request->get_path_parameter("hash");
- InfoHash infoHash(hash);
- if (!infoHash)
- infoHash = InfoHash::get(hash);
- session->fetch(content_length,
- [=](const std::shared_ptr s, const restbed::Bytes& b)
- {
- try {
- std::string err;
- Json::Value root;
- Json::CharReaderBuilder rbuilder;
- auto* char_data = reinterpret_cast(b.data());
- auto reader = std::unique_ptr(rbuilder.newCharReader());
- if (!reader->parse(char_data, char_data + b.size(), &root, &err)) {
- s->close(restbed::BAD_REQUEST, "{\"err\":\"Incorrect JSON\"}");
- return;
- }
- auto pushToken = root["key"].asString();
- if (pushToken.empty()) return;
- auto clientId = root["client_id"].asString();
-
- cancelPushListen(pushToken, infoHash, clientId);
- s->close(restbed::OK);
- } catch (...) {
- s->close(restbed::INTERNAL_SERVER_ERROR, "{\"err\":\"Internal server error\"}");
- }
- }
- );
-}
+//DhtProxyServer::subscribe()
+//DhtProxyServer::unsubscribe()
void
DhtProxyServer::cancelPushListen(const std::string& pushToken, const dht::InfoHash& key, const std::string& clientId)
@@ -568,45 +309,7 @@ DhtProxyServer::cancelPushListen(const std::string& pushToken, const dht::InfoHa
}
}
-void
-DhtProxyServer::sendPushNotification(const std::string& token, Json::Value&& json, bool isAndroid) const
-{
- if (pushServer_.empty())
- return;
- restbed::Uri uri(proxy::HTTP_PROTO + pushServer_ + "/api/push");
- auto req = std::make_shared(uri);
- req->set_method("POST");
-
- // 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 content;
- content["notifications"] = std::move(notifications);
-
- Json::StreamWriterBuilder wbuilder;
- wbuilder["commentStyle"] = "None";
- wbuilder["indentation"] = "";
- auto valueStr = Json::writeString(wbuilder, content);
-
- req->set_header("Content-Type", "application/json");
- req->set_header("Accept", "*/*");
- req->set_header("Host", pushServer_);
- req->set_header("Content-Length", std::to_string(valueStr.length()));
- req->set_body(valueStr);
-
- // Send request.
- restbed::Http::async(req, {});
-}
+//DhtProxyServer::sendPushNotification()
#endif //OPENDHT_PUSH_NOTIFICATIONS
@@ -630,264 +333,144 @@ DhtProxyServer::cancelPut(const InfoHash& key, Value::Id vid)
puts_.erase(sPuts);
}
-void
-DhtProxyServer::put(const std::shared_ptr& session)
+RequestStatus
+DhtProxyServer::put(restinio::request_handle_t request,
+ restinio::router::route_params_t params)
{
requestNum_++;
- const auto request = session->get_request();
- int content_length = std::stoi(request->get_header("Content-Length", "0"));
- auto hash = request->get_path_parameter("hash");
- InfoHash infoHash(hash);
+ dht::InfoHash infoHash(params["hash"].to_string());
if (!infoHash)
- infoHash = InfoHash::get(hash);
+ infoHash = dht::InfoHash::get(params["hash"].to_string());
- session->fetch(content_length,
- [=](const std::shared_ptr s, const restbed::Bytes& b)
- {
- try {
- if (dht_) {
- if(b.empty()) {
- std::string response("{\"err\":\"Missing parameters\"}");
- s->close(restbed::BAD_REQUEST, response);
- } else {
- std::string err;
- Json::Value root;
- Json::CharReaderBuilder rbuilder;
- auto* char_data = reinterpret_cast(b.data());
- auto reader = std::unique_ptr(rbuilder.newCharReader());
- if (reader->parse(char_data, char_data + b.size(), &root, &err)) {
- // Build the Value from json
- auto value = std::make_shared(root);
- bool permanent = root.isMember("permanent");
- std::cout << "Got put " << infoHash << " " << *value << " " << (permanent ? "permanent" : "") << std::endl;
+ if (!dht_){
+ auto response = this->initHttpResponse(
+ request->create_response(restinio::status_service_unavailable()));
+ response.set_body(this->RESP_MSG_SERVICE_UNAVAILABLE);
+ return response.done();
+ }
+ else if (request->body().empty()) {
+ auto response = this->initHttpResponse(
+ request->create_response(restinio::status_bad_request()));
+ response.set_body(this->RESP_MSG_MISSING_PARAMS);
+ return response.done();
+ }
- if (permanent) {
- std::string pushToken, clientId, platform;
- auto& pVal = root["permanent"];
- if (pVal.isObject()) {
- pushToken = pVal["key"].asString();
- clientId = pVal["client_id"].asString();
- platform = pVal["platform"].asString();
- }
- std::unique_lock lock(schedulerLock_);
- scheduler_.syncTime();
- auto timeout = scheduler_.time() + proxy::OP_TIMEOUT;
- auto vid = value->id;
- auto sPuts = puts_.emplace(infoHash, SearchPuts{}).first;
- auto r = sPuts->second.puts.emplace(vid, PermanentPut{});
- auto& pput = r.first->second;
- if (r.second) {
- pput.expireJob = scheduler_.add(timeout, [this, infoHash, vid]{
- std::cout << "Permanent put expired: " << infoHash << " " << vid << std::endl;
- cancelPut(infoHash, vid);
- });
-#ifdef OPENDHT_PUSH_NOTIFICATIONS
- if (not pushToken.empty()) {
- bool isAndroid = platform == "android";
- pput.expireNotifyJob = scheduler_.add(timeout - proxy::OP_MARGIN,
- [this, infoHash, vid, pushToken, clientId, isAndroid]
- {
- std::cout << "Permanent put refresh: " << infoHash << " " << vid << std::endl;
- Json::Value json;
- json["timeout"] = infoHash.toString();
- json["to"] = clientId;
- json["vid"] = std::to_string(vid);
- sendPushNotification(pushToken, std::move(json), isAndroid);
- });
- }
-#endif
- } else {
- scheduler_.edit(pput.expireJob, timeout);
- if (pput.expireNotifyJob)
- scheduler_.edit(pput.expireNotifyJob, timeout - proxy::OP_MARGIN);
- }
- lock.unlock();
- schedulerCv_.notify_one();
- }
+ try {
+ std::string err;
+ Json::Value root;
+ Json::CharReaderBuilder rbuilder;
+ auto* char_data = reinterpret_cast(request->body().data());
+ auto reader = std::unique_ptr(rbuilder.newCharReader());
- dht_->put(infoHash, value, [s, value](bool ok) {
- if (ok) {
- Json::StreamWriterBuilder wbuilder;
- wbuilder["commentStyle"] = "None";
- wbuilder["indentation"] = "";
- if (s->is_open())
- s->close(restbed::OK, Json::writeString(wbuilder, value->toJson()) + "\n");
- } else {
- if (s->is_open())
- s->close(restbed::BAD_GATEWAY, "{\"err\":\"put failed\"}");
- }
- }, time_point::max(), permanent);
- } else {
- s->close(restbed::BAD_REQUEST, "{\"err\":\"Incorrect JSON\"}");
- }
- }
- } else {
- s->close(restbed::SERVICE_UNAVAILABLE, "{\"err\":\"Incorrect DhtRunner\"}");
+ if (reader->parse(char_data, char_data + request->body().size(), &root, &err)){
+ // Build the dht::Value from json, NOTE: {"data": "base64value", ...}
+ auto value = std::make_shared(root);
+ bool permanent = root.isMember("permanent");
+ std::cout << "Got put " << infoHash << " " << *value <<
+ " " << (permanent ? "permanent" : "") << std::endl;
+ if (permanent) {
+ std::string pushToken, clientId, platform;
+ auto& pVal = root["permanent"];
+ if (pVal.isObject()) {
+ pushToken = pVal["key"].asString();
+ clientId = pVal["client_id"].asString();
+ platform = pVal["platform"].asString();
}
- } catch (const std::exception& e) {
- std::cout << "Error performing put: " << e.what() << std::endl;
- s->close(restbed::INTERNAL_SERVER_ERROR, "{\"err\":\"Internal server error\"}");
+ std::unique_lock lock(schedulerLock_);
+ scheduler_.syncTime();
+ auto timeout = scheduler_.time() + proxy::OP_TIMEOUT;
+ auto vid = value->id;
+ auto sPuts = puts_.emplace(infoHash, SearchPuts{}).first;
+ auto r = sPuts->second.puts.emplace(vid, PermanentPut{});
+ auto& pput = r.first->second;
+ if (r.second) {
+ pput.expireJob = scheduler_.add(timeout, [this, infoHash, vid]{
+ std::cout << "Permanent put expired: " << infoHash << " " << vid << std::endl;
+ cancelPut(infoHash, vid);
+ });
+#ifdef OPENDHT_PUSH_NOTIFICATIONS
+ if (not pushToken.empty()) {
+ bool isAndroid = platform == "android";
+ pput.expireNotifyJob = scheduler_.add(timeout - proxy::OP_MARGIN,
+ [this, infoHash, vid, pushToken, clientId, isAndroid]
+ {
+ std::cout << "Permanent put refresh: " << infoHash << " " << vid << std::endl;
+ Json::Value json;
+ json["timeout"] = infoHash.toString();
+ json["to"] = clientId;
+ json["vid"] = std::to_string(vid);
+ sendPushNotification(pushToken, std::move(json), isAndroid);
+ });
+ }
+#endif
+ } else {
+ scheduler_.edit(pput.expireJob, timeout);
+ if (pput.expireNotifyJob)
+ scheduler_.edit(pput.expireNotifyJob, timeout - proxy::OP_MARGIN);
+ }
+ lock.unlock();
+ schedulerCv_.notify_one();
}
+ dht_->put(infoHash, value, [this, request, value](bool ok){
+ if (ok){
+ Json::StreamWriterBuilder wbuilder;
+ wbuilder["commentStyle"] = "None";
+ wbuilder["indentation"] = "";
+ auto output = Json::writeString(jsonBuilder_, value->toJson()) + "\n";
+ std::cout << output << std::endl;
+ auto response = this->initHttpResponse(request->create_response());
+ response.append_body(output);
+ response.done();
+ } else {
+ auto response = this->initHttpResponse(request->create_response(
+ restinio::status_bad_gateway()));
+ response.set_body(this->RESP_MSG_PUT_FAILED);
+ response.done();
+ }
+
+ }, dht::time_point::max(), permanent);
+ } else {
+ auto response = this->initHttpResponse(
+ request->create_response(restinio::status_bad_request()));
+ response.set_body(this->RESP_MSG_INCORRECT_JSON);
+ return response.done();
}
- );
+ } catch (const std::exception& e) {
+ std::cout << "Error performing put: " << e.what() << std::endl;
+ auto response = this->initHttpResponse(
+ request->create_response(restinio::status_internal_server_error()));
+ response.set_body(this->RESP_MSG_INTERNAL_SERVER_ERRROR);
+ return response.done();
+ }
+ return restinio::request_handling_status_t::accepted;
}
#ifdef OPENDHT_PROXY_SERVER_IDENTITY
-void
-DhtProxyServer::putSigned(const std::shared_ptr& session) const
-{
- requestNum_++;
- const auto request = session->get_request();
- int content_length = std::stoi(request->get_header("Content-Length", "0"));
- auto hash = request->get_path_parameter("hash");
- InfoHash infoHash(hash);
- if (!infoHash)
- infoHash = InfoHash::get(hash);
- session->fetch(content_length,
- [=](const std::shared_ptr s, const restbed::Bytes& b)
- {
- try {
- if (dht_) {
- if(b.empty()) {
- std::string response("{\"err\":\"Missing parameters\"}");
- s->close(restbed::BAD_REQUEST, response);
- } else {
- std::string err;
- Json::Value root;
- Json::CharReaderBuilder rbuilder;
- auto* char_data = reinterpret_cast(b.data());
- auto reader = std::unique_ptr(rbuilder.newCharReader());
- if (reader->parse(char_data, char_data + b.size(), &root, &err)) {
- auto value = std::make_shared(root);
+//DhtProxyServer::putSigned()
- Json::StreamWriterBuilder wbuilder;
- wbuilder["commentStyle"] = "None";
- wbuilder["indentation"] = "";
- auto output = Json::writeString(wbuilder, value->toJson()) + "\n";
- dht_->putSigned(infoHash, value);
- s->close(restbed::OK, output);
- } else {
- s->close(restbed::BAD_REQUEST, "{\"err\":\"Incorrect JSON\"}");
- }
- }
- } else {
- s->close(restbed::SERVICE_UNAVAILABLE, "{\"err\":\"Incorrect DhtRunner\"}");
- }
- } catch (...) {
- s->close(restbed::INTERNAL_SERVER_ERROR, "{\"err\":\"Internal server error\"}");
- }
- }
- );
-}
+//DhtProxyServer::putEncrypted()
-void
-DhtProxyServer::putEncrypted(const std::shared_ptr& session) const
-{
- requestNum_++;
- const auto request = session->get_request();
- int content_length = std::stoi(request->get_header("Content-Length", "0"));
- auto hash = request->get_path_parameter("hash");
- InfoHash key(hash);
- if (!key)
- key = InfoHash::get(hash);
-
- session->fetch(content_length,
- [=](const std::shared_ptr s, const restbed::Bytes& b)
- {
- try {
- if (dht_) {
- if(b.empty()) {
- std::string response("{\"err\":\"Missing parameters\"}");
- s->close(restbed::BAD_REQUEST, response);
- } else {
- std::string err;
- Json::Value root;
- Json::CharReaderBuilder rbuilder;
- auto* char_data = reinterpret_cast(b.data());
- auto reader = std::unique_ptr(rbuilder.newCharReader());
- bool parsingSuccessful = reader->parse(char_data, char_data + b.size(), &root, &err);
- InfoHash to(root["to"].asString());
- if (parsingSuccessful && to) {
- auto value = std::make_shared(root);
- Json::StreamWriterBuilder wbuilder;
- wbuilder["commentStyle"] = "None";
- wbuilder["indentation"] = "";
- auto output = Json::writeString(wbuilder, value->toJson()) + "\n";
- dht_->putEncrypted(key, to, value);
- s->close(restbed::OK, output);
- } else {
- if(!parsingSuccessful)
- s->close(restbed::BAD_REQUEST, "{\"err\":\"Incorrect JSON\"}");
- else
- s->close(restbed::BAD_REQUEST, "{\"err\":\"No destination found\"}");
- }
- }
- } else {
- s->close(restbed::SERVICE_UNAVAILABLE, "{\"err\":\"Incorrect DhtRunner\"}");
- }
- } catch (...) {
- s->close(restbed::INTERNAL_SERVER_ERROR, "{\"err\":\"Internal server error\"}");
- }
- }
- );
-}
#endif // OPENDHT_PROXY_SERVER_IDENTITY
-void
-DhtProxyServer::handleOptionsMethod(const std::shared_ptr& session) const
+RequestStatus
+DhtProxyServer::options(restinio::request_handle_t request,
+ restinio::router::route_params_t params)
{
- requestNum_++;
+ this->requestNum_++;
#ifdef OPENDHT_PROXY_SERVER_IDENTITY
- const auto allowed = "OPTIONS, GET, POST, LISTEN, SIGN, ENCRYPT";
+ const auto methods = "OPTIONS, GET, POST, LISTEN, SIGN, ENCRYPT";
#else
- const auto allowed = "OPTIONS, GET, POST, LISTEN";
-#endif //OPENDHT_PROXY_SERVER_IDENTITY
- session->close(restbed::OK, {{"Access-Control-Allow-Methods", allowed},
- {"Access-Control-Allow-Headers", "content-type"},
- {"Access-Control-Max-Age", "86400"}});
+ const auto methods = "OPTIONS, GET, POST, LISTEN";
+#endif
+ auto response = initHttpResponse(request->create_response());
+ response.append_header(restinio::http_field::access_control_allow_methods, methods);
+ response.append_header(restinio::http_field::access_control_allow_headers, "content-type");
+ response.append_header(restinio::http_field::access_control_max_age, "86400");
+ return response.done();
}
-void
-DhtProxyServer::getFiltered(const std::shared_ptr& session) const
-{
- requestNum_++;
- const auto request = session->get_request();
- int content_length = std::stoi(request->get_header("Content-Length", "0"));
- auto hash = request->get_path_parameter("hash");
- auto value = request->get_path_parameter("value");
- session->fetch(content_length,
- [=](const std::shared_ptr s, const restbed::Bytes& /*b* */)
- {
- try {
- if (dht_) {
- InfoHash infoHash(hash);
- if (!infoHash) {
- infoHash = InfoHash::get(hash);
- }
- s->yield(restbed::OK, "", [=]( const std::shared_ptr< restbed::Session > s) {
- dht_->get(infoHash, [s](std::shared_ptr v) {
- // Send values as soon as we get them
- Json::StreamWriterBuilder wbuilder;
- wbuilder["commentStyle"] = "None";
- wbuilder["indentation"] = "";
- auto output = Json::writeString(wbuilder, v->toJson()) + "\n";
- s->yield(output, [](const std::shared_ptr /*session*/){ });
- return true;
- }, [s](bool /*ok* */) {
- // Communication is finished
- s->close();
- }, {}, value);
- });
- } else {
- s->close(restbed::SERVICE_UNAVAILABLE, "{\"err\":\"Incorrect DhtRunner\"}");
- }
- } catch (...) {
- s->close(restbed::INTERNAL_SERVER_ERROR, "{\"err\":\"Internal server error\"}");
- }
- }
- );
-}
+//DhtProxyServer::getFiltered()
void
DhtProxyServer::removeClosedListeners(bool testSession)
@@ -896,7 +479,7 @@ DhtProxyServer::removeClosedListeners(bool testSession)
std::lock_guard lock(lockListener_);
auto listener = currentListeners_.begin();
while (listener != currentListeners_.end()) {
- auto cancel = dht_ and (not testSession or listener->session->is_closed());
+ auto cancel = dht_ and (not testSession);// or listener->session->is_closed());
if (cancel) {
dht_->cancelListen(listener->hash, std::move(listener->token));
// Remove listener if unused