diff --git a/src/dht.cpp b/src/dht.cpp index 2ebd09cc..277911ce 100644 --- a/src/dht.cpp +++ b/src/dht.cpp @@ -811,6 +811,7 @@ Dht::listen(const InfoHash& id, GetCallback cb, Value::Filter f, Where where) } if (!newvals.empty()) { if (!cb(newvals)) { + // cancelListen is useful here, because we need to cancel on IPv4 and 6 cancelListen(id, token); return false; } diff --git a/src/dht_proxy_client.cpp b/src/dht_proxy_client.cpp index e934150b..6d2c6853 100644 --- a/src/dht_proxy_client.cpp +++ b/src/dht_proxy_client.cpp @@ -200,7 +200,7 @@ DhtProxyClient::get(const InfoHash& key, const GetCallback& cb, DoneCallback don if ((not filterChain or filterChain(*value)) && cb) { std::lock_guard lock(lockCallbacks); callbacks_.emplace_back([cb, value, finished]() { - if (not cb({value})) + if (not *finished and not cb({value})) *finished = true; }); loopSignal_(); @@ -445,14 +445,40 @@ DhtProxyClient::listen(const InfoHash& key, GetCallback cb, Value::Filter filter Listener l; ++listener_token_; + auto isCanceled = std::make_shared(false); + + // Wrap cb in another callback to avoid value duplication. + auto vals = std::make_shared>>(); + auto gcb = [=](const std::vector>& values) { + std::vector> newvals; + for (const auto& v : values) { + auto it = vals->find(v->id); + if (it == vals->cend() || !(*it->second == *v)) + newvals.push_back(v); + } + if (!newvals.empty()) { + if (not *isCanceled and !cb(newvals)) { + *isCanceled = true; + cancelListen(key, listener_token_); + return false; + } + for (const auto& v : newvals) { + auto it = vals->emplace(v->id, v); + if (not it.second) + it.first->second = v; + } + } + return true; + }; + l.key = key.toString(); l.token = listener_token_; l.req = req; - l.cb = cb; + l.cb = gcb; l.callbackId = callbackId; l.pushNotifToken = pushNotifToken; l.filterChain = std::move(filterChain); - l.isCanceledViaClose = std::make_shared(false); + l.isCanceledViaClose = isCanceled; std::weak_ptr isCanceledViaClose(l.isCanceledViaClose); l.thread = std::thread([=]() { @@ -472,7 +498,7 @@ DhtProxyClient::listen(const InfoHash& key, GetCallback cb, Value::Filter filter }; auto state = std::make_shared(); restbed::Http::async(req, - [this, filterChain, cb, pushNotifToken, state](const std::shared_ptr& req, + [this, filterChain, gcb, pushNotifToken, state](const std::shared_ptr& req, const std::shared_ptr& reply) { auto code = reply->get_status_code(); if (code == 200) { @@ -507,12 +533,11 @@ DhtProxyClient::listen(const InfoHash& key, GetCallback cb, Value::Filter filter auto* char_data = reinterpret_cast(&body[0]); if (reader->parse(char_data, char_data + body.size(), &json, &err)) { auto value = std::make_shared(json); - if ((not filterChain or filterChain(*value)) && cb) { + if (not filterChain or filterChain(*value)) { std::lock_guard lock(lockCallbacks); - callbacks_.emplace_back([cb, value, state]() { - if (not state->cancel and not cb({value})) { + callbacks_.emplace_back([gcb, value, state]() { + if (not state->cancel and not gcb({value})) state->cancel = true; - } }); loopSignal_(); } diff --git a/src/value.cpp b/src/value.cpp index fddbcb16..54b96f3e 100644 --- a/src/value.cpp +++ b/src/value.cpp @@ -176,8 +176,13 @@ Value::msgpack_unpack_body(const msgpack::object& o) Value::Value(Json::Value& json) { try { - if (json.isMember("id")) - id = ValueType::Id(json["id"].asInt()); + if (json.isMember("id")) { + if (json["id"].isString()) { + id = Value::Id(std::stoull(json["id"].asString())); + } else { + id = Value::Id(json["id"].asLargestUInt()); + } + } } catch (...) { } if (json.isMember("cypher")) { auto cypherStr = json["cypher"].asString();