From 70f09ded3fe1deceaae6ca4081974745a5b460ea Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=A9bastien=20Blin?= Date: Tue, 8 Jan 2019 16:29:05 -0500 Subject: [PATCH] dhtproxy: avoid notification after subscribe to retrieve values --- include/opendht/dht_proxy_client.h | 2 +- src/dht_proxy_client.cpp | 69 +++++++++++++++++++++++++----- src/dht_proxy_server.cpp | 28 +++++++----- 3 files changed, 77 insertions(+), 22 deletions(-) diff --git a/include/opendht/dht_proxy_client.h b/include/opendht/dht_proxy_client.h index 71d30027..c94ebdd1 100644 --- a/include/opendht/dht_proxy_client.h +++ b/include/opendht/dht_proxy_client.h @@ -281,7 +281,7 @@ private: struct ListenState; void sendListen(const std::shared_ptr& request, const ValueCallback&, const Value::Filter& filter, const Sp& state); - void sendSubscribe(const std::shared_ptr& request, const Sp& state); + void sendSubscribe(const std::shared_ptr& request, const ValueCallback&, const Value::Filter& filter, const Sp& state); void doPut(const InfoHash&, Sp, DoneCallback, time_point created, bool permanent); diff --git a/src/dht_proxy_client.cpp b/src/dht_proxy_client.cpp index 49640949..6aa1d689 100644 --- a/src/dht_proxy_client.cpp +++ b/src/dht_proxy_client.cpp @@ -750,19 +750,62 @@ DhtProxyClient::sendListen(const std::shared_ptr& req, const V } void -DhtProxyClient::sendSubscribe(const std::shared_ptr& req, const Sp& state) { +DhtProxyClient::sendSubscribe(const std::shared_ptr& req, const ValueCallback& cb, const Value::Filter& filter, const Sp& state) { #if OPENDHT_PUSH_NOTIFICATIONS req->set_method("SUBSCRIBE"); try { fillBody(req); - restbed::Http::async(req, [this,state](const std::shared_ptr&, - const std::shared_ptr& reply) { + restbed::Http::async(req, [this, state, cb, filter, req](const std::shared_ptr&, + const std::shared_ptr& reply) { + auto code = reply->get_status_code(); if (code == 200) { try { - restbed::Http::fetch("\n", reply); - auto code = reply->get_status_code(); - state->ok = code == 200; + DHT_LOG.e("################1."); + while (restbed::Http::is_open(req) and not state->cancel) { + DHT_LOG.e("################2."); + restbed::Http::fetch("\n", reply); + DHT_LOG.e("################3."); + auto code = reply->get_status_code(); + DHT_LOG.e("################4."); + state->ok = code == 200; + if (state->cancel) { + DHT_LOG.e("################5."); + + break; + } + std::string body; + reply->get_body(body); + reply->set_body(""); // Reset the body for the next fetch + DHT_LOG.e("###BODY: %s\n", body.c_str());; + if (body.empty() || body == "{}") { + DHT_LOG.e("################6."); + break; + + } + DHT_LOG.e("################7."); + + Json::Value json; + std::string err; + Json::CharReaderBuilder rbuilder; + auto reader = std::unique_ptr(rbuilder.newCharReader()); + if (reader->parse(body.data(), body.data() + body.size(), &json, &err)) { + if (json.size() == 0) { + DHT_LOG.e("################*."); + break; + } + auto expired = json.get("expired", Json::Value(false)).asBool(); + 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, expired]() { + if (not state->cancel and not cb({value}, expired)) + state->cancel = true; + }); + loopSignal_(); + } + } + } } catch (const std::exception& e) { if (not state->cancel) { DHT_LOG.e("sendSubscribe: error: %s", e.what()); @@ -864,12 +907,14 @@ DhtProxyClient::doListen(const InfoHash& key, ValueCallback cb, Value::Filter fi } } }); - l->second.thread = std::thread([this, req, state](){ - sendSubscribe(req, state); + auto vcb = l->second.cb; + auto filter = l->second.filter; + l->second.thread = std::thread([this, req, vcb, filter, state](){ + sendSubscribe(req, vcb, filter, state); }); } else { l->second.thread = std::thread([this, req, vcb, filter, state]{ - sendListen(req,vcb,filter,state); + sendListen(req, vcb, filter, state); }); } return token; @@ -1080,8 +1125,10 @@ DhtProxyClient::resubscribe(const InfoHash& key, Listener& listener) state->ok = true; listener.req = req; scheduler.edit(listener.refreshJob, scheduler.time() + proxy::OP_TIMEOUT - proxy::OP_MARGIN); - listener.thread = std::thread([this, req, state]() { - sendSubscribe(req, state); + auto vcb = listener.cb; + auto filter = listener.filter; + listener.thread = std::thread([this, req, vcb, filter, state]() { + sendSubscribe(req, vcb, filter, state); }); #endif } diff --git a/src/dht_proxy_server.cpp b/src/dht_proxy_server.cpp index 09403315..3f99a252 100644 --- a/src/dht_proxy_server.cpp +++ b/src/dht_proxy_server.cpp @@ -287,7 +287,7 @@ DhtProxyServer::get(const Sp& session) const Json::StreamWriterBuilder wbuilder; wbuilder["commentStyle"] = "None"; wbuilder["indentation"] = ""; - auto output = Json::writeString(wbuilder, value->toJson()) + "\n"; + auto output = Json::writeString(wbuilder, value->toJson()) + "\n"; s->yield(output, [](const Sp& /*session*/){ }); return true; }, [s](bool /*ok* */) { @@ -428,17 +428,24 @@ DhtProxyServer::subscribe(const std::shared_ptr& session) if (listener.clientId == clientId) { scheduler_.edit(listener.expireJob, timeout); scheduler_.edit(listener.expireNotifyJob, timeout - proxy::OP_MARGIN); - s->close(restbed::OK, "{}\n"); - schedulerCv_.notify_one(); dht_->get(infoHash, - [this, infoHash, pushToken, isAndroid, clientId](std::vector> /*value*/) { - // Build message content. - Json::Value json; - json["key"] = infoHash.toString(); - json["to"] = clientId; - sendPushNotification(pushToken, json, isAndroid); + [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; - }, [](bool /*ok* */) {}); + }, [s](bool /*ok* */) { + // Communication is finished + if (not s->is_closed()) { + std::cout << "X: " << std::endl; + s->close(); + } + }); + schedulerCv_.notify_one(); return; } } @@ -476,6 +483,7 @@ DhtProxyServer::subscribe(const std::shared_ptr& session) ); } schedulerCv_.notify_one(); + std::cout << "CLOSE: " << std::endl; s->close(restbed::OK, "{}\n"); } catch (...) { s->close(restbed::INTERNAL_SERVER_ERROR, "{\"err\":\"Internal server error\"}");