diff --git a/include/opendht/dht_proxy_client.h b/include/opendht/dht_proxy_client.h index fad59729..416318b1 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 fa2db648..351db038 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 @@ -149,8 +151,13 @@ struct DhtProxyClient::ProxySearch { DhtProxyClient::DhtProxyClient() {} +<<<<<<< HEAD DhtProxyClient::DhtProxyClient(std::function signal, const std::string& serverHost, const std::string& pushClientId, const Logger& l) : DhtInterface(l), serverHost_(serverHost), pushClientId_(pushClientId), loopSignal_(signal) +======= +DhtProxyClient::DhtProxyClient(std::function signal, const std::string& serverHost, const std::string& pushClientId) +: serverHost_(serverHost), pushClientId_(pushClientId), loopSignal_(signal), io_context(new restinio::asio_ns::io_context()) +>>>>>>> dhtproxy: switch threading to io_context { auto hostAndPort = splitPort(serverHost_); this->serverHostIp_ = hostAndPort.first; @@ -265,6 +272,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 +366,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 +374,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 +385,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); }