allow to cancel listen through dhtrunner

include/opendht/dhtrunner.h

@@ -37,6 +37,7 @@
 #include <mutex>
 #include <atomic>
 #include <condition_variable>
+#include <future>
 #include <exception>
 #include <queue>
 
@@ -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<size_t> listen(InfoHash hash, Dht::GetCallback vcb, Value::Filter f = Value::AllFilter());
+    std::future<size_t> 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<size_t> 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<std::mutex> 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);

src/dht.cpp

@@ -1153,15 +1153,20 @@ size_t
 Dht::listen(const InfoHash& id, GetCallback cb, Value::Filter f)
 {
     auto vals = std::make_shared<std::map<Value::Id, std::shared_ptr<Value>>>();
+    auto token = ++listener_token;
+
     auto gcb = [=](const std::vector<std::shared_ptr<Value>>& values) {
         std::vector<std::shared_ptr<Value>> 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<std::shared_ptr<Value>> 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<std::shared_ptr<Value>> 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)

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<size_t>
 DhtRunner::listen(InfoHash hash, Dht::GetCallback vcb, Value::Filter f)
 {
     std::unique_lock<std::mutex> lck(storage_mtx);
+    auto ret_token = std::make_shared<std::promise<size_t>>();
     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<size_t>
+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<std::mutex> 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<size_t> token)
+{
+    std::unique_lock<std::mutex> 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>(value), cb);
 }
 
+void
+DhtRunner::cancelPut(const InfoHash& h , const Value::Id& id)
+{
+    std::unique_lock<std::mutex> 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)
 {