diff --git a/include/opendht/dht_proxy_client.h b/include/opendht/dht_proxy_client.h index c94ebdd1..88af36e9 100644 --- a/include/opendht/dht_proxy_client.h +++ b/include/opendht/dht_proxy_client.h @@ -280,8 +280,7 @@ 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); - void sendSubscribe(const std::shared_ptr& request, const ValueCallback&, const Value::Filter& filter, const Sp& state); + void sendListen(const std::shared_ptr& request, const ValueCallback&, const Value::Filter& filter, const Sp& state, bool usePush = false); 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 6aa1d689..e0df087e 100644 --- a/src/dht_proxy_client.cpp +++ b/src/dht_proxy_client.cpp @@ -694,12 +694,19 @@ DhtProxyClient::cancelListen(const InfoHash& key, size_t gtoken) { void -DhtProxyClient::sendListen(const std::shared_ptr& req, const ValueCallback& cb, const Value::Filter& filter, const Sp& state) { +DhtProxyClient::sendListen(const std::shared_ptr& req, const ValueCallback& cb, const Value::Filter& filter, const Sp& state, bool usePush) { auto settings = std::make_shared(); - 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"); + 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 { +#if OPENDHT_PUSH_NOTIFICATIONS + if (usePush) fillBody(req); +#endif restbed::Http::async(req, [this, filter, cb, state](const std::shared_ptr& req, const std::shared_ptr& reply) { @@ -719,6 +726,10 @@ DhtProxyClient::sendListen(const std::shared_ptr& req, const V 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) { + // Empty value, it's the end + break; + } auto expired = json.get("expired", Json::Value(false)).asBool(); auto value = std::make_shared(json); if ((not filter or filter(*value)) and cb) { @@ -749,82 +760,6 @@ DhtProxyClient::sendListen(const std::shared_ptr& req, const V opFailed(); } -void -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, cb, filter, req](const std::shared_ptr&, - const std::shared_ptr& reply) { - - auto code = reply->get_status_code(); - if (code == 200) { - try { - 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()); - state->ok = false; - } - } - } else { - state->ok = false; - } - }).get(); - } catch(const std::exception& e) { - state->ok = false; - } - auto& s = *state; - if (not s.ok and not s.cancel) - opFailed(); -#endif -} - size_t DhtProxyClient::doListen(const InfoHash& key, ValueCallback cb, Value::Filter filter/*, Where where*/) { @@ -907,16 +842,10 @@ DhtProxyClient::doListen(const InfoHash& key, ValueCallback cb, Value::Filter fi } } }); - 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); - }); } + l->second.thread = std::thread([this, req, vcb, filter, state](){ + sendListen(req, vcb, filter, state, !deviceKey_.empty()); + }); return token; } @@ -1128,7 +1057,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]() { - sendSubscribe(req, vcb, filter, state); + sendListen(req, vcb, filter, state, true); }); #endif } diff --git a/src/dht_proxy_server.cpp b/src/dht_proxy_server.cpp index 3f99a252..ef9c544a 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,6 +428,7 @@ 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->yield(restbed::OK); dht_->get(infoHash, [this, s](const Sp& value) { if (s->is_closed()) return false; @@ -441,8 +442,7 @@ DhtProxyServer::subscribe(const std::shared_ptr& session) }, [s](bool /*ok* */) { // Communication is finished if (not s->is_closed()) { - std::cout << "X: " << std::endl; - s->close(); + s->close("{}\n"); } }); schedulerCv_.notify_one(); @@ -483,7 +483,6 @@ 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\"}");