diff --git a/src/dht_proxy_server.cpp b/src/dht_proxy_server.cpp index 0a04e000..0fce0812 100644 --- a/src/dht_proxy_server.cpp +++ b/src/dht_proxy_server.cpp @@ -69,48 +69,60 @@ struct ListenerSession std::future token; std::shared_ptr> response; }; -} +} // namespace http class opendht_logger_t { public: - opendht_logger_t(std::shared_ptr logger = {}) : m_logger(std::move(logger)) {} + opendht_logger_t(std::shared_ptr logger = {}) + : m_logger(std::move(logger)) + {} - template - void trace(Builder&& /* msg_builder */) { + template + void trace(Builder&& /* msg_builder */) + { /* if (m_logger) m_logger->d("[proxy:server] %s", msg_builder().c_str()); */ } - template - void info(Builder&& msg_builder) { - if (m_logger) m_logger->d("[proxy:server] %s", msg_builder().c_str()); + template + void info(Builder&& msg_builder) + { + if (m_logger) + m_logger->d("[proxy:server] %s", msg_builder().c_str()); } - template - void warn(Builder&& msg_builder) { - if (m_logger) m_logger->w("[proxy:server] %s", msg_builder().c_str()); + template + void warn(Builder&& msg_builder) + { + if (m_logger) + m_logger->w("[proxy:server] %s", msg_builder().c_str()); } - template - void error(Builder&& msg_builder) { - if (m_logger) m_logger->e("[proxy:server] %s", msg_builder().c_str()); + template + void error(Builder&& msg_builder) + { + if (m_logger) + m_logger->e("[proxy:server] %s", msg_builder().c_str()); } private: std::shared_ptr m_logger; }; - std::string DhtProxyServer::ServerStats::toString() const { auto ret = fmt::format("Listens: {}, Puts: {}, PushListeners: {}\n" - "Push requests in the last {}: [Android: {}], [iOS: {}], [Unified: {}]\n" - "Requests: {} per second.", - listenCount, putCount, pushListenersCount, - print_duration(lastUpdated - serverStartTime), - androidPush.toString(), iosPush.toString(), unifiedPush.toString(), - requestRate); + "Push requests in the last {}: [Android: {}], [iOS: {}], [Unified: {}]\n" + "Requests: {} per second.", + listenCount, + putCount, + pushListenersCount, + print_duration(lastUpdated - serverStartTime), + androidPush.toString(), + iosPush.toString(), + unifiedPush.toString(), + requestRate); if (nodeInfo) { auto& ipv4 = nodeInfo->ipv4; if (ipv4.table_depth > 1) @@ -142,7 +154,8 @@ DhtProxyServer::ServerStats::toJson() const } restinio::request_handling_status_t -DhtProxyServer::serverError(restinio::request_t& request) { +DhtProxyServer::serverError(restinio::request_t& request) +{ auto response = initHttpResponse(request.create_response(restinio::status_internal_server_error())); response.set_body(RESP_MSG_INTERNAL_SERVER_ERRROR); return response.done(); @@ -154,7 +167,8 @@ class DhtProxyServer::ConnectionListener { public: ConnectionListener() {}; - explicit ConnectionListener(std::function onClosed) : onClosed_(std::move(onClosed)) {}; + explicit ConnectionListener(std::function onClosed) + : onClosed_(std::move(onClosed)) {}; ~ConnectionListener() {}; /** @@ -258,23 +272,25 @@ DhtProxyServer::Listener::msgpack_unpack(const msgpack::object& o) #endif DhtProxyServer::DhtProxyServer(const std::shared_ptr& dht, - const ProxyServerConfig& config, - const std::shared_ptr& logger -) - : ioContext_(std::make_shared()), - dht_(dht), persistPath_(config.persistStatePath), logger_(logger), - printStatsTimer_(std::make_unique(*ioContext_, 3s)), - serverStartTime_(clock::now()), - connListener_(std::make_shared(std::bind(&DhtProxyServer::onConnectionClosed, this, std::placeholders::_1))), - pushServer_(config.pushServer), - bundleId_(config.bundleId) + const ProxyServerConfig& config, + const std::shared_ptr& logger) + : ioContext_(std::make_shared()) + , dht_(dht) + , persistPath_(config.persistStatePath) + , logger_(logger) + , printStatsTimer_(std::make_unique(*ioContext_, 3s)) + , serverStartTime_(clock::now()) + , connListener_(std::make_shared( + std::bind(&DhtProxyServer::onConnectionClosed, this, std::placeholders::_1))) + , pushServer_(config.pushServer) + , bundleId_(config.bundleId) { if (not dht_) throw std::invalid_argument("A DHT instance must be provided"); if (logger_) logger_->d("[proxy:server] [init] running on %i", config.port); - if (not pushServer_.empty()){ + if (not pushServer_.empty()) { #ifdef OPENDHT_PUSH_NOTIFICATIONS if (logger_) logger_->d("[proxy:server] [init] using push server %s", pushServer_.c_str()); @@ -287,23 +303,24 @@ DhtProxyServer::DhtProxyServer(const std::shared_ptr& dht, jsonBuilder_["commentStyle"] = "None"; jsonBuilder_["indentation"] = ""; - if (!pushServer_.empty()){ + if (!pushServer_.empty()) { // no host delim, assume port only if (pushServer_.find(":") == std::string::npos) - pushServer_ = "localhost:" + pushServer_; + pushServer_ = "localhost:" + pushServer_; // define http request destination for push notifications pushHostPort_ = splitPort(pushServer_); if (logger_) - logger_->d("Using push server for notifications: %s:%s", pushHostPort_.first.c_str(), - pushHostPort_.second.c_str()); + logger_->d("Using push server for notifications: %s:%s", + pushHostPort_.first.c_str(), + pushHostPort_.second.c_str()); } if (config.identity.first and config.identity.second) { asio::error_code ec; // define tls context - asio::ssl::context tls_context { asio::ssl::context::sslv23 }; - tls_context.set_options(asio::ssl::context::default_workarounds - | asio::ssl::context::no_sslv2 - | asio::ssl::context::single_dh_use, ec); + asio::ssl::context tls_context {asio::ssl::context::sslv3_server}; + tls_context.set_options(asio::ssl::context::default_workarounds | asio::ssl::context::no_sslv2 + | asio::ssl::context::single_dh_use, + ec); if (ec) throw std::runtime_error("Error setting tls context options: " + ec.message()); // add more security options @@ -312,13 +329,14 @@ DhtProxyServer::DhtProxyServer(const std::shared_ptr& dht, #endif // node private key auto key = config.identity.first->serialize(); - tls_context.use_private_key(asio::const_buffer{key.data(), key.size()}, - asio::ssl::context::file_format::pem, ec); + tls_context.use_private_key(asio::const_buffer {key.data(), key.size()}, + asio::ssl::context::file_format::pem, + ec); if (ec) throw std::runtime_error("Error setting node's private key: " + ec.message()); // certificate chain - auto certchain = config.identity.second->toString(true/*chain*/); - tls_context.use_certificate_chain(asio::const_buffer{certchain.data(), certchain.size()}, ec); + auto certchain = config.identity.second->toString(true /*chain*/); + tls_context.use_certificate_chain(asio::const_buffer {certchain.data(), certchain.size()}, ec); if (ec) throw std::runtime_error("Error setting certificate chain: " + ec.message()); if (logger_) @@ -331,32 +349,23 @@ DhtProxyServer::DhtProxyServer(const std::shared_ptr& dht, settings.port(config.port); settings.tls_context(std::move(tls_context)); httpsServer_ = std::make_unique>( - ioContext_, - std::forward>(std::move(settings)) - ); - // run http server - serverThread_ = std::thread([this]{ - httpsServer_->open_async([]{/*ok*/}, [](std::exception_ptr ex){ - std::rethrow_exception(ex); - }); + ioContext_, std::forward>(std::move(settings))); + // run https server + serverThread_ = std::thread([this] { + httpsServer_->open_async([] { /*ok*/ }, [](std::exception_ptr ex) { std::rethrow_exception(ex); }); httpsServer_->io_context().run(); }); - } - else { + } else { auto settings = restinio::run_on_this_thread_settings_t(); addServerSettings(settings); if (not config.address.empty()) settings.address(config.address); settings.port(config.port); httpServer_ = std::make_unique>( - ioContext_, - std::forward>(std::move(settings)) - ); + ioContext_, std::forward>(std::move(settings))); // run http server - serverThread_ = std::thread([this](){ - httpServer_->open_async([]{/*ok*/}, [](std::exception_ptr ex){ - std::rethrow_exception(ex); - }); + serverThread_ = std::thread([this]() { + httpServer_->open_async([] { /*ok*/ }, [](std::exception_ptr ex) { std::rethrow_exception(ex); }); httpServer_->io_context().run(); }); } @@ -371,7 +380,10 @@ DhtProxyServer::DhtProxyServer(const std::shared_ptr& dht, std::streamsize size = stateFile.tellg(); stateFile.seekg(0, std::ios::beg); if (logger_) - logger_->d("Loading proxy state from %.*s (%td bytes)", (int)persistPath_.size(), persistPath_.c_str(), size); + logger_->d("Loading proxy state from %.*s (%td bytes)", + (int) persistPath_.size(), + persistPath_.c_str(), + size); loadState(stateFile, size); } } catch (const std::exception& e) { @@ -381,9 +393,10 @@ DhtProxyServer::DhtProxyServer(const std::shared_ptr& dht, } } -template +template void -DhtProxyServer::saveState(Os& stream) { +DhtProxyServer::saveState(Os& stream) +{ msgpack::packer pk(&stream); pk.pack_map(2); { @@ -400,9 +413,10 @@ DhtProxyServer::saveState(Os& stream) { #endif } -template +template void -DhtProxyServer::loadState(Is& is, size_t size) { +DhtProxyServer::loadState(Is& is, size_t size) +{ msgpack::unpacker pac; pac.reserve_buffer(size); if (is.read(pac.buffer(), size)) { @@ -419,12 +433,19 @@ DhtProxyServer::loadState(Is& is, size_t size) { logger_->d("Loading %zu persistent puts", puts_.size()); for (auto& put : puts_) { for (auto& pput : put.second.puts) { - pput.second.expireTimer = std::make_unique(io_context(), pput.second.expiration); - pput.second.expireTimer->async_wait(std::bind(&DhtProxyServer::handleCancelPermamentPut, this, - std::placeholders::_1, put.first, pput.first)); + pput.second.expireTimer = std::make_unique(io_context(), + pput.second.expiration); + pput.second.expireTimer->async_wait(std::bind(&DhtProxyServer::handleCancelPermamentPut, + this, + std::placeholders::_1, + put.first, + pput.first)); #ifdef OPENDHT_PUSH_NOTIFICATIONS if (not pput.second.pushToken.empty()) { - auto jsonProvider = [infoHash=put.first.toString(), clientId=pput.second.clientId, vid = pput.first, sessionCtx = pput.second.sessionCtx](){ + auto jsonProvider = [infoHash = put.first.toString(), + clientId = pput.second.clientId, + vid = pput.first, + sessionCtx = pput.second.sessionCtx]() { Json::Value json; json["timeout"] = infoHash; json["to"] = clientId; @@ -435,13 +456,20 @@ DhtProxyServer::loadState(Is& is, size_t size) { } return json; }; - pput.second.expireNotifyTimer = std::make_unique(io_context(), pput.second.expiration - proxy::OP_MARGIN); - pput.second.expireNotifyTimer->async_wait(std::bind( - &DhtProxyServer::handleNotifyPushListenExpire, this, - std::placeholders::_1, pput.second.pushToken, std::move(jsonProvider), pput.second.type, pput.second.topic)); + pput.second.expireNotifyTimer = std::make_unique(io_context(), + pput.second.expiration + - proxy::OP_MARGIN); + pput.second.expireNotifyTimer->async_wait( + std::bind(&DhtProxyServer::handleNotifyPushListenExpire, + this, + std::placeholders::_1, + pput.second.pushToken, + std::move(jsonProvider), + pput.second.type, + pput.second.topic)); } #endif - dht_->put(put.first, pput.second.value, DoneCallbackSimple{}, time_point::max(), true); + dht_->put(put.first, pput.second.value, DoneCallbackSimple {}, time_point::max(), true); } } } else { @@ -458,16 +486,32 @@ DhtProxyServer::loadState(Is& is, size_t size) { for (auto& listeners : pushListener.second.listeners) { for (auto& listener : listeners.second) { // start listening - listener.internalToken = dht_->listen(listeners.first, - [this, infoHash=listeners.first, pushToken=pushListener.first, type=listener.type, - clientId=listener.clientId, sessionCtx = listener.sessionCtx, topic=listener.topic] - (const std::vector>& values, bool expired) { - return this->handlePushListen(infoHash, pushToken, type, clientId, sessionCtx, topic, values, expired); - } - ); + listener.internalToken + = dht_->listen(listeners.first, + [this, + infoHash = listeners.first, + pushToken = pushListener.first, + type = listener.type, + clientId = listener.clientId, + sessionCtx = listener.sessionCtx, + topic = listener.topic](const std::vector>& values, + bool expired) { + return this->handlePushListen(infoHash, + pushToken, + type, + clientId, + sessionCtx, + topic, + values, + expired); + }); // expire notify - listener.expireNotifyTimer = std::make_unique(io_context(), listener.expiration - proxy::OP_MARGIN); - auto jsonProvider = [infoHash = listeners.first.toString(), clientId = listener.clientId, sessionCtx = listener.sessionCtx](){ + listener.expireNotifyTimer = std::make_unique(io_context(), + listener.expiration + - proxy::OP_MARGIN); + auto jsonProvider = [infoHash = listeners.first.toString(), + clientId = listener.clientId, + sessionCtx = listener.sessionCtx]() { Json::Value json; json["timeout"] = infoHash; json["to"] = clientId; @@ -475,12 +519,23 @@ DhtProxyServer::loadState(Is& is, size_t size) { json["s"] = sessionCtx->sessionId; return json; }; - listener.expireNotifyTimer->async_wait(std::bind(&DhtProxyServer::handleNotifyPushListenExpire, this, - std::placeholders::_1, pushListener.first, std::move(jsonProvider), listener.type, listener.topic)); + listener.expireNotifyTimer->async_wait( + std::bind(&DhtProxyServer::handleNotifyPushListenExpire, + this, + std::placeholders::_1, + pushListener.first, + std::move(jsonProvider), + listener.type, + listener.topic)); // cancel push listen - listener.expireTimer = std::make_unique(io_context(), listener.expiration); - listener.expireTimer->async_wait(std::bind(&DhtProxyServer::handleCancelPushListen, this, - std::placeholders::_1, pushListener.first, listeners.first, listener.clientId)); + listener.expireTimer = std::make_unique(io_context(), + listener.expiration); + listener.expireTimer->async_wait(std::bind(&DhtProxyServer::handleCancelPushListen, + this, + std::placeholders::_1, + pushListener.first, + listeners.first, + listener.clientId)); } } } @@ -495,7 +550,6 @@ DhtProxyServer::loadState(Is& is, size_t size) { } } - asio::io_context& DhtProxyServer::io_context() const { @@ -506,7 +560,7 @@ DhtProxyServer::~DhtProxyServer() { if (not persistPath_.empty()) { if (logger_) - logger_->d("Saving proxy state to %.*s", (int)persistPath_.size(), persistPath_.c_str()); + logger_->d("Saving proxy state to %.*s", (int) persistPath_.size(), persistPath_.c_str()); std::ofstream stateFile(persistPath_, std::ios::binary); saveState(stateFile); } @@ -518,8 +572,8 @@ DhtProxyServer::~DhtProxyServer() l.second.response->done(); } #ifdef OPENDHT_PUSH_NOTIFICATIONS - for (auto& lm: pushListeners_) { - for (auto& ls: lm.second.listeners) + for (auto& lm : pushListeners_) { + for (auto& ls : lm.second.listeners) for (auto& l : ls.second) { if (l.expireNotifyTimer) l.expireNotifyTimer->cancel(); @@ -540,7 +594,7 @@ DhtProxyServer::~DhtProxyServer() logger_->d("[proxy:server] http server closed"); } -template< typename ServerSettings > +template void DhtProxyServer::addServerSettings(ServerSettings& settings, const unsigned int max_pipelined_requests) { @@ -565,9 +619,9 @@ DhtProxyServer::addServerSettings(ServerSettings& settings, const unsigned int m settings.write_http_response_timelimit(60s); settings.handle_request_timeout(timeout_request); // socket options - settings.socket_options_setter([](auto & options){ - options.set_option(asio::ip::tcp::no_delay{true}); - options.set_option(asio::socket_base::keep_alive{true}); + settings.socket_options_setter([](auto& options) { + options.set_option(asio::ip::tcp::no_delay {true}); + options.set_option(asio::socket_base::keep_alive {true}); }); settings.connection_state_listener(connListener_); } @@ -604,8 +658,9 @@ DhtProxyServer::updateStats(std::shared_ptr info) const } void -DhtProxyServer::updateStats() { - dht_->getNodeInfo([this](std::shared_ptr newInfo){ +DhtProxyServer::updateStats() +{ + dht_->getNodeInfo([this](std::shared_ptr newInfo) { stats_ = updateStats(newInfo); nodeInfo_ = newInfo; if (logger_) { @@ -616,7 +671,7 @@ DhtProxyServer::updateStats() { } void -DhtProxyServer::handlePrintStats(const asio::error_code &ec) +DhtProxyServer::handlePrintStats(const asio::error_code& ec) { if (ec == asio::error::operation_aborted) return; @@ -625,8 +680,9 @@ DhtProxyServer::handlePrintStats(const asio::error_code &ec) printStatsTimer_->async_wait(std::bind(&DhtProxyServer::handlePrintStats, this, std::placeholders::_1)); } -template -HttpResponse DhtProxyServer::initHttpResponse(HttpResponse response) +template +HttpResponse +DhtProxyServer::initHttpResponse(HttpResponse response) { response.append_header("Server", "RESTinio"); response.append_header(restinio::http_field::content_type, "application/json"); @@ -644,8 +700,7 @@ DhtProxyServer::createRestRouter() // node.info router->http_get("/", std::bind(&DhtProxyServer::getNodeInfo, this, _1, _2)); // key.options - router->add_handler(restinio::http_method_options(), - "/:hash", std::bind(&DhtProxyServer::options, this, _1, _2)); + router->add_handler(restinio::http_method_options(), "/:hash", std::bind(&DhtProxyServer::options, this, _1, _2)); // key.get router->http_get("/:hash", std::bind(&DhtProxyServer::get, this, _1, _2)); // key.post @@ -653,11 +708,13 @@ DhtProxyServer::createRestRouter() #ifdef OPENDHT_PUSH_NOTIFICATIONS // key.subscribe router->add_handler(restinio::http_method_subscribe(), - "/:hash", std::bind(&DhtProxyServer::subscribe, this, _1, _2)); + "/:hash", + std::bind(&DhtProxyServer::subscribe, this, _1, _2)); // key.unsubscribe router->add_handler(restinio::http_method_unsubscribe(), - "/:hash", std::bind(&DhtProxyServer::unsubscribe, this, _1, _2)); -#endif //OPENDHT_PUSH_NOTIFICATIONS + "/:hash", + std::bind(&DhtProxyServer::unsubscribe, this, _1, _2)); +#endif // OPENDHT_PUSH_NOTIFICATIONS #ifdef OPENDHT_PROXY_SERVER_IDENTITY #endif // OPENDHT_PROXY_SERVER_IDENTITY @@ -673,7 +730,8 @@ DhtProxyServer::createRestRouter() // key.post router->http_post("/key/:hash", std::bind(&DhtProxyServer::put, this, _1, _2)); router->add_handler(restinio::http_method_options(), - "/key/:hash/listen", std::bind(&DhtProxyServer::options, this, _1, _2)); + "/key/:hash/listen", + std::bind(&DhtProxyServer::options, this, _1, _2)); // key.listen router->http_get("/key/:hash/listen", std::bind(&DhtProxyServer::listen, this, _1, _2)); #ifdef OPENDHT_PUSH_NOTIFICATIONS @@ -681,11 +739,13 @@ DhtProxyServer::createRestRouter() router->http_post("/node/pingPush", std::bind(&DhtProxyServer::pingPush, this, _1, _2)); // key.subscribe router->add_handler(restinio::http_method_subscribe(), - "/key/:hash", std::bind(&DhtProxyServer::subscribe, this, _1, _2)); + "/key/:hash", + std::bind(&DhtProxyServer::subscribe, this, _1, _2)); // key.unsubscribe router->add_handler(restinio::http_method_unsubscribe(), - "/key/:hash", std::bind(&DhtProxyServer::unsubscribe, this, _1, _2)); -#endif //OPENDHT_PUSH_NOTIFICATIONS + "/key/:hash", + std::bind(&DhtProxyServer::unsubscribe, this, _1, _2)); +#endif // OPENDHT_PUSH_NOTIFICATIONS #ifdef OPENDHT_PROXY_SERVER_IDENTITY // key.sign router->http_post("/key/:hash/sign", std::bind(&DhtProxyServer::putSigned, this, _1, _2)); @@ -697,8 +757,7 @@ DhtProxyServer::createRestRouter() } RequestStatus -DhtProxyServer::getNodeInfo(restinio::request_handle_t request, - restinio::router::route_params_t /*params*/) const +DhtProxyServer::getNodeInfo(restinio::request_handle_t request, restinio::router::route_params_t /*params*/) const { try { if (auto nodeInfo = nodeInfo_) { @@ -718,8 +777,7 @@ DhtProxyServer::getNodeInfo(restinio::request_handle_t request, } RequestStatus -DhtProxyServer::getStats(restinio::request_handle_t request, - restinio::router::route_params_t /*params*/) +DhtProxyServer::getStats(restinio::request_handle_t request, restinio::router::route_params_t /*params*/) { requestNum_++; try { @@ -732,14 +790,13 @@ DhtProxyServer::getStats(restinio::request_handle_t request, response.set_body(RESP_MSG_SERVICE_UNAVAILABLE); return response.done(); } - } catch (...){ + } catch (...) { return serverError(*request); } } RequestStatus -DhtProxyServer::get(restinio::request_handle_t request, - restinio::router::route_params_t params) +DhtProxyServer::get(restinio::request_handle_t request, restinio::router::route_params_t params) { requestNum_++; try { @@ -749,27 +806,26 @@ DhtProxyServer::get(restinio::request_handle_t request, auto response = std::make_shared( initHttpResponse(request->create_response())); response->flush(); - dht_->get(infoHash, [this, response](const std::vector>& values) { - std::ostringstream output; - for (const auto& value : values) { - output << Json::writeString(jsonBuilder_, value->toJson()) << "\n"; - } - response->append_chunk(output.str()); - response->flush(); - return true; - }, - [response] (bool /*ok*/){ - response->done(); - }); + dht_->get( + infoHash, + [this, response](const std::vector>& values) { + std::ostringstream output; + for (const auto& value : values) { + output << Json::writeString(jsonBuilder_, value->toJson()) << "\n"; + } + response->append_chunk(output.str()); + response->flush(); + return true; + }, + [response](bool /*ok*/) { response->done(); }); return restinio::request_handling_status_t::accepted; - } catch (const std::exception& e){ + } catch (const std::exception& e) { return serverError(*request); } } RequestStatus -DhtProxyServer::listen(restinio::request_handle_t request, - restinio::router::route_params_t params) +DhtProxyServer::listen(restinio::request_handle_t request, restinio::router::route_params_t params) { requestNum_++; @@ -782,12 +838,11 @@ DhtProxyServer::listen(restinio::request_handle_t request, response->flush(); std::lock_guard lock(lockListener_); // save the listener to handle a disconnect - auto &session = listeners_[request->connection_id()]; + auto& session = listeners_[request->connection_id()]; session.hash = infoHash; session.response = response; - session.token = dht_->listen(infoHash, [this, response] - (const std::vector>& values, bool expired){ - for (const auto& value: values){ + session.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; @@ -797,7 +852,7 @@ DhtProxyServer::listen(restinio::request_handle_t request, return true; }); return restinio::request_handling_status_t::accepted; - } catch (const std::exception& e){ + } catch (const std::exception& e) { return serverError(*request); } } @@ -805,7 +860,8 @@ DhtProxyServer::listen(restinio::request_handle_t request, #ifdef OPENDHT_PUSH_NOTIFICATIONS PushType -DhtProxyServer::getTypeFromString(const std::string& type) { +DhtProxyServer::getTypeFromString(const std::string& type) +{ if (type == "android") return PushType::Android; else if (type == "ios") @@ -816,13 +872,13 @@ DhtProxyServer::getTypeFromString(const std::string& type) { } std::string -DhtProxyServer::getDefaultTopic(PushType) { +DhtProxyServer::getDefaultTopic(PushType) +{ return bundleId_; } RequestStatus -DhtProxyServer::pingPush(restinio::request_handle_t request, - restinio::router::route_params_t /*params*/) +DhtProxyServer::pingPush(restinio::request_handle_t request, restinio::router::route_params_t /*params*/) { requestNum_++; try { @@ -830,14 +886,14 @@ DhtProxyServer::pingPush(restinio::request_handle_t request, Json::Value r; auto* char_data = reinterpret_cast(request->body().data()); auto reader = std::unique_ptr(jsonReaderBuilder_.newCharReader()); - if (!reader->parse(char_data, char_data + request->body().size(), &r, &err)){ + if (!reader->parse(char_data, char_data + request->body().size(), &r, &err)) { auto response = initHttpResponse(request->create_response(restinio::status_bad_request())); response.set_body(RESP_MSG_JSON_INCORRECT); return response.done(); } const Json::Value& root(r); // parse using const Json so [] never creates element auto pushToken = root["key"].asString(); - if (pushToken.empty()){ + if (pushToken.empty()) { auto response = initHttpResponse(request->create_response(restinio::status_bad_request())); response.set_body(RESP_MSG_NO_TOKEN); return response.done(); @@ -858,8 +914,7 @@ DhtProxyServer::pingPush(restinio::request_handle_t request, } RequestStatus -DhtProxyServer::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_++; try { @@ -871,14 +926,14 @@ DhtProxyServer::subscribe(restinio::request_handle_t request, Json::Value r; auto* char_data = reinterpret_cast(request->body().data()); auto reader = std::unique_ptr(jsonReaderBuilder_.newCharReader()); - if (!reader->parse(char_data, char_data + request->body().size(), &r, &err)){ + if (!reader->parse(char_data, char_data + request->body().size(), &r, &err)) { auto response = initHttpResponse(request->create_response(restinio::status_bad_request())); response.set_body(RESP_MSG_JSON_INCORRECT); return response.done(); } const Json::Value& root(r); // parse using const Json so [] never creates element auto pushToken = root["key"].asString(); - if (pushToken.empty()){ + if (pushToken.empty()) { auto response = initHttpResponse(request->create_response(restinio::status_bad_request())); response.set_body(RESP_MSG_NO_TOKEN); return response.done(); @@ -892,7 +947,10 @@ DhtProxyServer::subscribe(restinio::request_handle_t request, auto sessionId = root["session_id"].asString(); if (logger_) - logger_->d("[proxy:server] [subscribe %s] [client %s] [session %s]", infoHash.toString().c_str(), clientId.c_str(), sessionId.c_str()); + logger_->d("[proxy:server] [subscribe %s] [client %s] [session %s]", + infoHash.toString().c_str(), + clientId.c_str(), + sessionId.c_str()); // Insert new or return existing push listeners of a token std::lock_guard lock(lockListener_); @@ -904,7 +962,7 @@ DhtProxyServer::subscribe(restinio::request_handle_t request, }); bool newListener = listIt == pushListeners.end(); if (newListener) { - pushListeners.emplace_back(Listener{}); + pushListeners.emplace_back(Listener {}); listIt = std::prev(pushListeners.end()); listIt->clientId = clientId; listIt->sessionCtx = std::make_shared(sessionId); @@ -923,7 +981,7 @@ DhtProxyServer::subscribe(restinio::request_handle_t request, listener.expireNotifyTimer->expires_at(timeout - proxy::OP_MARGIN); else listener.expireNotifyTimer = std::make_unique(io_context(), timeout - proxy::OP_MARGIN); - auto jsonProvider = [h=infoHash.toString(), clientId, sessionCtx = listener.sessionCtx](){ + auto jsonProvider = [h = infoHash.toString(), clientId, sessionCtx = listener.sessionCtx]() { Json::Value json; json["timeout"] = h; json["to"] = clientId; @@ -931,33 +989,45 @@ DhtProxyServer::subscribe(restinio::request_handle_t request, json["s"] = sessionCtx->sessionId; return json; }; - listener.expireNotifyTimer->async_wait(std::bind(&DhtProxyServer::handleNotifyPushListenExpire, this, - std::placeholders::_1, pushToken, std::move(jsonProvider), listener.type, listener.topic)); + listener.expireNotifyTimer->async_wait(std::bind(&DhtProxyServer::handleNotifyPushListenExpire, + this, + std::placeholders::_1, + pushToken, + std::move(jsonProvider), + listener.type, + listener.topic)); if (!listener.expireTimer) listener.expireTimer = std::make_unique(io_context(), timeout); else listener.expireTimer->expires_at(timeout); - listener.expireTimer->async_wait(std::bind(&DhtProxyServer::handleCancelPushListen, this, - std::placeholders::_1, pushToken, infoHash, clientId)); + listener.expireTimer->async_wait(std::bind(&DhtProxyServer::handleCancelPushListen, + this, + std::placeholders::_1, + pushToken, + infoHash, + clientId)); // Send response if (not newListener) { if (logger_) - logger_->d("[proxy:server] [subscribe %s] found [client %s]", infoHash.toString().c_str(), listener.clientId.c_str()); + logger_->d("[proxy:server] [subscribe %s] found [client %s]", + infoHash.toString().c_str(), + listener.clientId.c_str()); // Send response header - auto response = std::make_shared(initHttpResponse(request->create_response())); + auto response = std::make_shared( + initHttpResponse(request->create_response())); response->flush(); if (!root["refresh"].asBool()) { // No Refresh - dht_->get(infoHash, [this, response](const Sp& value) { - auto output = Json::writeString(jsonBuilder_, value->toJson()) + "\n"; - response->append_chunk(output); - response->flush(); - return true; - }, - [response] (bool){ - response->done(); - }); + dht_->get( + infoHash, + [this, response](const Sp& value) { + auto output = Json::writeString(jsonBuilder_, value->toJson()) + "\n"; + response->append_chunk(output); + response->flush(); + return true; + }, + [response](bool) { response->done(); }); } else { // Refresh response->append_chunk("{}\n"); @@ -969,26 +1039,35 @@ DhtProxyServer::subscribe(restinio::request_handle_t request, if (logger_) logger_->d("[proxy:server] [subscribe %s] new", infoHash.toString().c_str()); listener.internalToken = dht_->listen(infoHash, - [this, infoHash, pushToken, type, clientId, sessionCtx = listener.sessionCtx, topic] - (const std::vector>& values, bool expired) { - return this->handlePushListen(infoHash, pushToken, type, clientId, sessionCtx, topic, values, expired); - } - ); + [this, + infoHash, + pushToken, + type, + clientId, + sessionCtx = listener.sessionCtx, + topic](const std::vector>& values, bool expired) { + return this->handlePushListen(infoHash, + pushToken, + type, + clientId, + sessionCtx, + topic, + values, + expired); + }); // Send response header auto response = initHttpResponse(request->create_response()); response.set_body("{}\n"); return response.done(); } - } - catch (...) { + } catch (...) { return serverError(*request); } return restinio::request_handling_status_t::accepted; } RequestStatus -DhtProxyServer::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_++; @@ -1005,9 +1084,8 @@ DhtProxyServer::unsubscribe(restinio::request_handle_t request, auto* char_data = reinterpret_cast(request->body().data()); auto reader = std::unique_ptr(jsonReaderBuilder_.newCharReader()); - if (!reader->parse(char_data, char_data + request->body().size(), &root, &err)){ - auto response = initHttpResponse( - request->create_response(restinio::status_bad_request())); + if (!reader->parse(char_data, char_data + request->body().size(), &root, &err)) { + auto response = initHttpResponse(request->create_response(restinio::status_bad_request())); response.set_body(RESP_MSG_JSON_INCORRECT); return response.done(); } @@ -1019,15 +1097,17 @@ DhtProxyServer::unsubscribe(restinio::request_handle_t request, handleCancelPushListen(asio::error_code() /*success*/, pushToken, infoHash, clientId); auto response = initHttpResponse(request->create_response()); return response.done(); - } - catch (...) { + } catch (...) { return serverError(*request); } } void -DhtProxyServer::handleNotifyPushListenExpire(const asio::error_code &ec, const std::string pushToken, - std::function jsonProvider, PushType type, const std::string& topic) +DhtProxyServer::handleNotifyPushListenExpire(const asio::error_code& ec, + const std::string pushToken, + std::function jsonProvider, + PushType type, + const std::string& topic) { if (ec == asio::error::operation_aborted) return; @@ -1041,19 +1121,19 @@ DhtProxyServer::handleNotifyPushListenExpire(const asio::error_code &ec, const s } void -DhtProxyServer::handleCancelPushListen(const asio::error_code &ec, const std::string pushToken, - const InfoHash key, const std::string clientId) +DhtProxyServer::handleCancelPushListen(const asio::error_code& ec, + const std::string pushToken, + const InfoHash key, + const std::string clientId) { if (ec == asio::error::operation_aborted) return; - else if (ec){ + else if (ec) { if (logger_) - logger_->e("[proxy:server] [listen:push %s] error cancel: %s", - key.toString().c_str(), ec.message().c_str()); + logger_->e("[proxy:server] [listen:push %s] error cancel: %s", key.toString().c_str(), ec.message().c_str()); } if (logger_) - logger_->d("[proxy:server] [listen:push %s] cancelled for %s", - key.toString().c_str(), clientId.c_str()); + logger_->d("[proxy:server] [listen:push %s] cancelled for %s", key.toString().c_str(), clientId.c_str()); std::lock_guard lock(lockListener_); auto pushListener = pushListeners_.find(pushToken); @@ -1063,8 +1143,8 @@ DhtProxyServer::handleCancelPushListen(const asio::error_code &ec, const std::st 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); @@ -1079,9 +1159,14 @@ DhtProxyServer::handleCancelPushListen(const asio::error_code &ec, const std::st } bool -DhtProxyServer::handlePushListen(const InfoHash& infoHash, const std::string& pushToken, PushType type, const std::string& clientId, - const std::shared_ptr& sessionCtx, const std::string& topic, - const std::vector>& values, bool expired) +DhtProxyServer::handlePushListen(const InfoHash& infoHash, + const std::string& pushToken, + PushType type, + const std::string& clientId, + const std::shared_ptr& sessionCtx, + const std::string& topic, + const std::vector>& values, + bool expired) { Json::Value json; json["key"] = infoHash.toString(); @@ -1097,8 +1182,9 @@ DhtProxyServer::handlePushListen(const InfoHash& infoHash, const std::string& pu // Build a comma-separated list of ids from the values. This is intended to be used when // streaming from the iOS notification extension in order to filter out unwanted values. std::string ids; - for(const auto& value : values) { - if (!ids.empty()) ids += ","; + for (const auto& value : values) { + if (!ids.empty()) + ids += ","; ids += std::to_string(value->id); } json["ids"] = ids; @@ -1115,7 +1201,12 @@ DhtProxyServer::handlePushListen(const InfoHash& infoHash, const std::string& pu if (logger_) logger_->d("[proxy:server] [listen %s] [client %s] [session %s] [expired %i] [priority %i] [values %zu]", - infoHash.toString().c_str(), clientId.c_str(), sessionCtx->sessionId.c_str(), expired, minPriority, values.size()); + infoHash.toString().c_str(), + clientId.c_str(), + sessionCtx->sessionId.c_str(), + expired, + minPriority, + values.size()); sendPushNotification(pushToken, std::move(json), type, !expired and minPriority == 0, topic); @@ -1123,7 +1214,8 @@ DhtProxyServer::handlePushListen(const InfoHash& infoHash, const std::string& pu } void -DhtProxyServer::sendPushNotification(const std::string& token, Json::Value&& json, PushType type, bool highPriority, const std::string& topic) +DhtProxyServer::sendPushNotification( + const std::string& token, Json::Value&& json, PushType type, bool highPriority, const std::string& topic) { if (pushServer_.empty() && type != PushType::UnifiedPush) return; @@ -1131,17 +1223,29 @@ DhtProxyServer::sendPushNotification(const std::string& token, Json::Value&& jso unsigned reqid = 0; try { std::shared_ptr request; - http::Url tokenUrl(token); - if (type == PushType::UnifiedPush) - request = std::make_shared(io_context(), tokenUrl.protocol + "://" + tokenUrl.host + (tokenUrl.service.empty() ? "" : (":" + tokenUrl.service)), logger_); - else - request = std::make_shared(io_context(), pushHostPort_.first, pushHostPort_.second, pushHostPort_.first.find("https://") == 0, logger_);; + if (type == PushType::UnifiedPush) { + http::Url tokenUrl(token); + request = std::make_shared(io_context(), + tokenUrl.protocol + "://" + tokenUrl.host, + tokenUrl.service, + tokenUrl.protocol == "https", + logger_); + request->set_target(tokenUrl.target); + request->set_header_field(restinio::http_field_t::host, tokenUrl.host); + } else { + request = std::make_shared(io_context(), + pushHostPort_.first, + pushHostPort_.second, + pushHostPort_.first.find("https") == 0, + logger_); + request->set_target("/api/push"); + request->set_header_field(restinio::http_field_t::host, pushServer_); + } reqid = request->id(); - request->set_target(type == PushType::UnifiedPush ? (tokenUrl.target) : "/api/push"); request->set_method(restinio::http_method_post()); - request->set_header_field(restinio::http_field_t::host, type == PushType::UnifiedPush ? tokenUrl.host.c_str() : pushServer_.c_str()); request->set_header_field(restinio::http_field_t::user_agent, "RESTinio client"); request->set_header_field(restinio::http_field_t::accept, "*/*"); + if (type == PushType::UnifiedPush) { // Unified Push with AND_3 expects encrypted WebPush requests. We can make // our request look like a real WebPush request by adding Content-Encoding @@ -1151,17 +1255,15 @@ DhtProxyServer::sendPushNotification(const std::string& token, Json::Value&& jso request->set_header_field(restinio::http_field_t::content_encoding, "aes128gcm"); request->set_header_field(restinio::http_field_t::ttl, "86400"); request->set_header_field(restinio::http_field_t::urgency, highPriority ? "high" : "normal"); - } else { - request->set_header_field(restinio::http_field_t::content_type, "application/json"); - } - if (type == PushType::UnifiedPush) { Json::Value notification(Json::objectValue); notification["message"] = Json::writeString(jsonBuilder_, std::move(json)); notification["topic"] = token; notification["priority"] = highPriority ? 5 : 1; request->set_body(Json::writeString(jsonBuilder_, std::move(json))); } else { + request->set_header_field(restinio::http_field_t::content_type, "application/json"); + // NOTE: see https://github.com/appleboy/gorush auto isResubscribe = json.isMember("timeout"); Json::Value notification(Json::objectValue); @@ -1185,7 +1287,7 @@ DhtProxyServer::sendPushNotification(const std::string& token, Json::Value&& jso notification["topic"] = topic; if (highPriority || isResubscribe) { Json::Value alert(Json::objectValue); - alert["title"]="hello"; + alert["title"] = "hello"; notification["push_type"] = "alert"; notification["alert"] = alert; notification["mutable_content"] = true; @@ -1197,16 +1299,15 @@ DhtProxyServer::sendPushNotification(const std::string& token, Json::Value&& jso } Json::Value notifications(Json::arrayValue); - notifications[0] = notification; + notifications[0] = std::move(notification); Json::Value content; content["notifications"] = std::move(notifications); request->set_body(Json::writeString(jsonBuilder_, content)); } - request->add_on_state_change_callback([this, reqid] - (http::Request::State state, const http::Response& response){ - if (state == http::Request::State::DONE){ + request->add_on_state_change_callback([this, reqid](http::Request::State state, const http::Response& response) { + if (state == http::Request::State::DONE) { if (logger_ and response.status_code != 200) logger_->e("[proxy:server] [notification] push failed: %i", response.status_code); std::lock_guard l(requestLock_); @@ -1233,8 +1334,7 @@ DhtProxyServer::sendPushNotification(const std::string& token, Json::Value&& jso default: break; } - } - catch (const std::exception &e){ + } catch (const std::exception& e) { if (logger_) logger_->e("[proxy:server] [notification] error send push: %s", e.what()); if (reqid) { @@ -1244,14 +1344,14 @@ DhtProxyServer::sendPushNotification(const std::string& token, Json::Value&& jso } } -#endif //OPENDHT_PUSH_NOTIFICATIONS +#endif // OPENDHT_PUSH_NOTIFICATIONS void -DhtProxyServer::handleCancelPermamentPut(const asio::error_code &ec, const InfoHash& key, Value::Id vid) +DhtProxyServer::handleCancelPermamentPut(const asio::error_code& ec, const InfoHash& key, Value::Id vid) { if (ec == asio::error::operation_aborted) return; - else if (ec){ + else if (ec) { if (logger_) logger_->e("[proxy:server] [put:permament] error sending put refresh: %s", ec.message().c_str()); } @@ -1277,15 +1377,14 @@ DhtProxyServer::handleCancelPermamentPut(const asio::error_code &ec, const InfoH } RequestStatus -DhtProxyServer::put(restinio::request_handle_t request, - restinio::router::route_params_t params) +DhtProxyServer::put(restinio::request_handle_t request, restinio::router::route_params_t params) { requestNum_++; InfoHash infoHash(params["hash"]); if (!infoHash) infoHash = InfoHash::get(params["hash"]); - if (request->body().empty()){ + if (request->body().empty()) { auto response = initHttpResponse(request->create_response(restinio::status_bad_request())); response.set_body(RESP_MSG_MISSING_PARAMS); return response.done(); @@ -1297,16 +1396,18 @@ DhtProxyServer::put(restinio::request_handle_t request, auto* char_data = reinterpret_cast(request->body().data()); auto reader = std::unique_ptr(jsonReaderBuilder_.newCharReader()); - if (reader->parse(char_data, char_data + request->body().size(), &root, &err)){ + if (reader->parse(char_data, char_data + request->body().size(), &root, &err)) { auto value = std::make_shared(root); bool permanent = root.isMember("permanent"); if (logger_) - logger_->d("[proxy:server] [put %s] %s %s", infoHash.toString().c_str(), - value->toString().c_str(), (permanent ? "permanent" : "")); + logger_->d("[proxy:server] [put %s] %s %s", + infoHash.toString().c_str(), + value->toString().c_str(), + (permanent ? "permanent" : "")); if (permanent) { std::string pushToken, clientId, sessionId, platform, topic; auto& pVal = root["permanent"]; - if (pVal.isObject()){ + if (pVal.isObject()) { pushToken = pVal["key"].asString(); clientId = pVal["client_id"].asString(); platform = pVal["platform"].asString(); @@ -1318,13 +1419,14 @@ DhtProxyServer::put(restinio::request_handle_t request, auto& sPuts = puts_[infoHash]; if (value->id == Value::INVALID_ID) { for (auto& pp : sPuts.puts) { - if (pp.second.pushToken == pushToken - and pp.second.clientId == clientId - and pp.second.value->contentEquals(*value)) - { + if (pp.second.pushToken == pushToken and pp.second.clientId == clientId + and pp.second.value->contentEquals(*value)) { pp.second.expireTimer->expires_at(timeout); - pp.second.expireTimer->async_wait(std::bind(&DhtProxyServer::handleCancelPermamentPut, this, - std::placeholders::_1, infoHash, pp.second.value->id)); + pp.second.expireTimer->async_wait(std::bind(&DhtProxyServer::handleCancelPermamentPut, + this, + std::placeholders::_1, + infoHash, + pp.second.value->id)); if (not sessionId.empty()) { if (not pp.second.sessionCtx) pp.second.sessionCtx = std::make_shared(sessionId); @@ -1338,7 +1440,7 @@ DhtProxyServer::put(restinio::request_handle_t request, return response.done(); } } - value->id = std::uniform_int_distribution{1}(rd); + value->id = std::uniform_int_distribution {1}(rd); } auto vid = value->id; @@ -1370,13 +1472,13 @@ DhtProxyServer::put(restinio::request_handle_t request, } pput.expireTimer->expires_at(timeout); } - pput.expireTimer->async_wait(std::bind(&DhtProxyServer::handleCancelPermamentPut, this, - std::placeholders::_1, infoHash, vid)); + pput.expireTimer->async_wait( + std::bind(&DhtProxyServer::handleCancelPermamentPut, this, std::placeholders::_1, infoHash, vid)); #ifdef OPENDHT_PUSH_NOTIFICATIONS // notify put permanent expiration if (pput.sessionCtx) { - auto jsonProvider = [infoHash, clientId, vid, sessionCtx = pput.sessionCtx](){ + auto jsonProvider = [infoHash, clientId, vid, sessionCtx = pput.sessionCtx]() { Json::Value json; json["timeout"] = infoHash.toString(); json["to"] = clientId; @@ -1387,34 +1489,43 @@ DhtProxyServer::put(restinio::request_handle_t request, }; if (!pput.expireNotifyTimer) pput.expireNotifyTimer = std::make_unique(io_context(), - timeout - proxy::OP_MARGIN); + timeout - proxy::OP_MARGIN); else pput.expireNotifyTimer->expires_at(timeout - proxy::OP_MARGIN); - pput.expireNotifyTimer->async_wait(std::bind( - &DhtProxyServer::handleNotifyPushListenExpire, this, - std::placeholders::_1, pushToken, std::move(jsonProvider), pput.type, pput.topic)); + pput.expireNotifyTimer->async_wait(std::bind(&DhtProxyServer::handleNotifyPushListenExpire, + this, + std::placeholders::_1, + pushToken, + std::move(jsonProvider), + pput.type, + pput.topic)); } #endif } - dht_->put(infoHash, value, [this, request, value](bool ok){ - if (ok){ - auto response = initHttpResponse(request->create_response()); - response.append_body(Json::writeString(jsonBuilder_, value->toJson()) + "\n"); - response.done(); - } else { - auto response = initHttpResponse(request->create_response(restinio::status_bad_gateway())); - response.set_body(RESP_MSG_PUT_FAILED); - response.done(); - } - }, time_point::max(), permanent); + dht_->put( + infoHash, + value, + [this, request, value](bool ok) { + if (ok) { + auto response = initHttpResponse(request->create_response()); + response.append_body(Json::writeString(jsonBuilder_, value->toJson()) + "\n"); + response.done(); + } else { + auto response = initHttpResponse(request->create_response(restinio::status_bad_gateway())); + response.set_body(RESP_MSG_PUT_FAILED); + response.done(); + } + }, + time_point::max(), + permanent); return restinio::request_handling_status_t::accepted; } else { auto response = initHttpResponse(request->create_response(restinio::status_bad_request())); response.set_body(RESP_MSG_JSON_INCORRECT); return response.done(); } - } catch (const std::exception& e){ + } catch (const std::exception& e) { if (logger_) logger_->d("[proxy:server] error in put: %s", e.what()); return serverError(*request); @@ -1424,15 +1535,14 @@ DhtProxyServer::put(restinio::request_handle_t request, #ifdef OPENDHT_PROXY_SERVER_IDENTITY RequestStatus -DhtProxyServer::putSigned(restinio::request_handle_t request, - restinio::router::route_params_t params) const +DhtProxyServer::putSigned(restinio::request_handle_t request, restinio::router::route_params_t params) const { requestNum_++; InfoHash infoHash(params["hash"]); if (!infoHash) infoHash = InfoHash::get(params["hash"]); - if (request->body().empty()){ + if (request->body().empty()) { auto response = initHttpResponse(request->create_response(restinio::status_bad_request())); response.set_body(RESP_MSG_MISSING_PARAMS); return response.done(); @@ -1444,12 +1554,11 @@ DhtProxyServer::putSigned(restinio::request_handle_t request, auto* char_data = reinterpret_cast(request->body().data()); auto reader = std::unique_ptr(jsonReaderBuilder_.newCharReader()); - if (reader->parse(char_data, char_data + request->body().size(), &root, &err)){ - + 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){ + dht_->putSigned(infoHash, value, [this, request, value](bool ok) { + if (ok) { auto output = Json::writeString(jsonBuilder_, value->toJson()) + "\n"; auto response = initHttpResponse(request->create_response()); response.append_body(output); @@ -1466,7 +1575,7 @@ DhtProxyServer::putSigned(restinio::request_handle_t request, response.set_body(RESP_MSG_JSON_INCORRECT); return response.done(); } - } catch (const std::exception& e){ + } catch (const std::exception& e) { if (logger_) logger_->d("[proxy:server] error in putSigned: %s", e.what()); return serverError(*request); @@ -1474,15 +1583,14 @@ DhtProxyServer::putSigned(restinio::request_handle_t request, } RequestStatus -DhtProxyServer::putEncrypted(restinio::request_handle_t request, - restinio::router::route_params_t params) +DhtProxyServer::putEncrypted(restinio::request_handle_t request, restinio::router::route_params_t params) { requestNum_++; InfoHash infoHash(params["hash"]); if (!infoHash) infoHash = InfoHash::get(params["hash"]); - if (request->body().empty()){ + if (request->body().empty()) { auto response = initHttpResponse(request->create_response(restinio::status_bad_request())); response.set_body(RESP_MSG_MISSING_PARAMS); return response.done(); @@ -1494,16 +1602,16 @@ DhtProxyServer::putEncrypted(restinio::request_handle_t request, auto* char_data = reinterpret_cast(request->body().data()); auto reader = std::unique_ptr(jsonReaderBuilder_.newCharReader()); - if (reader->parse(char_data, char_data + request->body().size(), &root, &err)){ + if (reader->parse(char_data, char_data + request->body().size(), &root, &err)) { InfoHash to(root["to"].asString()); - if (!to){ + if (!to) { auto response = initHttpResponse(request->create_response(restinio::status_bad_request())); response.set_body(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){ + dht_->putEncrypted(infoHash, to, value, [this, request, value](bool ok) { + if (ok) { auto response = initHttpResponse(request->create_response()); response.append_body(Json::writeString(jsonBuilder_, value->toJson()) + "\n"); response.done(); @@ -1519,7 +1627,7 @@ DhtProxyServer::putEncrypted(restinio::request_handle_t request, response.set_body(RESP_MSG_JSON_INCORRECT); return response.done(); } - } catch (const std::exception& e){ + } catch (const std::exception& e) { if (logger_) logger_->d("[proxy:server] error in put: %s", e.what()); return serverError(*request); @@ -1529,8 +1637,7 @@ DhtProxyServer::putEncrypted(restinio::request_handle_t request, #endif // OPENDHT_PROXY_SERVER_IDENTITY RequestStatus -DhtProxyServer::options(restinio::request_handle_t request, - restinio::router::route_params_t /*params*/) +DhtProxyServer::options(restinio::request_handle_t request, restinio::router::route_params_t /*params*/) { requestNum_++; #ifdef OPENDHT_PROXY_SERVER_IDENTITY @@ -1546,8 +1653,7 @@ DhtProxyServer::options(restinio::request_handle_t request, } RequestStatus -DhtProxyServer::getFiltered(restinio::request_handle_t request, - restinio::router::route_params_t params) +DhtProxyServer::getFiltered(restinio::request_handle_t request, restinio::router::route_params_t params) { requestNum_++; auto query = params["value"]; @@ -1559,20 +1665,20 @@ DhtProxyServer::getFiltered(restinio::request_handle_t request, auto response = std::make_shared( initHttpResponse(request->create_response())); response->flush(); - dht_->get(infoHash, + dht_->get( + infoHash, [this, response](const Sp& value) { response->append_chunk(Json::writeString(jsonBuilder_, value->toJson()) + "\n"); response->flush(); return true; }, - [response] (bool /*ok*/){ - response->done(); - }, - {}, query); + [response](bool /*ok*/) { response->done(); }, + {}, + query); return restinio::request_handling_status_t::accepted; - } catch (const std::exception& e){ + } catch (const std::exception& e) { return serverError(*request); } } -} +} // namespace dht