From 75a14d360892b0267bc980deac8b62b8f868feaa Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=A9bastien=20Blin?= Date: Thu, 17 Jan 2019 14:08:53 -0500 Subject: [PATCH] proxy: only get old values at first subscribe --- include/opendht/dht_proxy_client.h | 12 ++++++-- src/dht_proxy_client.cpp | 44 +++++++++++++++++----------- src/dht_proxy_server.cpp | 47 ++++++++++++++++++++---------- 3 files changed, 68 insertions(+), 35 deletions(-) diff --git a/include/opendht/dht_proxy_client.h b/include/opendht/dht_proxy_client.h index 88af36e9..59eba71e 100644 --- a/include/opendht/dht_proxy_client.h +++ b/include/opendht/dht_proxy_client.h @@ -280,7 +280,15 @@ private: bool doCancelListen(const InfoHash& key, size_t token); struct ListenState; - void sendListen(const std::shared_ptr& request, const ValueCallback&, const Value::Filter& filter, const Sp& state, bool usePush = false); + enum class ListenMethod { + LISTEN, + SUBSCRIBE, + RESUBSCRIBE, + }; + void sendListen(const std::shared_ptr &request, + const ValueCallback &, const Value::Filter &filter, + const Sp &state, + ListenMethod method = ListenMethod::LISTEN); void doPut(const InfoHash&, Sp, DoneCallback, time_point created, bool permanent); @@ -372,7 +380,7 @@ private: const std::function loopSignal_; #if OPENDHT_PUSH_NOTIFICATIONS - void fillBody(std::shared_ptr request); + void fillBody(std::shared_ptr request, bool resubscribe); void getPushRequest(Json::Value&) const; #endif // OPENDHT_PUSH_NOTIFICATIONS diff --git a/src/dht_proxy_client.cpp b/src/dht_proxy_client.cpp index e0df087e..da5ae862 100644 --- a/src/dht_proxy_client.cpp +++ b/src/dht_proxy_client.cpp @@ -692,20 +692,24 @@ DhtProxyClient::cancelListen(const InfoHash& key, size_t gtoken) { return canceled; } - -void -DhtProxyClient::sendListen(const std::shared_ptr& req, const ValueCallback& cb, const Value::Filter& filter, const Sp& state, bool usePush) { - auto settings = std::make_shared(); - if (usePush) { - req->set_method("SUBSCRIBE"); - } else { - std::chrono::milliseconds timeout(std::numeric_limits::max()); - settings->set_connection_timeout(timeout); // Avoid the client to close the socket after 5 seconds. - req->set_method("LISTEN"); - } - try { +void DhtProxyClient::sendListen(const std::shared_ptr &req, + const ValueCallback &cb, + const Value::Filter &filter, + const Sp &state, + ListenMethod method) { + auto settings = std::make_shared(); + if (method != ListenMethod::LISTEN) { + req->set_method("SUBSCRIBE"); + } else { + std::chrono::milliseconds timeout(std::numeric_limits::max()); + settings->set_connection_timeout( + timeout); // Avoid the client to close the socket after 5 seconds. + req->set_method("LISTEN"); + } + try { #if OPENDHT_PUSH_NOTIFICATIONS - if (usePush) fillBody(req); + if (method != ListenMethod::LISTEN) + fillBody(req, method == ListenMethod::RESUBSCRIBE); #endif restbed::Http::async(req, [this, filter, cb, state](const std::shared_ptr& req, @@ -843,8 +847,10 @@ DhtProxyClient::doListen(const InfoHash& key, ValueCallback cb, Value::Filter fi } }); } - l->second.thread = std::thread([this, req, vcb, filter, state](){ - sendListen(req, vcb, filter, state, !deviceKey_.empty()); + l->second.thread = std::thread([this, req, vcb, filter, state]() { + sendListen(req, vcb, filter, state, + deviceKey_.empty() ? ListenMethod::LISTEN + : ListenMethod::SUBSCRIBE); }); return token; } @@ -1057,7 +1063,7 @@ DhtProxyClient::resubscribe(const InfoHash& key, Listener& listener) auto vcb = listener.cb; auto filter = listener.filter; listener.thread = std::thread([this, req, vcb, filter, state]() { - sendListen(req, vcb, filter, state, true); + sendListen(req, vcb, filter, state, ListenMethod::RESUBSCRIBE); }); #endif } @@ -1077,7 +1083,7 @@ DhtProxyClient::getPushRequest(Json::Value& body) const } void -DhtProxyClient::fillBody(std::shared_ptr req) +DhtProxyClient::fillBody(std::shared_ptr req, bool resubscribe) { // Fill body with // { @@ -1085,6 +1091,10 @@ DhtProxyClient::fillBody(std::shared_ptr req) // } Json::Value body; getPushRequest(body); + if (!resubscribe) { + // This is the first listen, we want to retrieve previous values. + body["previous_values"] = true; + } Json::StreamWriterBuilder wbuilder; wbuilder["commentStyle"] = "None"; wbuilder["indentation"] = ""; diff --git a/src/dht_proxy_server.cpp b/src/dht_proxy_server.cpp index e51dc782..8860db5d 100644 --- a/src/dht_proxy_server.cpp +++ b/src/dht_proxy_server.cpp @@ -429,22 +429,37 @@ DhtProxyServer::subscribe(const std::shared_ptr& session) scheduler_.edit(listener.expireJob, timeout); scheduler_.edit(listener.expireNotifyJob, timeout - proxy::OP_MARGIN); s->yield(restbed::OK); - dht_->get(infoHash, - [this, s](const Sp& value) { - if (s->is_closed()) return false; - // Send values as soon as we get them - Json::StreamWriterBuilder wbuilder; - wbuilder["commentStyle"] = "None"; - wbuilder["indentation"] = ""; - auto output = Json::writeString(wbuilder, 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"); - } - }); + + if (root.isMember("previous_values") && + root["previous_values"].asBool()) { + dht_->get( + infoHash, + [this, s](const Sp &value) { + if (s->is_closed()) + return false; + // Send values as soon as we get them + Json::StreamWriterBuilder wbuilder; + wbuilder["commentStyle"] = "None"; + wbuilder["indentation"] = ""; + auto output = Json::writeString( + wbuilder, 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; }