From a520657fac2d7d6db21590cc1cb6755ab002cfac Mon Sep 17 00:00:00 2001 From: Seva Date: Tue, 14 May 2019 13:09:21 -0400 Subject: [PATCH 01/17] dhtproxy: switch server to restinio rewrite options, getNodeInfo, get & put opendht: bump to c++14 --- CMakeLists.txt | 30 +- README.md | 6 +- cmake/FindRestinio.cmake | 14 + include/opendht/dht_proxy_server.h | 53 +- python/opendht.pyx | 2 +- python/setup.py.in | 4 +- src/dht_proxy_server.cpp | 845 ++++++++--------------------- 7 files changed, 295 insertions(+), 659 deletions(-) create mode 100644 cmake/FindRestinio.cmake 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 From eef2838a12938480e0a70bc9b738869bc008569f Mon Sep 17 00:00:00 2001 From: Seva Date: Thu, 16 May 2019 12:31:58 -0400 Subject: [PATCH 02/17] dhtproxy: rewrite stats with restinio --- include/opendht/dht_proxy_server.h | 9 ++-- src/dht_proxy_server.cpp | 82 +++++++++++++++++++----------- 2 files changed, 57 insertions(+), 34 deletions(-) diff --git a/include/opendht/dht_proxy_server.h b/include/opendht/dht_proxy_server.h index 1e3d1a02..510dfb8e 100644 --- a/include/opendht/dht_proxy_server.h +++ b/include/opendht/dht_proxy_server.h @@ -154,7 +154,8 @@ private: * Result: HTTP 200, body: Node infos in JSON format * @param session */ - //void getStats(const std::shared_ptr& session) const; + RequestStatus getStats(restinio::request_handle_t request, + restinio::router::route_params_t params); /** * Return Values of an infoHash @@ -288,7 +289,6 @@ private: using time_point = clock::time_point; std::thread server_thread {}; - //std::unique_ptr service_; std::shared_ptr dht_; Json::StreamWriterBuilder jsonBuilder_; @@ -303,10 +303,8 @@ private: mutable NodeInfo nodeInfo_ {}; // Handle client quit for listen. - // NOTE: can be simplified when we will supports restbed 5.0 std::thread listenThread_; struct SessionToHashToken { - //std::shared_ptr session; InfoHash hash; std::future token; }; @@ -333,7 +331,8 @@ private: proxy::ListenToken tokenPushNotif_ {0}; #endif //OPENDHT_PUSH_NOTIFICATIONS - const std::string RESP_MSG_INCORRECT_JSON = "{\"err:\":\"Incorrect JSON\"}"; + const std::string RESP_MSG_JSON_NOT_ENABLED = "{\"err\":\"JSON not enabled on this instance\"}"; + const std::string RESP_MSG_JSON_INCORRECT = "{\"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\"}"; diff --git a/src/dht_proxy_server.cpp b/src/dht_proxy_server.cpp index d9c73f9a..6efcb243 100644 --- a/src/dht_proxy_server.cpp +++ b/src/dht_proxy_server.cpp @@ -79,29 +79,27 @@ DhtProxyServer::DhtProxyServer(std::shared_ptr dht, in_port_t port , settings.address("0.0.0.0"); settings.port(port); settings.request_handler(this->createRestRouter()); + // time limits settings.read_next_http_message_timelimit(10s); settings.write_http_response_timelimit(10s); - settings.handle_request_timeout(10s); + std::chrono::milliseconds timeout_request(std::numeric_limits::max()); + settings.handle_request_timeout(timeout_request); + // socket options settings.socket_options_setter([](auto & options){ options.set_option(asio::ip::tcp::no_delay{true}); }); try { 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; + std::cerr << "Error starting RESTinio: " << ex.what() << std::endl; } }); - - /* listenThread_ = std::thread([this]() { - while (not service_->is_up() and not stopListeners) { + while (not server_thread.joinable() and not stopListeners) { std::this_thread::sleep_for(std::chrono::seconds(1)); } - while (service_->is_up() and not stopListeners) { + while (server_thread.joinable() and not stopListeners) { removeClosedListeners(); std::this_thread::sleep_for(std::chrono::seconds(1)); } @@ -109,10 +107,10 @@ DhtProxyServer::DhtProxyServer(std::shared_ptr dht, in_port_t port , removeClosedListeners(false); }); schedulerThread_ = std::thread([this]() { - while (not service_->is_up() and not stopListeners) { + while (not server_thread.joinable() and not stopListeners) { std::this_thread::sleep_for(std::chrono::seconds(1)); } - while (service_->is_up() and not stopListeners) { + while (server_thread.joinable() and not stopListeners) { std::unique_lock lock(schedulerLock_); auto next = scheduler_.run(); if (next == time_point::max()) @@ -121,12 +119,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 (stopListeners) + return; + if (server_thread.joinable()) + updateStats(); // Refresh stats cache auto newInfo = dht_->getNodeInfo(); { @@ -147,17 +145,12 @@ 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; - } + + std::lock_guard lock(lockListener_); + auto listener = currentListeners_.begin(); + while (listener != currentListeners_.end()) { + ++listener; } - */ stopListeners = true; schedulerCv_.notify_all(); // listenThreads_ will stop because there is no more sessions @@ -203,6 +196,7 @@ DhtProxyServer::createRestRouter() 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("/stats", std::bind(&DhtProxyServer::getStats, 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)); @@ -229,7 +223,38 @@ DhtProxyServer::getNodeInfo(restinio::request_handle_t request, return response.done(); } -//DhtProxyServer::getStats() +RequestStatus +DhtProxyServer::getStats(restinio::request_handle_t request, + restinio::router::route_params_t params) +{ + requestNum_++; + try { + if (dht_) { +#ifdef OPENDHT_JSONCPP + auto output = Json::writeString(jsonBuilder_, stats_.toJson()) + "\n"; + auto response = this->initHttpResponse(request->create_response()); + response.append_body(output); + response.done(); +#else + auto response = this->initHttpResponse( + request->create_response(restinio::status_not_found())); + response.set_body(this->RESP_MSG_JSON_NOT_ENABLED); + return response.done(); +#endif + } else { + auto response = this->initHttpResponse( + request->create_response(restinio::status_service_unavailable())); + response.set_body(this->RESP_MSG_SERVICE_UNAVAILABLE); + return response.done(); + } + } catch (...) { + 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; +} RequestStatus DhtProxyServer::get(restinio::request_handle_t request, @@ -417,7 +442,6 @@ DhtProxyServer::put(restinio::request_handle_t request, 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(); @@ -432,7 +456,7 @@ DhtProxyServer::put(restinio::request_handle_t request, } else { auto response = this->initHttpResponse( request->create_response(restinio::status_bad_request())); - response.set_body(this->RESP_MSG_INCORRECT_JSON); + response.set_body(this->RESP_MSG_JSON_INCORRECT); return response.done(); } } catch (const std::exception& e) { @@ -479,7 +503,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 not server_thread.joinable()); if (cancel) { dht_->cancelListen(listener->hash, std::move(listener->token)); // Remove listener if unused From 4a4becdbecf143717a5f4a5f0fb5e6250be66bc6 Mon Sep 17 00:00:00 2001 From: Seva Date: Thu, 16 May 2019 14:09:13 -0400 Subject: [PATCH 03/17] dhtproxy: rewrite un/subscribe with restinio --- include/opendht/dht_proxy_server.h | 11 +- src/dht_proxy_server.cpp | 188 ++++++++++++++++++++++++++++- 2 files changed, 191 insertions(+), 8 deletions(-) diff --git a/include/opendht/dht_proxy_server.h b/include/opendht/dht_proxy_server.h index 510dfb8e..987f8c82 100644 --- a/include/opendht/dht_proxy_server.h +++ b/include/opendht/dht_proxy_server.h @@ -258,7 +258,9 @@ private: * so you need to refresh the operation each six hours. * @param session */ - //void subscribe(const std::shared_ptr& session); + RequestStatus subscribe(restinio::request_handle_t request, + restinio::router::route_params_t params); + /** * Unsubscribe to push notifications for an iOS or Android device. * Method: UNSUBSCRIBE "/{InfoHash: .*}" @@ -266,13 +268,15 @@ private: * Return: nothing * @param session */ - //void unsubscribe(const std::shared_ptr& session); + RequestStatus unsubscribe(restinio::request_handle_t request, + restinio::router::route_params_t params); + /** * Send a push notification via a gorush push gateway * @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) const; /** * Remove a push listener between a client and a hash @@ -331,6 +335,7 @@ private: proxy::ListenToken tokenPushNotif_ {0}; #endif //OPENDHT_PUSH_NOTIFICATIONS + const std::string RESP_MSG_NO_TOKEN = "{\"err\":\"No token\"}"; const std::string RESP_MSG_JSON_NOT_ENABLED = "{\"err\":\"JSON not enabled on this instance\"}"; const std::string RESP_MSG_JSON_INCORRECT = "{\"err:\":\"Incorrect JSON\"}"; const std::string RESP_MSG_SERVICE_UNAVAILABLE = "{\"err\":\"Incorrect DhtRunner\"}"; diff --git a/src/dht_proxy_server.cpp b/src/dht_proxy_server.cpp index 6efcb243..07ecbf02 100644 --- a/src/dht_proxy_server.cpp +++ b/src/dht_proxy_server.cpp @@ -194,12 +194,18 @@ DhtProxyServer::createRestRouter() { using namespace std::placeholders; auto router = std::make_unique(); + router->http_get("/", std::bind(&DhtProxyServer::getNodeInfo, this, _1, _2)); router->add_handler(restinio::http_method_t::http_options, "/:hash", std::bind(&DhtProxyServer::options, this, _1, _2)); router->http_get("/stats", std::bind(&DhtProxyServer::getStats, 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)); +#ifdef OPENDHT_PUSH_NOTIFICATIONS + router->add_handler(restinio::http_method_t::http_subscribe, + "/:hash", std::bind(&DhtProxyServer::subscribe, this, _1, _2)); + router->add_handler(restinio::http_method_t::http_unsubscribe, + "/:hash", std::bind(&DhtProxyServer::unsubscribe, this, _1, _2)); +#endif //OPENDHT_PUSH_NOTIFICATIONS return router; } @@ -303,8 +309,182 @@ struct DhtProxyServer::PushListener { bool isAndroid; }; -//DhtProxyServer::subscribe() -//DhtProxyServer::unsubscribe() +RequestStatus +subscribe(restinio::request_handle_t request, restinio::router::route_params_t params) +{ + requestNum_++; + + dht::InfoHash infoHash(params["hash"].to_string()); + if (!infoHash) + 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(); + } + + 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)) { + auto response = this->initHttpResponse( + request->create_response(restinio::status_bad_request())); + response.set_body(this->RESP_MSG_JSON_INCORRECT); + return response.done(); + } + auto pushToken = root["key"].asString(); + if (pushToken.empty()) { + auto response = this->initHttpResponse( + request->create_response(restinio::status_bad_request())); + response.set_body(this->RESP_MSG_NO_TOKEN); + return response.done(); + } + 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); + + auto response = std::make_shared( + this->initHttpResponse(request->create_response())); + response->flush(); + + if (!root.isMember("refresh") or !root["refresh"].asBool()) { + 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(); + }); + } else { + response->append_chunk("{}\n"); + response->done(); + } + 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(); + auto response = this->initHttpResponse(request->create_response()); + response.set_body("{}\n"); + return response.done(); + } catch (...) { + 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; +} + +RequestStatus +unsubscribe(restinio::request_handle_t request, restinio::router::route_params_t params) +{ + requestNum_++; + + dht::InfoHash infoHash(params["hash"].to_string()); + if (!infoHash) + 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(); + } + + 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)) { + auto response = this->initHttpResponse( + request->create_response(restinio::status_bad_request())); + response.set_body(this->RESP_MSG_JSON_INCORRECT); + return response.done(); + } + auto pushToken = root["key"].asString(); + if (pushToken.empty()) return; + auto clientId = root["client_id"].asString(); + + cancelPushListen(pushToken, infoHash, clientId); + auto response = this->initHttpResponse(request->create_response()); + return response.done(); + } catch (...) { + auto response = this->initHttpResponse( + request->create_response(restinio::status_internal_server_error())); + response.set_body(this->RESP_MSG_INTERNAL_SERVER_ERRROR); + return response.done(); + } +} void DhtProxyServer::cancelPushListen(const std::string& pushToken, const dht::InfoHash& key, const std::string& clientId) @@ -334,8 +514,6 @@ DhtProxyServer::cancelPushListen(const std::string& pushToken, const dht::InfoHa } } -//DhtProxyServer::sendPushNotification() - #endif //OPENDHT_PUSH_NOTIFICATIONS void From a31cecdeace51dadf2fdc341dd88bb8ececfe4f3 Mon Sep 17 00:00:00 2001 From: Seva Date: Thu, 16 May 2019 15:19:26 -0400 Subject: [PATCH 04/17] dhtproxy: rewrite getFiltered with restinio --- include/opendht/dht_proxy_server.h | 5 +- src/dht_proxy_server.cpp | 81 +++++++++++++++++++++++------- 2 files changed, 66 insertions(+), 20 deletions(-) diff --git a/include/opendht/dht_proxy_server.h b/include/opendht/dht_proxy_server.h index 987f8c82..c64eacf1 100644 --- a/include/opendht/dht_proxy_server.h +++ b/include/opendht/dht_proxy_server.h @@ -192,7 +192,7 @@ private: * @param session */ RequestStatus put(restinio::request_handle_t request, - restinio::router::route_params_t params); + restinio::router::route_params_t params); void cancelPut(const InfoHash& key, Value::Id vid); @@ -230,7 +230,8 @@ private: * On error: HTTP 503, body: {"err":"xxxx"} * @param session */ - //void getFiltered(const std::shared_ptr& session) const; + RequestStatus getFiltered(restinio::request_handle_t request, + restinio::router::route_params_t params); /** * Respond allowed Methods diff --git a/src/dht_proxy_server.cpp b/src/dht_proxy_server.cpp index 07ecbf02..765c97cd 100644 --- a/src/dht_proxy_server.cpp +++ b/src/dht_proxy_server.cpp @@ -195,8 +195,6 @@ DhtProxyServer::createRestRouter() using namespace std::placeholders; auto router = std::make_unique(); router->http_get("/", std::bind(&DhtProxyServer::getNodeInfo, this, _1, _2)); - router->add_handler(restinio::http_method_t::http_options, - "/:hash", std::bind(&DhtProxyServer::options, this, _1, _2)); router->http_get("/stats", std::bind(&DhtProxyServer::getStats, 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)); @@ -206,6 +204,9 @@ DhtProxyServer::createRestRouter() router->add_handler(restinio::http_method_t::http_unsubscribe, "/:hash", std::bind(&DhtProxyServer::unsubscribe, this, _1, _2)); #endif //OPENDHT_PUSH_NOTIFICATIONS + router->add_handler(restinio::http_method_t::http_options, + "/:hash", std::bind(&DhtProxyServer::options, this, _1, _2)); + router->http_get("/:hash/:value", std::bind(&DhtProxyServer::getFiltered, this, _1, _2)); return router; } @@ -281,16 +282,22 @@ DhtProxyServer::get(restinio::request_handle_t request, 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(); - }); + try { + 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(); + }); + } catch (const std::exception& e) { + 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; } @@ -310,7 +317,8 @@ struct DhtProxyServer::PushListener { }; RequestStatus -subscribe(restinio::request_handle_t request, restinio::router::route_params_t params) +DhtProxyServer::subscribe(restinio::request_handle_t request, + restinio::router::route_params_t params) { requestNum_++; @@ -443,7 +451,8 @@ subscribe(restinio::request_handle_t request, restinio::router::route_params_t p } RequestStatus -unsubscribe(restinio::request_handle_t request, restinio::router::route_params_t params) +DhtProxyServer::unsubscribe(restinio::request_handle_t request, + restinio::router::route_params_t params) { requestNum_++; @@ -616,9 +625,6 @@ DhtProxyServer::put(restinio::request_handle_t request, } 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"; auto response = this->initHttpResponse(request->create_response()); response.append_body(output); @@ -672,7 +678,46 @@ DhtProxyServer::options(restinio::request_handle_t request, return response.done(); } -//DhtProxyServer::getFiltered() +RequestStatus +DhtProxyServer::getFiltered(restinio::request_handle_t request, + restinio::router::route_params_t params) +{ + requestNum_++; + auto value = params["value"].to_string(); + dht::InfoHash infoHash(params["hash"].to_string()); + if (!infoHash) + 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(); + try { + 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(); + }, + {}, value + ); + } catch (const std::exception& e) { + 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; +} void DhtProxyServer::removeClosedListeners(bool testSession) From efe869fe42f0ffbb6999ffcbd74f366542bcdb3d Mon Sep 17 00:00:00 2001 From: Seva Date: Thu, 16 May 2019 16:06:47 -0400 Subject: [PATCH 05/17] dhtproxy: rewrite putSigned with restinio --- include/opendht/dht_proxy_server.h | 5 +++ src/dht_proxy_server.cpp | 66 ++++++++++++++++++++++++++++-- 2 files changed, 68 insertions(+), 3 deletions(-) diff --git a/include/opendht/dht_proxy_server.h b/include/opendht/dht_proxy_server.h index c64eacf1..900f50f5 100644 --- a/include/opendht/dht_proxy_server.h +++ b/include/opendht/dht_proxy_server.h @@ -207,6 +207,8 @@ private: * @param session */ //void putSigned(const std::shared_ptr& session) const; + RequestStatus putSigned(restinio::request_handle_t request, + restinio::router::route_params_t params) const; /** * Put a value to encrypt by the proxy on the DHT @@ -218,6 +220,9 @@ private: * @param session */ //void putEncrypted(const std::shared_ptr& session) const; + //RequestStatus putEncrypted(restinio::request_handle_t request, + // restinio::router::route_params_t params); + #endif // OPENDHT_PROXY_SERVER_IDENTITY /** diff --git a/src/dht_proxy_server.cpp b/src/dht_proxy_server.cpp index 765c97cd..32813a5f 100644 --- a/src/dht_proxy_server.cpp +++ b/src/dht_proxy_server.cpp @@ -204,6 +204,9 @@ DhtProxyServer::createRestRouter() router->add_handler(restinio::http_method_t::http_unsubscribe, "/:hash", std::bind(&DhtProxyServer::unsubscribe, this, _1, _2)); #endif //OPENDHT_PUSH_NOTIFICATIONS +#ifdef OPENDHT_PROXY_SERVER_IDENTITY + router->http_post("/:hash/sign", std::bind(&DhtProxyServer::putSigned, this, _1, _2)); +#endif // OPENDHT_PROXY_SERVER_IDENTITY router->add_handler(restinio::http_method_t::http_options, "/:hash", std::bind(&DhtProxyServer::options, this, _1, _2)); router->http_get("/:hash/:value", std::bind(&DhtProxyServer::getFiltered, this, _1, _2)); @@ -575,7 +578,6 @@ DhtProxyServer::put(restinio::request_handle_t request, auto reader = std::unique_ptr(rbuilder.newCharReader()); 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 << @@ -635,7 +637,6 @@ DhtProxyServer::put(restinio::request_handle_t request, response.set_body(this->RESP_MSG_PUT_FAILED); response.done(); } - }, dht::time_point::max(), permanent); } else { auto response = this->initHttpResponse( @@ -655,7 +656,66 @@ DhtProxyServer::put(restinio::request_handle_t request, #ifdef OPENDHT_PROXY_SERVER_IDENTITY -//DhtProxyServer::putSigned() +RequestStatus DhtProxyServer::putSigned(restinio::request_handle_t request, + restinio::router::route_params_t params) const +{ + requestNum_++; + dht::InfoHash infoHash(params["hash"].to_string()); + if (!infoHash) + 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(); + } + 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(); + } + + 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()); + + if (reader->parse(char_data, char_data + request->body().size(), &root, &err)){ + + auto value = std::make_shared(root); + + dht_->putSigned(infoHash, value, [this, request, value](bool ok){ + if (ok){ + auto output = Json::writeString(jsonBuilder_, value->toJson()) + "\n"; + 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(); + } + }); + } else { + auto response = this->initHttpResponse( + request->create_response(restinio::status_bad_request())); + response.set_body(this->RESP_MSG_JSON_INCORRECT); + 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; +} //DhtProxyServer::putEncrypted() From 5f393490a577612cc73ced122c322af847cc5c53 Mon Sep 17 00:00:00 2001 From: Seva Date: Thu, 16 May 2019 17:15:31 -0400 Subject: [PATCH 06/17] dhtproxy: rewrite putEncrypted with restinio --- include/opendht/dht_proxy_server.h | 7 +- src/dht_proxy_server.cpp | 148 +++++++++++++++++++++-------- 2 files changed, 110 insertions(+), 45 deletions(-) diff --git a/include/opendht/dht_proxy_server.h b/include/opendht/dht_proxy_server.h index 900f50f5..23bc19b5 100644 --- a/include/opendht/dht_proxy_server.h +++ b/include/opendht/dht_proxy_server.h @@ -206,7 +206,6 @@ private: * HTTP 400, body: {"err":"xxxx"} if bad json * @param session */ - //void putSigned(const std::shared_ptr& session) const; RequestStatus putSigned(restinio::request_handle_t request, restinio::router::route_params_t params) const; @@ -219,9 +218,8 @@ private: * HTTP 400, body: {"err":"xxxx"} if bad json * @param session */ - //void putEncrypted(const std::shared_ptr& session) const; - //RequestStatus putEncrypted(restinio::request_handle_t request, - // restinio::router::route_params_t params); + RequestStatus putEncrypted(restinio::request_handle_t request, + restinio::router::route_params_t params); #endif // OPENDHT_PROXY_SERVER_IDENTITY @@ -341,6 +339,7 @@ private: proxy::ListenToken tokenPushNotif_ {0}; #endif //OPENDHT_PUSH_NOTIFICATIONS + const std::string RESP_MSG_DESTINATION_NOT_FOUND = "{\"err\":\"No destination found\"}"; const std::string RESP_MSG_NO_TOKEN = "{\"err\":\"No token\"}"; const std::string RESP_MSG_JSON_NOT_ENABLED = "{\"err\":\"JSON not enabled on this instance\"}"; const std::string RESP_MSG_JSON_INCORRECT = "{\"err:\":\"Incorrect JSON\"}"; diff --git a/src/dht_proxy_server.cpp b/src/dht_proxy_server.cpp index 32813a5f..ceb363ce 100644 --- a/src/dht_proxy_server.cpp +++ b/src/dht_proxy_server.cpp @@ -60,7 +60,7 @@ DhtProxyServer::DhtProxyServer(std::shared_ptr dht, in_port_t port , //service_ = std::unique_ptr(new restbed::Service()); std::cout << "Running DHT proxy server on port " << port << std::endl; - if (not pushServer.empty()) { + if (not pushServer.empty()){ #ifdef OPENDHT_PUSH_NOTIFICATIONS std::cout << "Using push notification server: " << pushServer << std::endl; #else @@ -71,7 +71,7 @@ DhtProxyServer::DhtProxyServer(std::shared_ptr dht, in_port_t port , jsonBuilder_["commentStyle"] = "None"; jsonBuilder_["indentation"] = ""; - server_thread = std::thread([this, port]() { + server_thread = std::thread([this, port](){ using namespace std::chrono; auto maxThreads = std::thread::hardware_concurrency() - 1; auto restThreads = maxThreads > 1 ? maxThreads : 1; @@ -91,26 +91,26 @@ DhtProxyServer::DhtProxyServer(std::shared_ptr dht, in_port_t port , try { restinio::run(std::move(settings)); } - catch(const std::exception &ex) { + catch(const std::exception &ex){ std::cerr << "Error starting RESTinio: " << ex.what() << std::endl; } }); - listenThread_ = std::thread([this]() { - while (not server_thread.joinable() and not stopListeners) { + listenThread_ = std::thread([this](){ + while (not server_thread.joinable() and not stopListeners){ std::this_thread::sleep_for(std::chrono::seconds(1)); } - while (server_thread.joinable() and not stopListeners) { + while (server_thread.joinable() and not stopListeners){ removeClosedListeners(); std::this_thread::sleep_for(std::chrono::seconds(1)); } // Remove last listeners removeClosedListeners(false); }); - schedulerThread_ = std::thread([this]() { - while (not server_thread.joinable() and not stopListeners) { + schedulerThread_ = std::thread([this](){ + while (not server_thread.joinable() and not stopListeners){ std::this_thread::sleep_for(std::chrono::seconds(1)); } - while (server_thread.joinable() and not stopListeners) { + while (server_thread.joinable() and not stopListeners){ std::unique_lock lock(schedulerLock_); auto next = scheduler_.run(); if (next == time_point::max()) @@ -148,7 +148,7 @@ DhtProxyServer::stop() std::lock_guard lock(lockListener_); auto listener = currentListeners_.begin(); - while (listener != currentListeners_.end()) { + while (listener != currentListeners_.end()){ ++listener; } stopListeners = true; @@ -206,6 +206,7 @@ DhtProxyServer::createRestRouter() #endif //OPENDHT_PUSH_NOTIFICATIONS #ifdef OPENDHT_PROXY_SERVER_IDENTITY router->http_post("/:hash/sign", std::bind(&DhtProxyServer::putSigned, this, _1, _2)); + router->http_post("/:hash/encrypt", std::bind(&DhtProxyServer::putEncrypted, this, _1, _2)); #endif // OPENDHT_PROXY_SERVER_IDENTITY router->add_handler(restinio::http_method_t::http_options, "/:hash", std::bind(&DhtProxyServer::options, this, _1, _2)); @@ -239,7 +240,7 @@ DhtProxyServer::getStats(restinio::request_handle_t request, { requestNum_++; try { - if (dht_) { + if (dht_){ #ifdef OPENDHT_JSONCPP auto output = Json::writeString(jsonBuilder_, stats_.toJson()) + "\n"; auto response = this->initHttpResponse(request->create_response()); @@ -257,7 +258,7 @@ DhtProxyServer::getStats(restinio::request_handle_t request, response.set_body(this->RESP_MSG_SERVICE_UNAVAILABLE); return response.done(); } - } catch (...) { + } catch (...){ auto response = this->initHttpResponse( request->create_response(restinio::status_internal_server_error())); response.set_body(this->RESP_MSG_INTERNAL_SERVER_ERRROR); @@ -295,7 +296,7 @@ DhtProxyServer::get(restinio::request_handle_t request, [response] (bool /*ok*/){ response->done(); }); - } catch (const std::exception& e) { + } catch (const std::exception& e){ auto response = this->initHttpResponse( request->create_response(restinio::status_internal_server_error())); response.set_body(this->RESP_MSG_INTERNAL_SERVER_ERRROR); @@ -342,14 +343,14 @@ DhtProxyServer::subscribe(restinio::request_handle_t request, 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)) { + if (!reader->parse(char_data, char_data + b.size(), &root, &err)){ auto response = this->initHttpResponse( request->create_response(restinio::status_bad_request())); response.set_body(this->RESP_MSG_JSON_INCORRECT); return response.done(); } auto pushToken = root["key"].asString(); - if (pushToken.empty()) { + if (pushToken.empty()){ auto response = this->initHttpResponse( request->create_response(restinio::status_bad_request())); response.set_body(this->RESP_MSG_NO_TOKEN); @@ -370,8 +371,8 @@ DhtProxyServer::subscribe(restinio::request_handle_t request, // 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) { + for (auto& listener: listeners->second){ + if (listener.clientId == clientId){ scheduler_.edit(listener.expireJob, timeout); scheduler_.edit(listener.expireNotifyJob, timeout - proxy::OP_MARGIN); @@ -379,7 +380,7 @@ DhtProxyServer::subscribe(restinio::request_handle_t request, this->initHttpResponse(request->create_response())); response->flush(); - if (!root.isMember("refresh") or !root["refresh"].asBool()) { + if (!root.isMember("refresh") or !root["refresh"].asBool()){ dht_->get(infoHash, [this, response](const dht::Sp& value){ auto output = Json::writeString(jsonBuilder_, value->toJson()) + "\n"; response->append_chunk(output); @@ -406,15 +407,15 @@ DhtProxyServer::subscribe(restinio::request_handle_t request, // 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]() { + [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) { + if (expired and values.size() < 3){ std::stringstream ss; - for(size_t i = 0; i < values.size(); ++i) { + for(size_t i = 0; i < values.size(); ++i){ if(i != 0) ss << ","; ss << values[i]->id; } @@ -444,7 +445,7 @@ DhtProxyServer::subscribe(restinio::request_handle_t request, auto response = this->initHttpResponse(request->create_response()); response.set_body("{}\n"); return response.done(); - } catch (...) { + } catch (...){ auto response = this->initHttpResponse( request->create_response(restinio::status_internal_server_error())); response.set_body(this->RESP_MSG_INTERNAL_SERVER_ERRROR); @@ -477,7 +478,7 @@ DhtProxyServer::unsubscribe(restinio::request_handle_t request, 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)) { + if (!reader->parse(char_data, char_data + b.size(), &root, &err)){ auto response = this->initHttpResponse( request->create_response(restinio::status_bad_request())); response.set_body(this->RESP_MSG_JSON_INCORRECT); @@ -490,7 +491,7 @@ DhtProxyServer::unsubscribe(restinio::request_handle_t request, cancelPushListen(pushToken, infoHash, clientId); auto response = this->initHttpResponse(request->create_response()); return response.done(); - } catch (...) { + } catch (...){ auto response = this->initHttpResponse( request->create_response(restinio::status_internal_server_error())); response.set_body(this->RESP_MSG_INTERNAL_SERVER_ERRROR); @@ -509,8 +510,8 @@ DhtProxyServer::cancelPushListen(const std::string& pushToken, const dht::InfoHa auto listeners = pushListener->second.listeners.find(key); if (listeners == pushListener->second.listeners.end()) return; - for (auto listener = listeners->second.begin(); listener != listeners->second.end();) { - if (listener->clientId == clientId) { + for (auto listener = listeners->second.begin(); listener != listeners->second.end();){ + if (listener->clientId == clientId){ if (dht_) dht_->cancelListen(key, std::move(listener->internalToken)); listener = listeners->second.erase(listener); @@ -518,10 +519,10 @@ DhtProxyServer::cancelPushListen(const std::string& pushToken, const dht::InfoHa ++listener; } } - if (listeners->second.empty()) { + if (listeners->second.empty()){ pushListener->second.listeners.erase(listeners); } - if (pushListener->second.listeners.empty()) { + if (pushListener->second.listeners.empty()){ pushListeners_.erase(pushListener); } } @@ -563,7 +564,7 @@ DhtProxyServer::put(restinio::request_handle_t request, response.set_body(this->RESP_MSG_SERVICE_UNAVAILABLE); return response.done(); } - else if (request->body().empty()) { + else if (request->body().empty()){ auto response = this->initHttpResponse( request->create_response(restinio::status_bad_request())); response.set_body(this->RESP_MSG_MISSING_PARAMS); @@ -582,10 +583,10 @@ DhtProxyServer::put(restinio::request_handle_t request, bool permanent = root.isMember("permanent"); std::cout << "Got put " << infoHash << " " << *value << " " << (permanent ? "permanent" : "") << std::endl; - if (permanent) { + if (permanent){ std::string pushToken, clientId, platform; auto& pVal = root["permanent"]; - if (pVal.isObject()) { + if (pVal.isObject()){ pushToken = pVal["key"].asString(); clientId = pVal["client_id"].asString(); platform = pVal["platform"].asString(); @@ -597,13 +598,13 @@ DhtProxyServer::put(restinio::request_handle_t request, auto sPuts = puts_.emplace(infoHash, SearchPuts{}).first; auto r = sPuts->second.puts.emplace(vid, PermanentPut{}); auto& pput = r.first->second; - if (r.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()) { + if (not pushToken.empty()){ bool isAndroid = platform == "android"; pput.expireNotifyJob = scheduler_.add(timeout - proxy::OP_MARGIN, [this, infoHash, vid, pushToken, clientId, isAndroid] @@ -644,7 +645,7 @@ DhtProxyServer::put(restinio::request_handle_t request, response.set_body(this->RESP_MSG_JSON_INCORRECT); return response.done(); } - } catch (const std::exception& e) { + } 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())); @@ -670,7 +671,7 @@ RequestStatus DhtProxyServer::putSigned(restinio::request_handle_t request, response.set_body(this->RESP_MSG_SERVICE_UNAVAILABLE); return response.done(); } - else if (request->body().empty()) { + else if (request->body().empty()){ auto response = this->initHttpResponse( request->create_response(restinio::status_bad_request())); response.set_body(this->RESP_MSG_MISSING_PARAMS); @@ -707,7 +708,7 @@ RequestStatus DhtProxyServer::putSigned(restinio::request_handle_t request, response.set_body(this->RESP_MSG_JSON_INCORRECT); return response.done(); } - } catch (const std::exception& e) { + } 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())); @@ -717,7 +718,72 @@ RequestStatus DhtProxyServer::putSigned(restinio::request_handle_t request, return restinio::request_handling_status_t::accepted; } -//DhtProxyServer::putEncrypted() +RequestStatus +DhtProxyServer::putEncrypted(restinio::request_handle_t request, + restinio::router::route_params_t params) +{ + requestNum_++; + dht::InfoHash infoHash(params["hash"].to_string()); + if (!infoHash) + 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(); + } + 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(); + } + + 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()); + + if (reader->parse(char_data, char_data + request->body().size(), &root, &err)){ + InfoHash to(root["to"].asString()); + if (!to){ + auto response = this->initHttpResponse( + request->create_response(restinio::status_bad_request())); + response.set_body(this->RESP_MSG_DESTINATION_NOT_FOUND); + return response.done(); + } + auto value = std::make_shared(root); + dht_->putEncrypted(infoHash, to, value, [this, request, value](bool ok){ + if (ok){ + auto output = Json::writeString(jsonBuilder_, value->toJson()) + "\n"; + 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(); + } + }); + } else { + auto response = this->initHttpResponse( + request->create_response(restinio::status_bad_request())); + response.set_body(this->RESP_MSG_JSON_INCORRECT); + 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; +} #endif // OPENDHT_PROXY_SERVER_IDENTITY @@ -770,7 +836,7 @@ DhtProxyServer::getFiltered(restinio::request_handle_t request, }, {}, value ); - } catch (const std::exception& e) { + } catch (const std::exception& e){ auto response = this->initHttpResponse( request->create_response(restinio::status_internal_server_error())); response.set_body(this->RESP_MSG_INTERNAL_SERVER_ERRROR); @@ -785,9 +851,9 @@ DhtProxyServer::removeClosedListeners(bool testSession) // clean useless listeners std::lock_guard lock(lockListener_); auto listener = currentListeners_.begin(); - while (listener != currentListeners_.end()) { + while (listener != currentListeners_.end()){ auto cancel = dht_ and (not testSession or not server_thread.joinable()); - if (cancel) { + if (cancel){ dht_->cancelListen(listener->hash, std::move(listener->token)); // Remove listener if unused listener = currentListeners_.erase(listener); From b56518f258f25800a70596b6f9c1e0de4d240ac7 Mon Sep 17 00:00:00 2001 From: Seva Date: Fri, 17 May 2019 14:03:07 -0400 Subject: [PATCH 07/17] dhtproxy: rewrite listen with restinio --- include/opendht/dht_proxy_server.h | 4 ++- src/dht_proxy_server.cpp | 48 ++++++++++++++++++++++++++++-- 2 files changed, 49 insertions(+), 3 deletions(-) diff --git a/include/opendht/dht_proxy_server.h b/include/opendht/dht_proxy_server.h index 23bc19b5..14ececca 100644 --- a/include/opendht/dht_proxy_server.h +++ b/include/opendht/dht_proxy_server.h @@ -180,7 +180,8 @@ private: * On error: HTTP 503, body: {"err":"xxxx"} * @param session */ - //void listen(const std::shared_ptr& session); + RequestStatus listen(restinio::request_handle_t request, + restinio::router::route_params_t params); /** * Put a value on the DHT @@ -314,6 +315,7 @@ private: std::thread listenThread_; struct SessionToHashToken { InfoHash hash; + restinio::connection_id_t connId; std::future token; }; std::vector currentListeners_; diff --git a/src/dht_proxy_server.cpp b/src/dht_proxy_server.cpp index ceb363ce..55d26d87 100644 --- a/src/dht_proxy_server.cpp +++ b/src/dht_proxy_server.cpp @@ -198,6 +198,7 @@ DhtProxyServer::createRestRouter() router->http_get("/stats", std::bind(&DhtProxyServer::getStats, 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)); + router->http_get("/:hash/listen", std::bind(&DhtProxyServer::listen, this, _1, _2)); #ifdef OPENDHT_PUSH_NOTIFICATIONS router->add_handler(restinio::http_method_t::http_subscribe, "/:hash", std::bind(&DhtProxyServer::subscribe, this, _1, _2)); @@ -305,7 +306,50 @@ DhtProxyServer::get(restinio::request_handle_t request, return restinio::request_handling_status_t::accepted; } -//DhtProxyServer::listen() +RequestStatus +DhtProxyServer::listen(restinio::request_handle_t request, + restinio::router::route_params_t params) +{ + requestNum_++; + dht::InfoHash infoHash(params["hash"].to_string()); + if (!infoHash) + 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(); + try { + SessionToHashToken listener; + listener.hash = infoHash; + listener.connId = request->connection_id(); + listener.token = dht_->listen(infoHash, [this, response] + (const std::vector>& values, bool expired){ + for (const auto& value: values){ + auto jsonVal = value->toJson(); + if (expired) + jsonVal["expired"] = true; + auto output = Json::writeString(jsonBuilder_, jsonVal) + "\n"; + response->append_chunk(output); + response->flush(); + } + return true; + }); + std::lock_guard lock(lockListener_); + currentListeners_.emplace_back(std::move(listener)); + } catch (const std::exception& e){ + 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_PUSH_NOTIFICATIONS @@ -852,7 +896,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 not server_thread.joinable()); + auto cancel = dht_ and (not testSession); //TODO or !open(listener->connId) if (cancel){ dht_->cancelListen(listener->hash, std::move(listener->token)); // Remove listener if unused From 6d5e3d8cd25a657c74f0b053fb57d108c4770d48 Mon Sep 17 00:00:00 2001 From: Seva Date: Fri, 17 May 2019 14:27:18 -0400 Subject: [PATCH 08/17] dhtproxy: cleanup server --- src/dht_proxy_server.cpp | 2 -- 1 file changed, 2 deletions(-) diff --git a/src/dht_proxy_server.cpp b/src/dht_proxy_server.cpp index 55d26d87..21120168 100644 --- a/src/dht_proxy_server.cpp +++ b/src/dht_proxy_server.cpp @@ -57,8 +57,6 @@ DhtProxyServer::DhtProxyServer(std::shared_ptr dht, in_port_t port , if (not dht_) throw std::invalid_argument("A DHT instance must be provided"); - //service_ = std::unique_ptr(new restbed::Service()); - std::cout << "Running DHT proxy server on port " << port << std::endl; if (not pushServer.empty()){ #ifdef OPENDHT_PUSH_NOTIFICATIONS From c7844fc32051ae052a2e45830dc4b9a33ff2f320 Mon Sep 17 00:00:00 2001 From: Seva Date: Wed, 22 May 2019 10:28:16 -0400 Subject: [PATCH 09/17] dhtproxy: implement restinio client --- include/opendht/dht_proxy_client.h | 23 +++++++++ src/dht_proxy_client.cpp | 83 ++++++++++++++++++++++++++++-- 2 files changed, 102 insertions(+), 4 deletions(-) diff --git a/include/opendht/dht_proxy_client.h b/include/opendht/dht_proxy_client.h index f7a3b47c..159ee49f 100644 --- a/include/opendht/dht_proxy_client.h +++ b/include/opendht/dht_proxy_client.h @@ -29,6 +29,29 @@ #include "scheduler.h" #include "proxy.h" +#include +#include +#include +#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); + + void do_request(const std::string& request, const std::string& addr, std::uint16_t port, + http_parser &parser, http_parser_settings &settings); + + 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; } diff --git a/src/dht_proxy_client.cpp b/src/dht_proxy_client.cpp index 5e40e94b..5e25f394 100644 --- a/src/dht_proxy_client.cpp +++ b/src/dht_proxy_client.cpp @@ -23,11 +23,86 @@ #include "op_cache.h" #include "utils.h" -#include -#include +namespace restinio { namespace client { -#include -#include +template +void do_with_socket(LAMBDA && lambda, const std::string & addr, 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}; + restinio::asio_ns::ip::tcp::resolver::query + query{restinio::asio_ns::ip::tcp::v4(), addr, std::to_string(port)}; + restinio::asio_ns::ip::tcp::resolver::iterator iterator = resolver.resolve(query); + + restinio::asio_ns::connect(socket, iterator); + lambda(socket, io_context); + socket.close(); +} + +void +do_request(const std::string & request, const std::string & addr, std::uint16_t port, + http_parser &parser, http_parser_settings &settings){ + do_with_socket([&](auto & socket, auto & io_context){ + // write request + restinio::asio_ns::streambuf b; + std::ostream req_stream(&b); + req_stream << request; + restinio::asio_ns::write(socket, b); + // read response + std::ostringstream sout; + restinio::asio_ns::error_code error; + restinio::asio_ns::streambuf response_stream; + restinio::asio_ns::read_until(socket, response_stream, "\r\n\r\n"); + while(restinio::asio_ns::read(socket, response_stream, + restinio::asio_ns::transfer_at_least(1), error)){ + sout << &response_stream; + auto nparsed = 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("read error: {}", error)}; + }, addr, port); +} + +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 { From b1a35b82a2a7619ea0724a0788fba33cd6bbc4f0 Mon Sep 17 00:00:00 2001 From: Seva Date: Wed, 22 May 2019 11:55:37 -0400 Subject: [PATCH 10/17] dhtproxy: add sendPushNotification to server --- include/opendht/dht_proxy_client.h | 2 +- include/opendht/dht_proxy_server.h | 3 +- src/dht_proxy_client.cpp | 4 +- src/dht_proxy_server.cpp | 67 +++++++++++++++++++++++++++--- 4 files changed, 67 insertions(+), 9 deletions(-) diff --git a/include/opendht/dht_proxy_client.h b/include/opendht/dht_proxy_client.h index 159ee49f..0cbb30d8 100644 --- a/include/opendht/dht_proxy_client.h +++ b/include/opendht/dht_proxy_client.h @@ -44,7 +44,7 @@ namespace restinio { namespace client { void do_with_socket(LAMBDA && lambda, const std::string & addr, std::uint16_t port); void do_request(const std::string& request, const std::string& addr, std::uint16_t port, - http_parser &parser, http_parser_settings &settings); + http_parser_settings &settings); std::string create_http_request(const restinio::http_request_header_t header, const restinio::http_header_fields_t header_fields, diff --git a/include/opendht/dht_proxy_server.h b/include/opendht/dht_proxy_server.h index 14ececca..ddc72174 100644 --- a/include/opendht/dht_proxy_server.h +++ b/include/opendht/dht_proxy_server.h @@ -27,6 +27,7 @@ #include "scheduler.h" #include "sockaddr.h" #include "value.h" +#include "dht_proxy_client.h" #include #include @@ -281,7 +282,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) const; /** * Remove a push listener between a client and a hash diff --git a/src/dht_proxy_client.cpp b/src/dht_proxy_client.cpp index 5e25f394..e1095f6a 100644 --- a/src/dht_proxy_client.cpp +++ b/src/dht_proxy_client.cpp @@ -42,7 +42,7 @@ void do_with_socket(LAMBDA && lambda, const std::string & addr, std::uint16_t po void do_request(const std::string & request, const std::string & addr, std::uint16_t port, - http_parser &parser, http_parser_settings &settings){ + http_parser_settings &settings){ do_with_socket([&](auto & socket, auto & io_context){ // write request restinio::asio_ns::streambuf b; @@ -50,6 +50,8 @@ do_request(const std::string & request, const std::string & addr, std::uint16_t req_stream << request; restinio::asio_ns::write(socket, b); // read response + http_parser parser; + http_parser_init(&parser, HTTP_RESPONSE); std::ostringstream sout; restinio::asio_ns::error_code error; restinio::asio_ns::streambuf response_stream; diff --git a/src/dht_proxy_server.cpp b/src/dht_proxy_server.cpp index 21120168..183b5827 100644 --- a/src/dht_proxy_server.cpp +++ b/src/dht_proxy_server.cpp @@ -383,9 +383,9 @@ DhtProxyServer::subscribe(restinio::request_handle_t request, std::string err; Json::Value root; Json::CharReaderBuilder rbuilder; - auto* char_data = reinterpret_cast(b.data()); + auto* char_data = reinterpret_cast(request->body().data()); auto reader = std::unique_ptr(rbuilder.newCharReader()); - if (!reader->parse(char_data, char_data + b.size(), &root, &err)){ + if (!reader->parse(char_data, char_data + request->body().size(), &root, &err)){ auto response = this->initHttpResponse( request->create_response(restinio::status_bad_request())); response.set_body(this->RESP_MSG_JSON_INCORRECT); @@ -437,7 +437,7 @@ DhtProxyServer::subscribe(restinio::request_handle_t request, response->done(); } schedulerCv_.notify_one(); - return; + return restinio::request_handling_status_t::accepted; } } listeners->second.emplace_back(Listener{}); @@ -517,17 +517,18 @@ DhtProxyServer::unsubscribe(restinio::request_handle_t request, std::string err; Json::Value root; Json::CharReaderBuilder rbuilder; - auto* char_data = reinterpret_cast(b.data()); + auto* char_data = reinterpret_cast(request->body().data()); auto reader = std::unique_ptr(rbuilder.newCharReader()); - if (!reader->parse(char_data, char_data + b.size(), &root, &err)){ + if (!reader->parse(char_data, char_data + request->body().size(), &root, &err)){ auto response = this->initHttpResponse( request->create_response(restinio::status_bad_request())); response.set_body(this->RESP_MSG_JSON_INCORRECT); return response.done(); } auto pushToken = root["key"].asString(); - if (pushToken.empty()) return; + if (pushToken.empty()) + return restinio::request_handling_status_t::rejected; auto clientId = root["client_id"].asString(); cancelPushListen(pushToken, infoHash, clientId); @@ -569,6 +570,60 @@ 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; + + http_parser_settings settings; + http_parser_settings_init(&settings); + settings.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); + + 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"); + + // 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 body = Json::writeString(wbuilder, content); + + auto request = restinio::client::create_http_request( + header, header_fields, restinio::http_connection_header_t::close, body); + + std::string addr = pushServer_.substr(0, pushServer_.find(":")); + uint16_t port = std::atoi(pushServer_.substr(pushServer_.find(":")).c_str()); + + restinio::client::do_request(request, addr, port, settings); +} + #endif //OPENDHT_PUSH_NOTIFICATIONS void From 598e38dbf4b19eb27f27d59e9ccccef9ea6816c1 Mon Sep 17 00:00:00 2001 From: Seva Date: Wed, 22 May 2019 16:21:07 -0400 Subject: [PATCH 11/17] dhtproxy: fix restinio client rx by parts --- src/dht_proxy_client.cpp | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/src/dht_proxy_client.cpp b/src/dht_proxy_client.cpp index e1095f6a..58a988d2 100644 --- a/src/dht_proxy_client.cpp +++ b/src/dht_proxy_client.cpp @@ -52,12 +52,12 @@ do_request(const std::string & request, const std::string & addr, std::uint16_t // read response http_parser parser; http_parser_init(&parser, HTTP_RESPONSE); - std::ostringstream sout; restinio::asio_ns::error_code error; restinio::asio_ns::streambuf response_stream; restinio::asio_ns::read_until(socket, response_stream, "\r\n\r\n"); while(restinio::asio_ns::read(socket, response_stream, restinio::asio_ns::transfer_at_least(1), error)){ + std::ostringstream sout; sout << &response_stream; auto nparsed = http_parser_execute(&parser, &settings, sout.str().c_str(), sout.str().size()); @@ -66,7 +66,6 @@ do_request(const std::string & request, const std::string & addr, std::uint16_t 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("read error: {}", error)}; }, addr, port); From bd994384f009c7cf201abceed550f422414dd75e Mon Sep 17 00:00:00 2001 From: Seva Date: Wed, 22 May 2019 17:23:36 -0400 Subject: [PATCH 12/17] dhtproxy: fix sendPushNotification --- src/dht_proxy_client.cpp | 2 ++ src/dht_proxy_server.cpp | 8 +++----- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/src/dht_proxy_client.cpp b/src/dht_proxy_client.cpp index 58a988d2..c70edd1d 100644 --- a/src/dht_proxy_client.cpp +++ b/src/dht_proxy_client.cpp @@ -47,6 +47,8 @@ do_request(const std::string & request, const std::string & addr, std::uint16_t // write request restinio::asio_ns::streambuf b; std::ostream req_stream(&b); + // TODO log + std::cout << "sending client request:\n" << request.c_str() << std::endl; req_stream << request; restinio::asio_ns::write(socket, b); // read response diff --git a/src/dht_proxy_server.cpp b/src/dht_proxy_server.cpp index 183b5827..253d6e20 100644 --- a/src/dht_proxy_server.cpp +++ b/src/dht_proxy_server.cpp @@ -575,7 +575,6 @@ 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 { @@ -618,10 +617,9 @@ DhtProxyServer::sendPushNotification(const std::string& token, Json::Value&& jso auto request = restinio::client::create_http_request( header, header_fields, restinio::http_connection_header_t::close, body); - std::string addr = pushServer_.substr(0, pushServer_.find(":")); - uint16_t port = std::atoi(pushServer_.substr(pushServer_.find(":")).c_str()); - - restinio::client::do_request(request, addr, port, settings); + // TODO move pushServer ip,port parsing supporting ipv6 to constructor + uint16_t port = std::atoi(pushServer_.c_str()); + restinio::client::do_request(request, "127.0.0.1", port, settings); } #endif //OPENDHT_PUSH_NOTIFICATIONS From ab082543ea9138b618e7d929472047f56d1cb341 Mon Sep 17 00:00:00 2001 From: Seva Date: Thu, 23 May 2019 15:45:59 -0400 Subject: [PATCH 13/17] dhtproxy: add ipv6 to restinio client & get method --- include/opendht/dht_proxy_client.h | 6 +- src/dht_proxy_client.cpp | 167 +++++++++++++++-------------- src/dht_proxy_server.cpp | 9 +- 3 files changed, 94 insertions(+), 88 deletions(-) diff --git a/include/opendht/dht_proxy_client.h b/include/opendht/dht_proxy_client.h index 0cbb30d8..d14671f3 100644 --- a/include/opendht/dht_proxy_client.h +++ b/include/opendht/dht_proxy_client.h @@ -44,7 +44,7 @@ namespace restinio { namespace client { void do_with_socket(LAMBDA && lambda, const std::string & addr, std::uint16_t port); void do_request(const std::string& request, const std::string& addr, std::uint16_t port, - http_parser_settings &settings); + http_parser &parser, http_parser_settings &settings); std::string create_http_request(const restinio::http_request_header_t header, const restinio::http_header_fields_t header_fields, @@ -69,6 +69,8 @@ public: explicit DhtProxyClient(std::function loopSignal, const std::string& serverHost, const std::string& pushClientId = ""); + restinio::http_header_fields_t initHeaderFields(); + virtual void setPushNotificationToken(const std::string& token) { #ifdef OPENDHT_PUSH_NOTIFICATIONS deviceKey_ = token; @@ -326,6 +328,8 @@ private: void cancelAllOperations(); std::string serverHost_; + std::string serverHostIp_; + uint16_t serverHostPort_; std::string pushClientId_; mutable std::mutex lockCurrentProxyInfos_; diff --git a/src/dht_proxy_client.cpp b/src/dht_proxy_client.cpp index c70edd1d..4c9effad 100644 --- a/src/dht_proxy_client.cpp +++ b/src/dht_proxy_client.cpp @@ -26,13 +26,14 @@ namespace restinio { namespace client { template -void do_with_socket(LAMBDA && lambda, const std::string & addr, std::uint16_t port){ +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}; - restinio::asio_ns::ip::tcp::resolver::query - query{restinio::asio_ns::ip::tcp::v4(), addr, std::to_string(port)}; + + 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); @@ -41,18 +42,17 @@ void do_with_socket(LAMBDA && lambda, const std::string & addr, std::uint16_t po } void -do_request(const std::string & request, const std::string & addr, std::uint16_t port, - http_parser_settings &settings){ +do_request(const std::string & request, const std::string &addr, std::uint16_t port, + http_parser &parser, http_parser_settings &settings){ do_with_socket([&](auto & socket, auto & io_context){ // write request restinio::asio_ns::streambuf b; std::ostream req_stream(&b); - // TODO log - std::cout << "sending client request:\n" << request.c_str() << std::endl; + // TODO DHT_LOG.w + std::cout << "Sending client request:\n" << request.c_str() << std::endl; req_stream << request; restinio::asio_ns::write(socket, b); // read response - http_parser parser; http_parser_init(&parser, HTTP_RESPONSE); restinio::asio_ns::error_code error; restinio::asio_ns::streambuf response_stream; @@ -157,6 +157,10 @@ DhtProxyClient::DhtProxyClient(std::function signal, const std::string& serverHost_ = proxy::HTTP_PROTO + serverHost_; if (!serverHost_.empty()) startProxy(); + + auto hostAndPort = splitPort(serverHost_); + this->serverHostIp_ = hostAndPort.first; + this->serverHostPort_ = std::atoi(hostAndPort.second.c_str()); } void @@ -337,88 +341,85 @@ DhtProxyClient::periodic(const uint8_t*, size_t, const SockAddr&) return scheduler.run(); } +restinio::http_header_fields_t +DhtProxyClient::initHeaderFields(){ + restinio::http_header_fields_t header_fields; + header_fields.append_field(restinio::http_field_t::host, (serverHost_).c_str()); + header_fields.append_field(restinio::http_field_t::user_agent, "RESTinio client"); + header_fields.append_field(restinio::http_field_t::accept, "*/*"); + return header_fields; +} + void DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Value::Filter&& f, Where&& w) { DHT_LOG.d(key, "[search %s]: get", key.to_c_str()); - restbed::Uri uri(serverHost_ + "/" + key.toString()); - auto req = std::make_shared(uri); - Value::Filter filter = w.empty() ? f : f.chain(w.getFilter()); + //restbed::Uri uri(serverHost_ + "/" + key.toString()); + restinio::http_request_header_t header; + 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, + restinio::http_connection_header_t::keep_alive, ""/*body*/); + printf(request.c_str()); - auto finished = std::make_shared(false); - Operation o; - o.req = req; - o.finished = finished; - o.thread = std::thread([=](){ - // Try to contact the proxy and set the status to connected when done. - // will change the connectivity status - struct GetState{ std::atomic_bool ok {true}; std::atomic_bool stop {false}; }; - auto state = std::make_shared(); - try { - restbed::Http::async(req, - [=](const std::shared_ptr& req, - const std::shared_ptr& reply) { - auto code = reply->get_status_code(); + //auto req = std::make_shared(uri); + struct GetContext { + std::atomic_bool ok {true}; + std::atomic_bool stop {false}; + Value::Filter filter; + GetCallback cb; + DoneCallback donecb; + }; + auto context = std::make_shared(); + context->filter = w.empty() ? f : f.chain(w.getFilter()); + context->cb = cb; context->donecb = donecb; - if (code == 200) { - try { - while (restbed::Http::is_open(req) and not *finished and not state->stop) { - restbed::Http::fetch("\n", reply); - if (*finished or state->stop) - break; - std::string body; - reply->get_body(body); - reply->set_body(""); // Reset the body for the next fetch - - std::string err; - Json::Value json; - Json::CharReaderBuilder rbuilder; - auto* char_data = reinterpret_cast(&body[0]); - auto reader = std::unique_ptr(rbuilder.newCharReader()); - if (reader->parse(char_data, char_data + body.size(), &json, &err)) { - auto value = std::make_shared(json); - if ((not filter or filter(*value)) and cb) { - { - std::lock_guard lock(lockCallbacks); - callbacks_.emplace_back([cb, value, state]() { - if (not state->stop and not cb({value})) - state->stop = true; - }); - } - loopSignal_(); - } - } else { - state->ok = false; - } - } - } catch (std::runtime_error& e) { } - } else { - state->ok = false; - } - }).wait(); - } catch(const std::exception& e) { - state->ok = false; - } - if (donecb) { - { - std::lock_guard lock(lockCallbacks); - callbacks_.emplace_back([=](){ - donecb(state->ok, {}); - state->stop = true; - }); + Operation op; + //o.req = req; + op.finished = std::make_shared(false); + op.thread = std::thread([this, request, &context](){ + http_parser parser; + parser.data = reinterpret_cast(&context); + http_parser_settings settings; + http_parser_settings_init(&settings); + settings.on_status = [](http_parser *parser, const char *at, size_t length) -> int { + auto context = reinterpret_cast(parser->data); + if (parser->status_code != 200) + context->ok = false; + return 0; + }; + settings.on_body = [](http_parser *parser, const char *at, size_t length) -> int { + std::string body, err; + Json::Value json; + Json::CharReaderBuilder rbuilder; + auto context = reinterpret_cast(parser->data); + try { + body = std::string(at, length).c_str(); + } catch(const std::exception& e) { + context->ok = false; + return 1; } - loopSignal_(); - } - if (!state->ok) { - // Connection failed, update connectivity - opFailed(); - } - *finished = true; + auto* char_data = reinterpret_cast(&body); + auto reader = std::unique_ptr(rbuilder.newCharReader()); + if (!reader->parse(char_data, char_data + body.size(), &json, &err)){ + context->ok = false; + return 1; + } + auto value = std::make_shared(json); + if ((not context->filter or context->filter(*value)) and context->cb){ + context->cb({value}); + } + return 0; + }; + settings.on_message_complete = [](http_parser * parser) -> int { + auto context = reinterpret_cast(parser->data); + if (context->donecb) + context->donecb(context->ok, {}); + return 0; + }; + restinio::client::do_request(request, serverHostIp_, serverHostPort_, parser, settings); }); - { - std::lock_guard lock(lockOperations_); - operations_.emplace_back(std::move(o)); - } } void diff --git a/src/dht_proxy_server.cpp b/src/dht_proxy_server.cpp index 253d6e20..76054211 100644 --- a/src/dht_proxy_server.cpp +++ b/src/dht_proxy_server.cpp @@ -74,7 +74,7 @@ DhtProxyServer::DhtProxyServer(std::shared_ptr dht, in_port_t port , auto maxThreads = std::thread::hardware_concurrency() - 1; auto restThreads = maxThreads > 1 ? maxThreads : 1; auto settings = restinio::on_thread_pool(restThreads); - settings.address("0.0.0.0"); + settings.address("127.0.0.1"); // TODO ipv6 listener? settings.port(port); settings.request_handler(this->createRestRouter()); // time limits @@ -617,9 +617,10 @@ DhtProxyServer::sendPushNotification(const std::string& token, Json::Value&& jso auto request = restinio::client::create_http_request( header, header_fields, restinio::http_connection_header_t::close, body); - // TODO move pushServer ip,port parsing supporting ipv6 to constructor - uint16_t port = std::atoi(pushServer_.c_str()); - restinio::client::do_request(request, "127.0.0.1", port, settings); + http_parser parser; + auto hostAndPort = splitPort(pushServer_); + uint16_t port = std::atoi(hostAndPort.second.c_str()); + restinio::client::do_request(request, hostAndPort.first, port, parser, settings); } #endif //OPENDHT_PUSH_NOTIFICATIONS From 52c413f05650dd58c80f52fdb29a0b1c1b14e677 Mon Sep 17 00:00:00 2001 From: Seva Date: Fri, 24 May 2019 13:18:06 -0400 Subject: [PATCH 14/17] dhtproxy: switch server to ipv6 (implicit ipv4 support) --- src/dht_proxy_server.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/dht_proxy_server.cpp b/src/dht_proxy_server.cpp index 76054211..d5be321c 100644 --- a/src/dht_proxy_server.cpp +++ b/src/dht_proxy_server.cpp @@ -74,8 +74,8 @@ DhtProxyServer::DhtProxyServer(std::shared_ptr dht, in_port_t port , auto maxThreads = std::thread::hardware_concurrency() - 1; auto restThreads = maxThreads > 1 ? maxThreads : 1; auto settings = restinio::on_thread_pool(restThreads); - settings.address("127.0.0.1"); // TODO ipv6 listener? settings.port(port); + settings.protocol(restinio::asio_ns::ip::tcp::v6()); settings.request_handler(this->createRestRouter()); // time limits settings.read_next_http_message_timelimit(10s); From 2545282624c5216386355dd1278d218a58753bcc Mon Sep 17 00:00:00 2001 From: Seva Date: Sat, 25 May 2019 17:46:35 -0400 Subject: [PATCH 15/17] dhtproxy: fix context type casting in client get --- include/opendht/dht_proxy_client.h | 2 +- src/dht_proxy_client.cpp | 68 ++++++++++++++++++------------ 2 files changed, 42 insertions(+), 28 deletions(-) diff --git a/include/opendht/dht_proxy_client.h b/include/opendht/dht_proxy_client.h index d14671f3..e9fb1a53 100644 --- a/include/opendht/dht_proxy_client.h +++ b/include/opendht/dht_proxy_client.h @@ -360,8 +360,8 @@ private: */ struct Operation { - std::shared_ptr req; std::thread thread; + std::shared_ptr req; std::shared_ptr finished; }; std::vector operations_; diff --git a/src/dht_proxy_client.cpp b/src/dht_proxy_client.cpp index 4c9effad..6c21574f 100644 --- a/src/dht_proxy_client.cpp +++ b/src/dht_proxy_client.cpp @@ -48,8 +48,6 @@ do_request(const std::string & request, const std::string &addr, std::uint16_t p // write request restinio::asio_ns::streambuf b; std::ostream req_stream(&b); - // TODO DHT_LOG.w - std::cout << "Sending client request:\n" << request.c_str() << std::endl; req_stream << request; restinio::asio_ns::write(socket, b); // read response @@ -61,6 +59,7 @@ do_request(const std::string & request, const std::string &addr, std::uint16_t p restinio::asio_ns::transfer_at_least(1), error)){ std::ostringstream sout; sout << &response_stream; + //std::cout << "{" << sout.str().c_str() << "}" << std::endl; auto nparsed = http_parser_execute(&parser, &settings, sout.str().c_str(), sout.str().size()); if (HPE_OK != parser.http_errno && HPE_PAUSED != parser.http_errno){ @@ -153,14 +152,14 @@ DhtProxyClient::DhtProxyClient() {} DhtProxyClient::DhtProxyClient(std::function signal, const std::string& serverHost, const std::string& pushClientId) : serverHost_(serverHost), pushClientId_(pushClientId), loopSignal_(signal) { + auto hostAndPort = splitPort(serverHost_); + this->serverHostIp_ = hostAndPort.first; + this->serverHostPort_ = std::atoi(hostAndPort.second.c_str()); + if (serverHost_.find("://") == std::string::npos) serverHost_ = proxy::HTTP_PROTO + serverHost_; if (!serverHost_.empty()) startProxy(); - - auto hostAndPort = splitPort(serverHost_); - this->serverHostIp_ = hostAndPort.first; - this->serverHostPort_ = std::atoi(hostAndPort.second.c_str()); } void @@ -212,6 +211,7 @@ DhtProxyClient::getLocalById(const InfoHash& k, Value::Id id) const { void DhtProxyClient::cancelAllOperations() { + /* std::lock_guard lock(lockOperations_); auto operation = operations_.begin(); while (operation != operations_.end()) { @@ -231,6 +231,7 @@ DhtProxyClient::cancelAllOperations() ++operation; } } + */ } void @@ -316,6 +317,7 @@ DhtProxyClient::periodic(const uint8_t*, size_t, const SockAddr&) // Remove finished operations { + /* std::lock_guard lock(lockOperations_); auto operation = operations_.begin(); while (operation != operations_.end()) { @@ -337,6 +339,7 @@ DhtProxyClient::periodic(const uint8_t*, size_t, const SockAddr&) ++operation; } } + */ } return scheduler.run(); } @@ -376,50 +379,61 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Va context->cb = cb; context->donecb = donecb; Operation op; - //o.req = req; + //o.req = req; // TODO save ptr to close op.finished = std::make_shared(false); - op.thread = std::thread([this, request, &context](){ + op.thread = std::thread([this, request, context](){ http_parser parser; - parser.data = reinterpret_cast(&context); + parser.data = static_cast(context.get()); http_parser_settings settings; http_parser_settings_init(&settings); settings.on_status = [](http_parser *parser, const char *at, size_t length) -> int { - auto context = reinterpret_cast(parser->data); + GetContext* context = reinterpret_cast(parser->data); if (parser->status_code != 200) context->ok = false; return 0; }; settings.on_body = [](http_parser *parser, const char *at, size_t length) -> int { - std::string body, err; - Json::Value json; - Json::CharReaderBuilder rbuilder; auto context = reinterpret_cast(parser->data); - try { - body = std::string(at, length).c_str(); + try{ + Json::Value json; + std::string err; + Json::CharReaderBuilder rbuilder; + auto body = std::string(at, length); + auto* char_data = reinterpret_cast(&body[0]); + auto reader = std::unique_ptr(rbuilder.newCharReader()); + if (!reader->parse(char_data, char_data + body.size(), &json, &err)){ + context->ok = false; + return 1; + } + auto value = std::make_shared(json); + if ((not context->filter or context->filter(*value)) and context->cb){ + context->cb({value}); + } } catch(const std::exception& e) { + std::cerr << "Error in get parsing: " << e.what() << std::endl; context->ok = false; return 1; } - auto* char_data = reinterpret_cast(&body); - auto reader = std::unique_ptr(rbuilder.newCharReader()); - if (!reader->parse(char_data, char_data + body.size(), &json, &err)){ - context->ok = false; - return 1; - } - auto value = std::make_shared(json); - if ((not context->filter or context->filter(*value)) and context->cb){ - context->cb({value}); - } return 0; }; settings.on_message_complete = [](http_parser * parser) -> int { auto context = reinterpret_cast(parser->data); - if (context->donecb) - context->donecb(context->ok, {}); + try { + if (context->donecb) + context->donecb(context->ok, {}); + } catch(const std::exception& e) { + std::cerr << "Error in get parsing: " << e.what() << std::endl; + context->ok = false; + return 1; + } return 0; }; restinio::client::do_request(request, serverHostIp_, serverHostPort_, parser, settings); }); + { + std::lock_guard lock(lockOperations_); + operations_.emplace_back(std::move(op)); + } } void From f3e304e6612517570c7502f79226fb5b0fa56e6f Mon Sep 17 00:00:00 2001 From: Seva Date: Mon, 27 May 2019 10:23:16 -0400 Subject: [PATCH 16/17] dhtproxy: switch threading to io_context --- include/opendht/dht_proxy_client.h | 8 +- src/dht_proxy_client.cpp | 122 ++++++++++++++--------------- src/dht_proxy_server.cpp | 3 + 3 files changed, 67 insertions(+), 66 deletions(-) diff --git a/include/opendht/dht_proxy_client.h b/include/opendht/dht_proxy_client.h index e9fb1a53..00e63062 100644 --- a/include/opendht/dht_proxy_client.h +++ b/include/opendht/dht_proxy_client.h @@ -41,10 +41,12 @@ namespace restinio { namespace client { template - void do_with_socket(LAMBDA && lambda, const std::string & addr, std::uint16_t port); + 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); + http_parser &parser, http_parser_settings &settings, + std::shared_ptr io_context = nullptr); std::string create_http_request(const restinio::http_request_header_t header, const restinio::http_header_fields_t header_fields, @@ -376,6 +378,8 @@ private: std::thread statusThread_; mutable std::mutex statusLock_; + std::shared_ptr io_context; + Scheduler scheduler; /** * Retrieve if we can connect to the proxy (update statusIpvX_) diff --git a/src/dht_proxy_client.cpp b/src/dht_proxy_client.cpp index 6c21574f..86c21656 100644 --- a/src/dht_proxy_client.cpp +++ b/src/dht_proxy_client.cpp @@ -26,8 +26,8 @@ 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; +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}; @@ -37,14 +37,17 @@ void do_with_socket(LAMBDA && lambda, const std::string &ip, std::uint16_t port) restinio::asio_ns::ip::tcp::resolver::iterator iterator = resolver.resolve(query); restinio::asio_ns::connect(socket, iterator); - lambda(socket, io_context); + 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){ - do_with_socket([&](auto & socket, auto & io_context){ + http_parser &parser, http_parser_settings &settings, + std::shared_ptr io_context){ + if (!io_context) + io_context = std::make_shared(); + do_with_socket([&](auto & socket){ // write request restinio::asio_ns::streambuf b; std::ostream req_stream(&b); @@ -59,9 +62,8 @@ do_request(const std::string & request, const std::string &addr, std::uint16_t p restinio::asio_ns::transfer_at_least(1), error)){ std::ostringstream sout; sout << &response_stream; - //std::cout << "{" << sout.str().c_str() << "}" << std::endl; - auto nparsed = http_parser_execute(&parser, &settings, - sout.str().c_str(), sout.str().size()); + //std::cout << "{" << sout.str().c_str() << "}" << std::endl;// 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; @@ -69,7 +71,7 @@ do_request(const std::string & request, const std::string &addr, std::uint16_t p } if (!restinio::error_is_eof(error)) throw std::runtime_error{fmt::format("read error: {}", error)}; - }, addr, port); + }, addr, port, *io_context); } std::string @@ -150,7 +152,7 @@ struct DhtProxyClient::ProxySearch { DhtProxyClient::DhtProxyClient() {} DhtProxyClient::DhtProxyClient(std::function signal, const std::string& serverHost, const std::string& pushClientId) -: serverHost_(serverHost), pushClientId_(pushClientId), loopSignal_(signal) +: serverHost_(serverHost), pushClientId_(pushClientId), loopSignal_(signal), io_context(new restinio::asio_ns::io_context()) { auto hostAndPort = splitPort(serverHost_); this->serverHostIp_ = hostAndPort.first; @@ -265,6 +267,8 @@ DhtProxyClient::cancelAllListeners() void DhtProxyClient::shutdown(ShutdownCallback cb) { + this->io_context->stop(); // call reset() before to allow gracefull finish + cancelAllOperations(); cancelAllListeners(); if (cb) @@ -357,7 +361,6 @@ void DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Value::Filter&& f, Where&& w) { DHT_LOG.d(key, "[search %s]: get", key.to_c_str()); - //restbed::Uri uri(serverHost_ + "/" + key.toString()); restinio::http_request_header_t header; header.request_target("/" + key.toString()); header.method(restinio::http_method_t::http_get); @@ -366,7 +369,6 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Va restinio::http_connection_header_t::keep_alive, ""/*body*/); printf(request.c_str()); - //auto req = std::make_shared(uri); struct GetContext { std::atomic_bool ok {true}; std::atomic_bool stop {false}; @@ -378,62 +380,54 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Va context->filter = w.empty() ? f : f.chain(w.getFilter()); context->cb = cb; context->donecb = donecb; - Operation op; - //o.req = req; // TODO save ptr to close - op.finished = std::make_shared(false); - op.thread = std::thread([this, request, context](){ - http_parser parser; - parser.data = static_cast(context.get()); - http_parser_settings settings; - http_parser_settings_init(&settings); - settings.on_status = [](http_parser *parser, const char *at, size_t length) -> int { - GetContext* context = reinterpret_cast(parser->data); - if (parser->status_code != 200) - context->ok = false; - return 0; - }; - settings.on_body = [](http_parser *parser, const char *at, size_t length) -> int { - auto context = reinterpret_cast(parser->data); - try{ - Json::Value json; - std::string err; - Json::CharReaderBuilder rbuilder; - auto body = std::string(at, length); - auto* char_data = reinterpret_cast(&body[0]); - auto reader = std::unique_ptr(rbuilder.newCharReader()); - if (!reader->parse(char_data, char_data + body.size(), &json, &err)){ - context->ok = false; - return 1; - } - auto value = std::make_shared(json); - if ((not context->filter or context->filter(*value)) and context->cb){ - context->cb({value}); - } - } catch(const std::exception& e) { - std::cerr << "Error in get parsing: " << e.what() << std::endl; + http_parser parser; + parser.data = static_cast(context.get()); + http_parser_settings settings; + http_parser_settings_init(&settings); + settings.on_status = [](http_parser *parser, const char *at, size_t length) -> int { + GetContext* context = reinterpret_cast(parser->data); + if (parser->status_code != 200) + context->ok = false; + return 0; + }; + settings.on_body = [](http_parser *parser, const char *at, size_t length) -> int { + auto context = reinterpret_cast(parser->data); + try{ + Json::Value json; + std::string err; + Json::CharReaderBuilder rbuilder; + auto body = std::string(at, length); + auto* char_data = reinterpret_cast(&body[0]); + auto reader = std::unique_ptr(rbuilder.newCharReader()); + if (!reader->parse(char_data, char_data + body.size(), &json, &err)){ context->ok = false; return 1; } - return 0; - }; - settings.on_message_complete = [](http_parser * parser) -> int { - auto context = reinterpret_cast(parser->data); - try { - if (context->donecb) - context->donecb(context->ok, {}); - } catch(const std::exception& e) { - std::cerr << "Error in get parsing: " << e.what() << std::endl; - context->ok = false; - return 1; + auto value = std::make_shared(json); + if ((not context->filter or context->filter(*value)) and context->cb){ + context->cb({value}); } - return 0; - }; - restinio::client::do_request(request, serverHostIp_, serverHostPort_, parser, settings); - }); - { - std::lock_guard lock(lockOperations_); - operations_.emplace_back(std::move(op)); - } + } catch(const std::exception& e) { + std::cerr << "Error in get parsing: " << e.what() << std::endl; + context->ok = false; + return 1; + } + return 0; + }; + settings.on_message_complete = [](http_parser * parser) -> int { + auto context = reinterpret_cast(parser->data); + try { + if (context->donecb) + context->donecb(context->ok, {}); + } catch(const std::exception& e) { + std::cerr << "Error in get parsing: " << e.what() << std::endl; + context->ok = false; + return 1; + } + return 0; + }; + restinio::client::do_request(request, serverHostIp_, serverHostPort_, + parser, settings, this->io_context); } void diff --git a/src/dht_proxy_server.cpp b/src/dht_proxy_server.cpp index d5be321c..7981a3f7 100644 --- a/src/dht_proxy_server.cpp +++ b/src/dht_proxy_server.cpp @@ -620,6 +620,9 @@ DhtProxyServer::sendPushNotification(const std::string& token, Json::Value&& jso 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); } From 1c3132e0efa2a53ea8be86a5279455e148c3d2b4 Mon Sep 17 00:00:00 2001 From: Seva Date: Mon, 27 May 2019 14:48:36 -0400 Subject: [PATCH 17/17] dhtproxy: add async put & swap threading for io_context --- include/opendht/dht_proxy_client.h | 16 +-- src/dht_proxy_client.cpp | 181 ++++++++++------------------- 2 files changed, 65 insertions(+), 132 deletions(-) diff --git a/include/opendht/dht_proxy_client.h b/include/opendht/dht_proxy_client.h index 00e63062..170a3d0e 100644 --- a/include/opendht/dht_proxy_client.h +++ b/include/opendht/dht_proxy_client.h @@ -357,17 +357,6 @@ private: std::map searches_; mutable std::mutex searchLock_; - /** - * Store current put and get requests. - */ - struct Operation - { - std::thread thread; - std::shared_ptr req; - std::shared_ptr finished; - }; - std::vector operations_; - std::mutex lockOperations_; /** * Callbacks should be executed in the main thread. */ @@ -378,7 +367,12 @@ 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_; Scheduler scheduler; /** diff --git a/src/dht_proxy_client.cpp b/src/dht_proxy_client.cpp index 86c21656..df6c308f 100644 --- a/src/dht_proxy_client.cpp +++ b/src/dht_proxy_client.cpp @@ -18,7 +18,6 @@ */ #include "dht_proxy_client.h" - #include "dhtrunner.h" #include "op_cache.h" #include "utils.h" @@ -45,8 +44,10 @@ 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){ - if (!io_context) + 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){ // write request restinio::asio_ns::streambuf b; @@ -57,12 +58,11 @@ do_request(const std::string & request, const std::string &addr, std::uint16_t p http_parser_init(&parser, HTTP_RESPONSE); restinio::asio_ns::error_code error; restinio::asio_ns::streambuf response_stream; - restinio::asio_ns::read_until(socket, response_stream, "\r\n\r\n"); while(restinio::asio_ns::read(socket, response_stream, restinio::asio_ns::transfer_at_least(1), error)){ std::ostringstream sout; sout << &response_stream; - //std::cout << "{" << sout.str().c_str() << "}" << std::endl;// DHT_LOG.w ? + printf("Got server response:%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); @@ -70,7 +70,7 @@ do_request(const std::string & request, const std::string &addr, std::uint16_t p } } if (!restinio::error_is_eof(error)) - throw std::runtime_error{fmt::format("read error: {}", error)}; + throw std::runtime_error{fmt::format("Client read error: {}", error)}; }, addr, port, *io_context); } @@ -162,6 +162,8 @@ 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();}); } void @@ -190,6 +192,8 @@ DhtProxyClient::~DhtProxyClient() infoState_->cancel = true; if (statusThread_.joinable()) statusThread_.join(); + if (io_context_thread_.joinable()) + io_context_thread_.join(); } std::vector> @@ -213,27 +217,8 @@ DhtProxyClient::getLocalById(const InfoHash& k, Value::Id id) const { void DhtProxyClient::cancelAllOperations() { - /* - std::lock_guard lock(lockOperations_); - auto operation = operations_.begin(); - while (operation != operations_.end()) { - if (operation->thread.joinable()) { - // Close connection to stop operation? - if (operation->req) { - try { - restbed::Http::close(operation->req); - } catch (const std::exception& e) { - DHT_LOG.w("Error closing socket: %s", e.what()); - } - operation->req.reset(); - } - operation->thread.join(); - operation = operations_.erase(operation); - } else { - ++operation; - } - } - */ + // for graceful finish, call reset() before + this->io_context->stop(); } void @@ -267,8 +252,6 @@ DhtProxyClient::cancelAllListeners() void DhtProxyClient::shutdown(ShutdownCallback cb) { - this->io_context->stop(); // call reset() before to allow gracefull finish - cancelAllOperations(); cancelAllListeners(); if (cb) @@ -318,42 +301,17 @@ DhtProxyClient::periodic(const uint8_t*, size_t, const SockAddr&) for (auto& callback : callbacks) callback(); callbacks.clear(); - - // Remove finished operations - { - /* - std::lock_guard lock(lockOperations_); - auto operation = operations_.begin(); - while (operation != operations_.end()) { - if (*(operation->finished)) { - if (operation->thread.joinable()) { - // Close connection to stop operation? - if (operation->req) { - try { - restbed::Http::close(operation->req); - } catch (const std::exception& e) { - DHT_LOG.w("Error closing socket: %s", e.what()); - } - operation->req.reset(); - } - operation->thread.join(); - } - operation = operations_.erase(operation); - } else { - ++operation; - } - } - */ - } return scheduler.run(); } restinio::http_header_fields_t DhtProxyClient::initHeaderFields(){ restinio::http_header_fields_t header_fields; - header_fields.append_field(restinio::http_field_t::host, (serverHost_).c_str()); + header_fields.append_field(restinio::http_field_t::host, + (serverHostIp_ + ":" + std::to_string(serverHostPort_)).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; } @@ -367,14 +325,13 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Va auto header_fields = this->initHeaderFields(); auto request = restinio::client::create_http_request(header, header_fields, restinio::http_connection_header_t::keep_alive, ""/*body*/); - printf(request.c_str()); + DHT_LOG.w(request.c_str()); struct GetContext { - std::atomic_bool ok {true}; - std::atomic_bool stop {false}; - Value::Filter filter; GetCallback cb; DoneCallback donecb; + Value::Filter filter; + std::atomic_bool ok {false}; }; auto context = std::make_shared(); context->filter = w.empty() ? f : f.chain(w.getFilter()); @@ -386,8 +343,10 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Va http_parser_settings_init(&settings); settings.on_status = [](http_parser *parser, const char *at, size_t length) -> int { GetContext* context = reinterpret_cast(parser->data); - if (parser->status_code != 200) - context->ok = false; + if (parser->status_code != 200){ + std::cerr << "Error in get status_code=" << parser->status_code << std::endl; + context->ok = true; + } return 0; }; settings.on_body = [](http_parser *parser, const char *at, size_t length) -> int { @@ -421,7 +380,6 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Va context->donecb(context->ok, {}); } catch(const std::exception& e) { std::cerr << "Error in get parsing: " << e.what() << std::endl; - context->ok = false; return 1; } return 0; @@ -475,9 +433,10 @@ void DhtProxyClient::doPut(const InfoHash& key, Sp val, DoneCallback cb, time_point /*created*/, bool permanent) { DHT_LOG.d(key, "[search %s] performing put of %s", key.to_c_str(), val->toString().c_str()); - restbed::Uri uri(serverHost_ + "/" + key.toString()); - auto req = std::make_shared(uri); - req->set_method("POST"); + restinio::http_request_header_t header; + header.request_target("/" + key.toString()); + header.method(restinio::http_method_t::http_post); + auto header_fields = this->initHeaderFields(); auto json = val->toJson(); if (permanent) { @@ -496,65 +455,45 @@ DhtProxyClient::doPut(const InfoHash& key, Sp val, DoneCallback cb, time_ Json::StreamWriterBuilder wbuilder; wbuilder["commentStyle"] = "None"; wbuilder["indentation"] = ""; - auto body = Json::writeString(wbuilder, json) + "\n"; - req->set_body(body); - req->set_header("Content-Length", std::to_string(body.size())); + auto body = Json::writeString(wbuilder, json); + auto request = restinio::client::create_http_request(header, header_fields, + restinio::http_connection_header_t::close, body); + // TODO dhtlog.w + printf("%s\n", request.c_str()); - auto finished = std::make_shared(false); - Operation o; - o.req = req; - o.finished = finished; - o.thread = std::thread([=](){ - auto ok = std::make_shared(true); + struct GetContext { + DoneCallback donecb; + std::atomic_bool ok {false}; + }; + auto context = std::make_shared(); + context->donecb = cb; + + http_parser parser; + parser.data = static_cast(context.get()); + http_parser_settings settings; + http_parser_settings_init(&settings); + settings.on_status = [](http_parser *parser, const char *at, size_t length) -> int { + GetContext* context = reinterpret_cast(parser->data); + if (parser->status_code == 200){ + context->ok = true; + } else { + std::cerr << "Error in get status_code=" << parser->status_code << std::endl; + } + return 0; + }; + settings.on_message_complete = [](http_parser * parser) -> int { + auto context = reinterpret_cast(parser->data); try { - restbed::Http::async(req, - [ok](const std::shared_ptr& /*req*/, - const std::shared_ptr& reply) { - auto code = reply->get_status_code(); - - if (code == 200) { - restbed::Http::fetch("\n", reply); - std::string body; - reply->get_body(body); - reply->set_body(""); // Reset the body for the next fetch - - try { - std::string err; - Json::Value json; - Json::CharReaderBuilder rbuilder; - auto* char_data = reinterpret_cast(&body[0]); - auto reader = std::unique_ptr(rbuilder.newCharReader()); - if (not reader->parse(char_data, char_data + body.size(), &json, &err)) - *ok = false; - } catch (...) { - *ok = false; - } - } else { - *ok = false; - } - }).wait(); + if (context->donecb) + context->donecb(context->ok, {}); } catch(const std::exception& e) { - *ok = false; + std::cerr << "Error in get parsing: " << e.what() << std::endl; + return 1; } - if (cb) { - { - std::lock_guard lock(lockCallbacks); - callbacks_.emplace_back([=](){ - cb(*ok, {}); - }); - } - loopSignal_(); - } - if (!ok) { - // Connection failed, update connectivity - opFailed(); - } - *finished = true; - }); - { - std::lock_guard lock(lockOperations_); - operations_.emplace_back(std::move(o)); - } + return 0; + }; + restinio::client::do_request(request, serverHostIp_, serverHostPort_, + parser, settings, this->io_context); } /**