diff --git a/include/opendht/dhtrunner.h b/include/opendht/dhtrunner.h index bcd09692..e0475c5c 100644 --- a/include/opendht/dhtrunner.h +++ b/include/opendht/dhtrunner.h @@ -37,6 +37,7 @@ #include #include #include +#include #include #include @@ -63,18 +64,14 @@ public: void get(InfoHash hash, Dht::GetCallback vcb, Dht::DoneCallback dcb=nullptr, Value::Filter f = Value::AllFilter()); void get(const std::string& key, Dht::GetCallback vcb, Dht::DoneCallback dcb=nullptr, Value::Filter f = Value::AllFilter()); - void listen(InfoHash hash, Dht::GetCallback vcb, Value::Filter f = Value::AllFilter()); - void listen(const std::string& key, Dht::GetCallback vcb, Value::Filter f = Value::AllFilter()); + std::future listen(InfoHash hash, Dht::GetCallback vcb, Value::Filter f = Value::AllFilter()); + std::future listen(const std::string& key, Dht::GetCallback vcb, Value::Filter f = Value::AllFilter()); + void cancelListen(InfoHash h, size_t token); + void cancelListen(InfoHash h, std::shared_future token); void put(InfoHash hash, Value&& value, Dht::DoneCallback cb=nullptr); void put(const std::string& key, Value&& value, Dht::DoneCallback cb=nullptr); - void cancelPut(const InfoHash& h , const Value::Id& id) { - std::unique_lock lck(storage_mtx); - pending_ops.emplace([=](SecureDht& dht) { - std::cout << "Processing cancelPut " << h << " / " << id << std::endl; - dht.cancelPut(h, id); - }); - } + void cancelPut(const InfoHash& h , const Value::Id& id); void putSigned(InfoHash hash, Value&& value, Dht::DoneCallback cb=nullptr); void putSigned(const std::string& key, Value&& value, Dht::DoneCallback cb=nullptr); diff --git a/src/dht.cpp b/src/dht.cpp index 33a97696..aebf3843 100644 --- a/src/dht.cpp +++ b/src/dht.cpp @@ -1153,15 +1153,20 @@ size_t Dht::listen(const InfoHash& id, GetCallback cb, Value::Filter f) { auto vals = std::make_shared>>(); + auto token = ++listener_token; + 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) ) + if (it == vals->cend() || !(*it->second == *v)) newvals.push_back(v); } if (!newvals.empty()) { - cb(newvals); + if (!cb(newvals)) { + cancelListen(id, token); + return false; + } for (const auto& v : newvals) vals->insert({v->id, v}); } @@ -1176,20 +1181,26 @@ Dht::listen(const InfoHash& id, GetCallback cb, Value::Filter f) } if (st) { if (not st->values.empty()) { - std::vector> vals; - vals.reserve(st->values.size()); - for (auto& v : st->values) - if (f(*v.data)) vals.push_back(v.data); - if (not vals.empty()) - gcb(vals); + std::vector> newvals {}; + newvals.reserve(st->values.size()); + for (auto& v : st->values) { + if (f(*v.data)) + newvals.push_back(v.data); + } + if (not newvals.empty()) { + if (!cb(newvals)) + return 0; + for (const auto& v : newvals) + vals->insert({v->id, v}); + } } tokenlocal = ++st->listener_token; - st->local_listeners.insert({tokenlocal, {f, cb}}); + st->local_listeners.insert({tokenlocal, {f, gcb}}); } auto token4 = Dht::listenTo(id, AF_INET, gcb, f); auto token6 = Dht::listenTo(id, AF_INET6, gcb, f); - auto token = ++listener_token; + listeners.insert({token, std::make_tuple(tokenlocal, token4, token6)}); return token; } @@ -1198,8 +1209,11 @@ bool Dht::cancelListen(const InfoHash& id, size_t token) { auto it = listeners.find(token); - if (it == listeners.end()) + if (it == listeners.end()) { + DHT_WARN("Listen token not found: %d", token); return false; + } + DHT_WARN("cancelListen %s with token %d", id.toString().c_str(), token); Storage* st = findStorage(id); auto tokenlocal = std::get<0>(it->second); if (st && tokenlocal) diff --git a/src/dhtrunner.cpp b/src/dhtrunner.cpp index 89e0c0b3..d9e477c4 100644 --- a/src/dhtrunner.cpp +++ b/src/dhtrunner.cpp @@ -238,21 +238,45 @@ DhtRunner::get(const std::string& key, Dht::GetCallback vcb, Dht::DoneCallback d get(InfoHash::get(key), vcb, dcb, f); } -void +std::future DhtRunner::listen(InfoHash hash, Dht::GetCallback vcb, Value::Filter f) { std::unique_lock lck(storage_mtx); + auto ret_token = std::make_shared>(); pending_ops.emplace([=](SecureDht& dht) { std::cout << "Processing listen (" << hash << ")" << std::endl; - dht.listen(hash, vcb, f); + ret_token->set_value(dht.listen(hash, vcb, f)); }); cv.notify_all(); + return ret_token->get_future(); +} + +std::future +DhtRunner::listen(const std::string& key, Dht::GetCallback vcb, Value::Filter f) +{ + return listen(InfoHash::get(key), vcb, f); } void -DhtRunner::listen(const std::string& key, Dht::GetCallback vcb, Value::Filter f) +DhtRunner::cancelListen(InfoHash h, size_t token) { - listen(InfoHash::get(key), vcb, f); + std::unique_lock lck(storage_mtx); + pending_ops.emplace([=](SecureDht& dht) { + std::cout << "Processing cancelListen " << h << std::endl; + dht.cancelListen(h, token); + }); +} + +void +DhtRunner::cancelListen(InfoHash h, std::shared_future token) +{ + std::unique_lock lck(storage_mtx); + pending_ops.emplace([=](SecureDht& dht) { + std::cout << "Processing cancelListen (shared_future) " << h << std::endl; + auto tk = token.get(); + std::cout << "token is " << tk << std::endl; + dht.cancelListen(h, tk); + }); } void @@ -273,6 +297,16 @@ DhtRunner::put(const std::string& key, Value&& value, Dht::DoneCallback cb) put(InfoHash::get(key), std::forward(value), cb); } +void +DhtRunner::cancelPut(const InfoHash& h , const Value::Id& id) +{ + std::unique_lock lck(storage_mtx); + pending_ops.emplace([=](SecureDht& dht) { + std::cout << "Processing cancelPut " << h << " / " << id << std::endl; + dht.cancelPut(h, id); + }); +} + void DhtRunner::putSigned(InfoHash hash, Value&& value, Dht::DoneCallback cb) {