diff --git a/include/opendht/dht_proxy_client.h b/include/opendht/dht_proxy_client.h index 52f28231..346e0906 100644 --- a/include/opendht/dht_proxy_client.h +++ b/include/opendht/dht_proxy_client.h @@ -291,6 +291,7 @@ private: void opFailed(); + void handleExpireListener(const asio::error_code &ec, const InfoHash& key); bool doCancelListen(const InfoHash& key, size_t token); struct Listener; diff --git a/src/dht_proxy_client.cpp b/src/dht_proxy_client.cpp index d12f1d12..9e3164ca 100644 --- a/src/dht_proxy_client.cpp +++ b/src/dht_proxy_client.cpp @@ -63,7 +63,7 @@ struct PermanentPut { struct DhtProxyClient::ProxySearch { SearchCache ops {}; - Sp opExpirationTimer; + std::unique_ptr opExpirationTimer; std::map listeners {}; std::map puts {}; }; @@ -80,7 +80,7 @@ DhtProxyClient::DhtProxyClient(std::function signal, const std::string& serverHostService_.second = serverHostService_.second.empty() ? "80" : serverHostService_.second; httpClient_ = std::make_unique(httpContext_, - serverHostService_.first, serverHostService_.second, logger); + serverHostService_.first, serverHostService_.second); // run http client httpClientThread_ = std::thread([this](){ try { @@ -274,8 +274,6 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, auto header_fields = this->initHeaderFields(); auto request = httpClient_->create_request(header, header_fields, restinio::http_connection_header_t::keep_alive, ""/*body*/); - if (logger_) - logger_->d(request.c_str()); struct GetContext { std::function)> cb; // wrapper @@ -375,7 +373,7 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, // keeping context data alive httpClient_->async_request(conn, request, std::move(parser), std::move(parser_s), [this, key, context](const asio::error_code& ec){ - if (ec && logger_) + if (ec && ec != asio::error::eof && logger_) logger_->e("[proxy:client] [get %s] failed: %s", key.to_c_str(), ec.message().c_str()); }); }); @@ -464,8 +462,6 @@ DhtProxyClient::doPut(const InfoHash& key, Sp val, DoneCallback cb, time_ auto body = Json::writeString(wbuilder, json); auto request = httpClient_->create_request(header, header_fields, restinio::http_connection_header_t::close, body); - if (logger_) - logger_->d("%s", request.c_str()); struct PutContext { DoneCallbackSimple donecb; @@ -519,7 +515,7 @@ DhtProxyClient::doPut(const InfoHash& key, Sp val, DoneCallback cb, time_ // keeping context data alive httpClient_->async_request(conn, request, std::move(parser), std::move(parser_s), [this, key, context](const asio::error_code& ec){ - if (ec && logger_) + if (ec && ec != asio::error::eof && logger_) logger_->e("[proxy:client] [put %s] failed: %s", key.to_c_str(), ec.message().c_str()); }); @@ -632,7 +628,7 @@ DhtProxyClient::handleProxyStatus(const asio::error_code& ec, auto request = httpClient_->create_request(header, header_fields, restinio::http_connection_header_t::keep_alive, ""/*body*/); if (logger_) - logger_->d("[proxy:client] [status] sending request:\n%s", request.c_str()); + logger_->d("[proxy:client] [status] sending request"); // initalise the parser callback data struct StatusContext { unsigned int family; @@ -704,7 +700,7 @@ DhtProxyClient::handleProxyStatus(const asio::error_code& ec, // keeping context data alive httpClient_->async_request(conn, request, std::move(parser), std::move(parser_s), [this, context](const asio::error_code& ec){ - if (ec && logger_) + if (ec && ec != asio::error::eof && logger_) logger_->e("[proxy:client] [status] failed: %s", ec.message().c_str()); }); @@ -792,14 +788,14 @@ DhtProxyClient::getPublicAddress(sa_family_t family) } size_t -DhtProxyClient::listen(const InfoHash& key, ValueCallback cb, Value::Filter filter, Where where) { +DhtProxyClient::listen(const InfoHash& key, ValueCallback cb, Value::Filter filter, Where where) +{ if (logger_) logger_->d("[proxy:client] [listen] [search %s]", key.to_c_str()); auto& search = searches_[key]; auto query = std::make_shared(Select{}, std::move(where)); - auto token = search.ops.listen(cb, query, filter, [this, key]( - Sp, ValueCallback cb, SyncCallback) -> size_t { + return search.ops.listen(cb, query, filter, [this, key](Sp, ValueCallback cb, SyncCallback) -> size_t { std::lock_guard lock(searchLock_); // Find search auto search = searches_.find(key); @@ -811,7 +807,6 @@ DhtProxyClient::listen(const InfoHash& key, ValueCallback cb, Value::Filter filt if (logger_) logger_->d("[proxy:client] [listen] [search %s] sending %s", key.to_c_str(), deviceKey_.empty() ? "listen" : "subscribe"); - // Add listener auto token = ++listenerToken_; auto l = search->second.listeners.find(token); @@ -823,8 +818,7 @@ DhtProxyClient::listen(const InfoHash& key, ValueCallback cb, Value::Filter filt if (l->second.state) l->second.state->cancel = true; } - - // Add callback + // Add cache callback auto state = std::make_shared(); l->second.state = state; l->second.cb = [this,key,token,state](const std::vector>& values, bool expired) { @@ -840,8 +834,6 @@ DhtProxyClient::listen(const InfoHash& key, ValueCallback cb, Value::Filter filt } return false; }; - auto vcb = l->second.cb; - if (not deviceKey_.empty()) { /* * Relaunch push listeners even if a timeout is not received @@ -884,54 +876,68 @@ DhtProxyClient::listen(const InfoHash& key, ValueCallback cb, Value::Filter filt header.method(restinio::http_method_subscribe()); header.request_target("/" + key.toString()); } - sendListen(header, vcb, state, l->second, method); + sendListen(header, l->second.cb, state, l->second, method); return token; }); - return token; } bool -DhtProxyClient::cancelListen(const InfoHash& key, size_t gtoken) { +DhtProxyClient::cancelListen(const InfoHash& key, size_t gtoken) +{ if (logger_) logger_->d(key, "[proxy:client] [search %s] cancel listen %zu", key.to_c_str(), gtoken); + + // find the listener in cache auto it = searches_.find(key); if (it == searches_.end()) return false; auto& ops = it->second.ops; bool canceled = ops.cancelListen(gtoken, std::chrono::steady_clock::now()); - // on new listener set the expiration to the max, - // in case a user redo a listen right after cancel, we won't impact the network. - if (!it->second.opExpirationTimer) { - it->second.opExpirationTimer = std::make_shared(httpContext_); - it->second.opExpirationTimer->expires_at(time_point::max()); - it->second.opExpirationTimer->async_wait([this, key](const asio::error_code ec){ - if (ec){ - if (logger_) - logger_->d("[proxy:client] [listen %s] error in cancel", key.toString().c_str()); - return false; - } - auto it = searches_.find(key); - if (it != searches_.end()) { - auto next = it->second.ops.expire(std::chrono::steady_clock::now(), - [this, key](size_t ltoken){ - doCancelListen(key, ltoken); - }); - if (next != time_point::max()) { - if (!it->second.opExpirationTimer){ - it->second.opExpirationTimer = std::make_shared< - asio::steady_timer>(httpContext_); - } - it->second.opExpirationTimer->expires_at(next); - } - } - }); + + // define real cancel listen only once + if (!it->second.opExpirationTimer){ + it->second.opExpirationTimer = std::make_unique(httpContext_, time_point::max()); + it->second.opExpirationTimer->async_wait(std::bind( + &DhtProxyClient::handleExpireListener, this, std::placeholders::_1, key)); + } + else { + // set expires to max if listeners else to last removed + expiration + it->second.opExpirationTimer->expires_at(ops.getExpiration()); } - // Let it expire when it is due. - it->second.opExpirationTimer->expires_at(ops.getExpiration()); - loopSignal_(); return canceled; } +void +DhtProxyClient::handleExpireListener(const asio::error_code &ec, const InfoHash& key) +{ + auto it = searches_.find(key); + // everytime a new expiry is set, a previous gets aborted + if (ec == asio::error::operation_aborted){ + it->second.opExpirationTimer->expires_at(it->second.ops.getExpiration()); + it->second.opExpirationTimer->async_wait(std::bind( + &DhtProxyClient::handleExpireListener, this, std::placeholders::_1, key)); + return; + } + else if (ec){ + if (logger_) + logger_->e("[proxy:client] [listen %s] error in cancel: %s", + key.toString().c_str(), ec.message().c_str()); + return; + } + // found an existing listener + if (it != searches_.end()){ + time_point next = it->second.ops.expire(std::chrono::steady_clock::now(), + [this, key](size_t ltoken){ + doCancelListen(key, ltoken); + }); + if (next != time_point::max()){ + it->second.opExpirationTimer->expires_at(next); + it->second.opExpirationTimer->async_wait(std::bind( + &DhtProxyClient::handleExpireListener, this, std::placeholders::_1, key)); + } + } +} + void DhtProxyClient::sendListen(const restinio::http_request_header_t header, const ValueCallback& cb, @@ -952,8 +958,6 @@ DhtProxyClient::sendListen(const restinio::http_request_header_t header, body = fillBody(method == ListenMethod::RESUBSCRIBE); #endif auto request = httpClient_->create_request(header, headers, connType, body); - if (logger_) - logger_->d(request.c_str()); struct ListenContext { std::shared_ptr logger; @@ -1032,8 +1036,8 @@ DhtProxyClient::sendListen(const restinio::http_request_header_t header, // keeping context data alive httpClient_->async_request(conn, request, std::move(parser), std::move(parser_s), [this, &listener, context](const asio::error_code& ec){ - if (ec && logger_) - logger_->e("[proxy:client] [send listen] [connection %i} failed: %s", + if (ec && ec != asio::error::eof && logger_ && !listener.state->cancel.load()) + logger_->e("[proxy:client] [send listen] [connection %i] failed: %s", listener.connId, ec.message().c_str()); }); }); @@ -1042,6 +1046,7 @@ DhtProxyClient::sendListen(const restinio::http_request_header_t header, bool DhtProxyClient::doCancelListen(const InfoHash& key, size_t ltoken) { + // only cancel on dht when it is the last listener in the cache std::lock_guard lock(searchLock_); auto search = searches_.find(key); @@ -1077,8 +1082,6 @@ DhtProxyClient::doCancelListen(const InfoHash& key, size_t ltoken) // build the request auto request = httpClient_->create_request(header, header_fields, restinio::http_connection_header_t::keep_alive, content); - if (logger_) - logger_->d(request.c_str()); // define context struct UnsubscribeContext { InfoHash key; @@ -1106,7 +1109,7 @@ DhtProxyClient::doCancelListen(const InfoHash& key, size_t ltoken) }; httpClient_->async_request(conn, request, std::move(parser), std::move(parser_s), [this, key](const asio::error_code& ec){ - if (ec && logger_) + if (ec && ec != asio::error::eof && logger_) logger_->e("[proxy:client] [unsubscribe %s] failed: %s", key.to_c_str(), ec.message().c_str()); }); }); diff --git a/src/dht_proxy_server.cpp b/src/dht_proxy_server.cpp index 17b7998b..adf3ff57 100644 --- a/src/dht_proxy_server.cpp +++ b/src/dht_proxy_server.cpp @@ -84,7 +84,7 @@ DhtProxyServer::DhtProxyServer(std::shared_ptr dht, in_port_t port, // build http client auto pushHostPort = splitPort(pushServer_); httpClient_.reset(new http::Client(httpServer_->io_context(), - pushHostPort.first, pushHostPort.second, logger_)); + pushHostPort.first, pushHostPort.second)); // run http server httpServerThread_ = std::thread([this](){ try { diff --git a/src/http.cpp b/src/http.cpp index b24bf558..c92bb3b9 100644 --- a/src/http.cpp +++ b/src/http.cpp @@ -162,7 +162,7 @@ Client::set_connection_timeout(const ConnectionId conn_id, const std::chrono::seconds timeout, HandlerCb cb) { auto& conn_context = connections_[conn_id]; - if (!conn_context.connection){ + if (logger_ && !conn_context.connection){ logger_->e("[http::client] [connection:%i] closed, can't timeout", conn_id); return; } @@ -199,10 +199,9 @@ Client::async_resolve(const std::string host, const std::string service, Handler const asio::error_code& ec, asio::ip::tcp::resolver::results_type endpoints) { - if (ec){ - if (logger_) - logger_->e("[http::client] [resolve %s:%s] error resolving: %s", - host.c_str(), service.c_str(), ec.message().c_str()); + if (ec and logger_){ + logger_->e("[http::client] [resolve %s:%s] error resolving: %s", + host.c_str(), service.c_str(), ec.message().c_str()); } else { for (auto it = endpoints.begin(); it != endpoints.end(); ++it){ @@ -283,7 +282,8 @@ Client::async_connect(ConnectionCb cb) const asio::ip::tcp::endpoint& endpoint){ // only reacts if all endpoints fail if (ec){ - logger_->e("[http::client] [connection] failed to connect to any endpoints"); + if (logger_) + logger_->e("[http::client] [connection] failed to connect to any endpoints"); close_connection(conn->id()); if (cb) cb(nullptr);