sockaddr: use more precise exception

CMakeLists.txt

@@ -1,7 +1,7 @@
 cmake_minimum_required (VERSION 3.1)
 project (opendht)
 set (opendht_VERSION_MAJOR 1)
-set (opendht_VERSION_MINOR 8.1)
+set (opendht_VERSION_MINOR 8.0)
 set (opendht_VERSION ${opendht_VERSION_MAJOR}.${opendht_VERSION_MINOR})
 set (PACKAGE_VERSION ${opendht_VERSION})
 set (VERSION "${opendht_VERSION}")
@@ -70,8 +70,6 @@ set (CMAKE_CXX_STANDARD_REQUIRED on)
 set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wno-return-type -Wall -Wextra -Wnon-virtual-dtor -pedantic-errors -fvisibility=hidden")
 if (OPENDHT_SANITIZE)
     set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -fsanitize=address -fstack-protector-strong")
-else ()
-    set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -fno-stack-protector")
 endif ()
 set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -DMSGPACK_DISABLE_LEGACY_NIL -DMSGPACK_DISABLE_LEGACY_CONVERT")
 if (NOT CMAKE_BUILD_TYPE)
@@ -306,7 +304,7 @@ install (EXPORT opendht DESTINATION ${CMAKE_INSTALL_LIBDIR}/cmake/opendht FILE o
 install (FILES ${CMAKE_CURRENT_BINARY_DIR}/opendhtConfigVersion.cmake DESTINATION ${CMAKE_INSTALL_LIBDIR}/cmake/opendht)
 
 # Unit tests
-if (OPENDHT_TESTS)
+IF(OPENDHT_TESTS)
   FIND_PACKAGE(Cppunit REQUIRED)
   # unit testing
     list (APPEND test_FILES
@@ -327,17 +325,16 @@ if (OPENDHT_TESTS)
       tests/tests_runner.cpp
       ${test_FILES}
     )
-    target_include_directories(opendht_unit_tests SYSTEM PRIVATE ${CPPUNIT_INCLUDE_DIRS})
     if (OPENDHT_SHARED)
-      target_link_libraries(opendht_unit_tests opendht)
+      TARGET_LINK_LIBRARIES(opendht_unit_tests opendht)
     else ()
-      target_link_libraries(opendht_unit_tests opendht-static)
+      TARGET_LINK_LIBRARIES(opendht_unit_tests opendht-static)
     endif ()
-    target_link_libraries(opendht_unit_tests
+    TARGET_LINK_LIBRARIES(opendht_unit_tests
        ${CMAKE_THREAD_LIBS_INIT}
        ${CPPUNIT_LIBRARIES}
        ${GNUTLS_LIBRARIES}
     )
     enable_testing()
     add_test(TEST opendht_unit_tests)
-endif()
+ENDIF()

configure.ac

@@ -1,7 +1,7 @@
 dnl define macros
 m4_define([opendht_major_version], 1)
 m4_define([opendht_minor_version], 8)
-m4_define([opendht_patch_version], 1)
+m4_define([opendht_patch_version], 0)
 m4_define([opendht_version],
 		  [opendht_major_version.opendht_minor_version.opendht_patch_version])
 

include/opendht/crypto.h

@@ -173,8 +173,6 @@ private:
     PrivateKey(const PrivateKey&) = delete;
     PrivateKey& operator=(const PrivateKey&) = delete;
     Blob decryptBloc(const uint8_t* src, size_t src_size) const;
-
-    //friend dht::crypto::Identity dht::crypto::generateIdentity(const std::string&, dht::crypto::Identity, unsigned key_length);
 };
 
 

include/opendht/dht.h

@@ -163,7 +163,7 @@ public:
     /**
      * Get locally stored data for the given hash.
      */
-    std::vector<Sp<Value>> getLocal(const InfoHash& key, Value::Filter f = {}) const;
+    std::vector<Sp<Value>> getLocal(const InfoHash& key, Value::Filter f = Value::AllFilter()) const;
 
     /**
      * Get locally stored data for the given key and value id.
@@ -324,10 +324,10 @@ private:
     static constexpr unsigned LISTEN_NODES {4};
 
     /* The maximum number of hashes we're willing to track. */
-    static constexpr unsigned MAX_HASHES {64 * 1024};
+    static constexpr unsigned MAX_HASHES {16384};
 
     /* The maximum number of searches we keep data about. */
-    static constexpr unsigned MAX_SEARCHES {64 * 1024};
+    static constexpr unsigned MAX_SEARCHES {16384};
 
     static constexpr std::chrono::minutes MAX_STORAGE_MAINTENANCE_EXPIRE_TIME {10};
 
@@ -469,7 +469,7 @@ private:
     Sp<Search> search(const InfoHash& id, sa_family_t af, GetCallback = {}, QueryCallback = {}, DoneCallback = {}, Value::Filter = {}, const Sp<Query>& q = {});
 
     void announce(const InfoHash& id, sa_family_t af, Sp<Value> value, DoneCallback callback, time_point created=time_point::max(), bool permanent = false);
-    size_t listenTo(const InfoHash& id, sa_family_t af, ValueCallback cb, Value::Filter f = {}, const Sp<Query>& q = {});
+    size_t listenTo(const InfoHash& id, sa_family_t af, ValueCallback cb, Value::Filter f = Value::AllFilter(), const Sp<Query>& q = {});
 
     /**
      * Refill the search with good nodes if possible.

include/opendht/dht_interface.h

@@ -106,7 +106,7 @@ public:
     /**
      * Get locally stored data for the given hash.
      */
-    virtual std::vector<Sp<Value>> getLocal(const InfoHash& key, Value::Filter f = {}) const = 0;
+    virtual std::vector<Sp<Value>> getLocal(const InfoHash& key, Value::Filter f = Value::AllFilter()) const = 0;
 
     /**
      * Get locally stored data for the given key and value id.

include/opendht/dht_proxy_client.h

@@ -41,6 +41,8 @@ namespace Json {
 
 namespace dht {
 
+class SearchCache;
+
 class OPENDHT_PUBLIC DhtProxyClient final : public DhtInterface {
 public:
 
@@ -274,18 +276,12 @@ private:
 
     void opFailed();
 
+    size_t doListen(const InfoHash& key, ValueCallback, Value::Filter);
     bool doCancelListen(const InfoHash& key, size_t token);
 
     struct ListenState;
-    enum class ListenMethod {
-        LISTEN,
-        SUBSCRIBE,
-        RESUBSCRIBE,
-    };
-    void sendListen(const std::shared_ptr<restbed::Request> &request,
-                    const ValueCallback &, const Value::Filter &filter,
-                    const Sp<ListenState> &state,
-                    ListenMethod method = ListenMethod::LISTEN);
+    void sendListen(const std::shared_ptr<restbed::Request>& request, const ValueCallback&, const Value::Filter& filter, const Sp<ListenState>& state);
+    void sendSubscribe(const std::shared_ptr<restbed::Request>& request, const Sp<proxy::ListenToken>&, const Sp<ListenState>& state);
 
     void doPut(const InfoHash&, Sp<Value>, DoneCallback, time_point created, bool permanent);
 
@@ -355,8 +351,6 @@ private:
      */
     void confirmProxy();
     Sp<Scheduler::Job> nextProxyConfirmation {};
-    Sp<Scheduler::Job> listenerRestart {};
-
     /**
      * Relaunch LISTEN requests if the client disconnect/reconnect.
      */
@@ -377,7 +371,7 @@ private:
     const std::function<void()> loopSignal_;
 
 #if OPENDHT_PUSH_NOTIFICATIONS
-    void fillBody(std::shared_ptr<restbed::Request> request, bool resubscribe);
+    void fillBodyToGetToken(std::shared_ptr<restbed::Request> request, unsigned token = 0);
     void getPushRequest(Json::Value&) const;
 #endif // OPENDHT_PUSH_NOTIFICATIONS
 

include/opendht/dhtrunner.h

@@ -63,20 +63,20 @@ public:
     DhtRunner();
     virtual ~DhtRunner();
 
-    void get(InfoHash id, GetCallbackSimple cb, DoneCallback donecb={}, Value::Filter f = {}, Where w = {}) {
+    void get(InfoHash id, GetCallbackSimple cb, DoneCallback donecb={}, Value::Filter f = Value::AllFilter(), Where w = {}) {
         get(id, bindGetCb(cb), donecb, f, w);
     }
 
-    void get(InfoHash id, GetCallbackSimple cb, DoneCallbackSimple donecb={}, Value::Filter f = {}, Where w = {}) {
+    void get(InfoHash id, GetCallbackSimple cb, DoneCallbackSimple donecb={}, Value::Filter f = Value::AllFilter(), Where w = {}) {
         get(id, bindGetCb(cb), donecb, f, w);
     }
 
     void get(InfoHash hash, GetCallback vcb, DoneCallback dcb, Value::Filter f={}, Where w = {});
 
-    void get(InfoHash id, GetCallback cb, DoneCallbackSimple donecb={}, Value::Filter f = {}, Where w = {}) {
+    void get(InfoHash id, GetCallback cb, DoneCallbackSimple donecb={}, Value::Filter f = Value::AllFilter(), Where w = {}) {
         get(id, cb, bindDoneCb(donecb), f, w);
     }
-    void get(const std::string& key, GetCallback vcb, DoneCallbackSimple dcb={}, Value::Filter f = {}, Where w = {});
+    void get(const std::string& key, GetCallback vcb, DoneCallbackSimple dcb={}, Value::Filter f = Value::AllFilter(), Where w = {});
 
     template <class T>
     void get(InfoHash hash, std::function<bool(std::vector<T>&&)> cb, DoneCallbackSimple dcb={})

include/opendht/network_engine.h

@@ -290,9 +290,9 @@ public:
      */
     Sp<Request> sendFindNode(Sp<Node> n,
                              const InfoHash& hash,
-                             want_t want = -1,
-                             RequestCb&& on_done = {},
-                             RequestExpiredCb&& on_expired = {});
+                             want_t want,
+                             RequestCb&& on_done,
+                             RequestExpiredCb&& on_expired);
     /**
      * Send a "get" request to a given node.
      *
@@ -531,8 +531,6 @@ private:
     std::set<SockAddr> blacklist {};
 
     Scheduler& scheduler;
-
-    bool logIncoming_ {false};
 };
 
 } /* namespace net  */

include/opendht/routing_table.h

@@ -42,11 +42,6 @@ struct Bucket {
     Sp<Node> randomNode();
 
     void sendCachedPing(net::NetworkEngine& ne);
-    void connectivityChanged() {
-        time = time_point::min();
-        for (auto& node : nodes)
-            node->setTime(time_point::min());
-    }
 };
 
 class RoutingTable : public std::list<Bucket> {
@@ -81,7 +76,7 @@ public:
     void connectivityChanged(const time_point& now) {
         grow_time = now;
         for (auto& b : *this)
-            b.connectivityChanged();
+            b.time = time_point::min();
     }
 
     bool onNewNode(const Sp<Node>& node, int comfirm, const time_point& now, const InfoHash& myid, net::NetworkEngine& ne);

include/opendht/sockaddr.h

@@ -67,7 +67,7 @@ public:
      */
     SockAddr(const sockaddr* sa, socklen_t length) {
         if (length > sizeof(sockaddr_storage))
-            throw std::runtime_error("Socket address length is too large");
+            throw std::invalid_argument("Socket address length is too large");
         set(sa, length);
     }
     SockAddr(const sockaddr* sa) {
@@ -78,7 +78,7 @@ public:
             else if(sa->sa_family == AF_INET6)
                 len = sizeof(sockaddr_in6);
             else
-                throw std::runtime_error("Unknown address family");
+                throw std::invalid_argument("Unknown address family");
         }
         set(sa, len);
     }

include/opendht/value.h

@@ -165,8 +165,8 @@ struct OPENDHT_PUBLIC Value
             return chainOr(std::move(f1), std::move(f2));
         }
         static Filter chain(Filter&& f1, Filter&& f2) {
-            if (not f1) return std::move(f1);
-            if (not f2) return std::move(f2);
+            if (not f1) return f2;
+            if (not f2) return f1;
             return [f1,f2](const Value& v) {
                 return f1(v) and f2(v);
             };
@@ -184,7 +184,7 @@ struct OPENDHT_PUBLIC Value
             return chainAll(std::vector<Filter>(l.begin(), l.end()));
         }
         static Filter chainOr(Filter&& f1, Filter&& f2) {
-            if (not f1 or not f2) return {};
+            if (not f1 or not f2) return AllFilter();
             return [f1,f2](const Value& v) {
                 return f1(v) or f2(v);
             };
@@ -577,7 +577,7 @@ struct OPENDHT_PUBLIC Value
     std::string user_type {};
 
     /**
-     * Sequence number to avoid replay attacks
+     * Sequence number
      */
     uint16_t seq {0};
 
@@ -591,8 +591,10 @@ struct OPENDHT_PUBLIC Value
      */
     Blob cypher {};
 
+    bool checkSignature();
+    Sp<Value> decrypt(const crypto::PrivateKey& key);
+
 private:
-    friend class SecureDht;
     /* Cache for crypto ops */
     bool signatureChecked {false};
     bool signatureValid {false};
@@ -990,7 +992,7 @@ template <typename T,
 Value::Filter
 getFilterSet()
 {
-    return {};
+    return Value::AllFilter();
 }
 
 template <class T>

python/opendht.pyx

@@ -139,12 +139,6 @@ cdef class SockAddr(object):
         return self._addr.setPort(port)
     def setFamily(SockAddr self, cpp.sa_family_t af):
         return self._addr.setFamily(af)
-    def isLoopback(SockAddr self):
-        return self._addr.isLoopback()
-    def isPrivate(SockAddr self):
-        return self._addr.isPrivate()
-    def isUnspecified(SockAddr self):
-        return self._addr.isUnspecified()
     def __str__(self):
         return self.toString().decode()
     def __repr__(self):

python/opendht_cpp.pxd

@@ -77,9 +77,6 @@ cdef extern from "opendht/sockaddr.h" namespace "dht":
         void setPort(in_port_t p)
         sa_family_t getFamily() const
         void setFamily(sa_family_t f)
-        bool isLoopback() const
-        bool isPrivate() const
-        bool isUnspecified() const
 
 ctypedef vector[uint8_t] Blob
 

python/tools/pingpong.py

@@ -51,7 +51,7 @@ def ping(node, h):
 
 def pong(node, h):
 	print(node.getNodeId().decode(), "got ping", h, i)
-	loop.call_soon_threadsafe(ping, node, h)
+	loop.call_soon_threadsafe(ping, node, h);
 	return True
 
 ping_node.listen(loc_ping, lambda v: pong(pong_node, loc_pong))

src/dht.cpp

@@ -125,16 +125,19 @@ Dht::trySearchInsert(const Sp<Node>& node)
     bool inserted {false};
 
     // insert forward
-    for (auto it = closest; it != srs.end(); it++) {
+    auto it = closest;
+    while (it != srs.end()) {
         auto& s = *it->second;
         if (s.insertNode(node, now)) {
             inserted = true;
             scheduler.edit(s.nextSearchStep, now);
         } else if (not s.expired and not s.done)
             break;
+        ++it;
     }
     // insert backward
-    for (auto it = closest; it != srs.begin();) {
+    it = closest;
+    while (it != srs.begin()) {
         --it;
         auto& s = *it->second;
         if (s.insertNode(node, now)) {
@@ -164,14 +167,8 @@ Dht::reportedAddr(const SockAddr& addr)
 void
 Dht::onNewNode(const Sp<Node>& node, int confirm)
 {
-    const auto& now = scheduler.time();
-    auto& b = buckets(node->getFamily());
-    auto wasEmpty = confirm < 2 && b.grow_time < now - std::chrono::minutes(5);
-    if (b.onNewNode(node, confirm, now, myid, network_engine) or confirm) {
+    if (buckets(node->getFamily()).onNewNode(node, confirm, scheduler.time(), myid, network_engine) or confirm) {
         trySearchInsert(node);
-        if (wasEmpty) {
-            scheduler.edit(nextNodesConfirmation, now + std::chrono::seconds(1));
-        }
     }
 }
 
@@ -282,7 +279,7 @@ void Dht::paginate(std::weak_ptr<Search> ws, Sp<Query> query, SearchNode* n) {
                 if (vid == Value::INVALID_ID) continue;
                 auto query_for_vid = std::make_shared<Query>(Select {}, Where {}.id(vid));
                 sn->pagination_queries[query].push_back(query_for_vid);
-                DHT_LOG.d(id, sn->node->id, "[search %s] [node %s] sending %s",
+                DHT_LOG.w(id, sn->node->id, "[search %s] [node %s] sending %s",
                         id.toString().c_str(), sn->node->toString().c_str(), query_for_vid->toString().c_str());
                 sn->getStatus[query_for_vid] = network_engine.sendGetValues(status.node,
                         id,
@@ -301,7 +298,7 @@ void Dht::paginate(std::weak_ptr<Search> ws, Sp<Query> query, SearchNode* n) {
     /* add pagination query key for tracking ongoing requests. */
     n->pagination_queries[query].push_back(select_q);
 
-    DHT_LOG.d(sr->id, n->node->id, "[search %s] [node %s] sending %s",
+    DHT_LOG.w(sr->id, n->node->id, "[search %s] [node %s] sending %s",
             sr->id.toString().c_str(), n->node->toString().c_str(), select_q->toString().c_str());
     n->getStatus[select_q] = network_engine.sendGetValues(n->node,
             sr->id,
@@ -435,20 +432,24 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr) {
             auto next_refresh_time = now + getType(a.value->type).expiration;
             /* only put the value if the node doesn't already have it */
             if (not hasValue or seq_no < a.value->seq) {
-                DHT_LOG.d(sr->id, sn->node->id, "[search %s] [node %s] sending 'put' (vid: %d)",
+                DHT_LOG.w(sr->id, sn->node->id, "[search %s] [node %s] sending 'put' (vid: %d)",
                         sr->id.toString().c_str(), sn->node->toString().c_str(), a.value->id);
-                auto created = a.permanent ? time_point::max() : a.created;
-                sn->acked[a.value->id] = {
-                    network_engine.sendAnnounceValue(sn->node, sr->id, a.value, created, sn->token, onDone, onExpired),
-                    next_refresh_time
-                };
+                sn->acked[a.value->id] = std::make_pair(network_engine.sendAnnounceValue(sn->node,
+                                                sr->id,
+                                                a.value,
+                                                a.permanent ? time_point::max() : a.created,
+                                                sn->token,
+                                                onDone,
+                                                onExpired), next_refresh_time);
             } else if (hasValue and a.permanent) {
                 DHT_LOG.w(sr->id, sn->node->id, "[search %s] [node %s] sending 'refresh' (vid: %d)",
                         sr->id.toString().c_str(), sn->node->toString().c_str(), a.value->id);
-                sn->acked[a.value->id] = {
-                    network_engine.sendRefreshValue(sn->node, sr->id, a.value->id, sn->token, onDone, onExpired),
-                    next_refresh_time
-                };
+                sn->acked[a.value->id] = std::make_pair(network_engine.sendRefreshValue(sn->node,
+                                                sr->id,
+                                                a.value->id,
+                                                sn->token,
+                                                onDone,
+                                                onExpired), next_refresh_time);
             } else {
                 DHT_LOG.w(sr->id, sn->node->id, "[search %s] [node %s] already has value (vid: %d). Aborting.",
                         sr->id.toString().c_str(), sn->node->toString().c_str(), a.value->id);
@@ -499,7 +500,7 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr) {
         if (sendQuery) {
             if (not probe_query)
                 probe_query = std::make_shared<Query>(Select {}.field(Value::Field::Id).field(Value::Field::SeqNum));
-            DHT_LOG.d(sr->id, n.node->id, "[search %s] [node %s] sending %s",
+            DHT_LOG.w(sr->id, n.node->id, "[search %s] [node %s] sending %s",
                     sr->id.toString().c_str(), n.node->toString().c_str(), probe_query->toString().c_str());
             n.probe_query = probe_query;
             n.getStatus[probe_query] = network_engine.sendGetValues(n.node,
@@ -523,8 +524,8 @@ Dht::searchSynchedNodeListen(const Sp<Search>& sr, SearchNode& n)
         auto list_token = l.first;
         if (n.getListenTime(query) > scheduler.time())
             continue;
-        // DHT_LOG.d(sr->id, n.node->id, "[search %s] [node %s] sending 'listen'",
-        //        sr->id.toString().c_str(), n.node->toString().c_str());
+        DHT_LOG.w(sr->id, n.node->id, "[search %s] [node %s] sending 'listen'",
+                sr->id.toString().c_str(), n.node->toString().c_str());
 
         auto r = n.listenStatus.find(query);
         if (r == n.listenStatus.end()) {
@@ -536,13 +537,6 @@ Dht::searchSynchedNodeListen(const Sp<Search>& sr, SearchNode& n)
                             l->second.get_cb(l->second.filter.filter(values), expired);
                         }
                     }
-                }, [ws,list_token] (ListenSyncStatus status) {
-                    if (auto sr = ws.lock()) {
-                        auto l = sr->listeners.find(list_token);
-                        if (l != sr->listeners.end()) {
-                            l->second.sync_cb(status);
-                        }
-                    }
                 }
             }).first;
             auto node = n.node;
@@ -560,10 +554,8 @@ Dht::searchSynchedNodeListen(const Sp<Search>& sr, SearchNode& n)
             { /* on done */
                 if (auto sr = ws.lock()) {
                     scheduler.edit(sr->nextSearchStep, scheduler.time());
-                    if (auto sn = sr->getNode(req.node)) {
+                    if (auto sn = sr->getNode(req.node))
                         scheduler.add(sn->getListenTime(query), std::bind(&Dht::searchStep, this, sr));
-                        sn->onListenSynced(query);
-                    }
                     onListenDone(req.node, answer, sr);
                 }
             },
@@ -603,9 +595,9 @@ Dht::searchStep(Sp<Search> sr)
     if (not sr or sr->expired or sr->done) return;
 
     const auto& now = scheduler.time();
-    /*if (auto req_count = sr->currentlySolicitedNodeCount())
+    if (auto req_count = sr->currentlySolicitedNodeCount())
         DHT_LOG.d(sr->id, "[search %s IPv%c] step (%d requests)",
-                sr->id.toString().c_str(), sr->af == AF_INET ? '4' : '6', req_count);*/
+                sr->id.toString().c_str(), sr->af == AF_INET ? '4' : '6', req_count);
     sr->step_time = now;
 
     if (sr->refill_time + Node::NODE_EXPIRE_TIME < now and sr->nodes.size()-sr->getNumberOfBadNodes() < SEARCH_NODES)
@@ -664,7 +656,18 @@ Dht::searchStep(Sp<Search> sr)
             sr->setDone();
     }
 
-    while (sr->currentlySolicitedNodeCount() < MAX_REQUESTED_SEARCH_NODES and searchSendGetValues(sr));
+    if (sr->currentlySolicitedNodeCount() < MAX_REQUESTED_SEARCH_NODES) {
+        unsigned i = 0;
+        SearchNode* sent;
+        do {
+            sent = searchSendGetValues(sr);
+            if (sent and not sent->candidate)
+                i++;
+        }
+        while (sent and sr->currentlySolicitedNodeCount() < MAX_REQUESTED_SEARCH_NODES);
+        /*DHT_LOG_DBG("[search %s IPv%c] step: sent %u requests (total %u).",
+            sr->id.toString().c_str(), sr->af == AF_INET ? '4' : '6', i, sr->currentlySolicitedNodeCount());*/
+    }
 
     if (sr->getNumberOfConsecutiveBadNodes() >= std::min(sr->nodes.size(),
                                                              static_cast<size_t>(SEARCH_MAX_BAD_NODES)))
@@ -675,6 +678,10 @@ Dht::searchStep(Sp<Search> sr)
     }
 
     /* dumpSearch(*sr, std::cout); */
+
+    /* periodic searchStep scheduling. */
+    //if (not sr->done)
+    //    scheduler.edit(sr->nextSearchStep, now);
 }
 
 unsigned Dht::refill(Dht::Search& sr) {
@@ -836,16 +843,14 @@ Dht::listen(const InfoHash& id, ValueCallback cb, Value::Filter f, Where where)
     auto vals = std::make_shared<std::map<Value::Id, Sp<Value>>>();
     auto token = ++listener_token;
 
-    auto gcb = OpValueCache::cacheCallback(std::move(cb), [this, id, token]{
-        cancelListen(id, token);
-    });
+    auto gcb = OpValueCache::cacheCallback(std::move(cb));
 
     auto query = std::make_shared<Query>(q);
     auto filter = f.chain(q.where.getFilter());
     size_t tokenlocal = 0;
     auto st = store.find(id);
     if (st == store.end() && store.size() < MAX_HASHES)
-        st = store.emplace(id, scheduler.time() + MAX_STORAGE_MAINTENANCE_EXPIRE_TIME).first;
+        st = store.emplace(id, Storage(scheduler.time() + MAX_STORAGE_MAINTENANCE_EXPIRE_TIME)).first;
     if (st != store.end()) {
         if (not st->second.empty()) {
             std::vector<Sp<Value>> newvals = st->second.get(filter);
@@ -864,12 +869,9 @@ Dht::listen(const InfoHash& id, ValueCallback cb, Value::Filter f, Where where)
     }
 
     auto token4 = Dht::listenTo(id, AF_INET, gcb, filter, query);
-    auto token6 = token4 == 0 ? 0 : Dht::listenTo(id, AF_INET6, gcb, filter, query);
-    if (token6 == 0 && st != store.end()) {
-        st->second.local_listeners.erase(tokenlocal);
-        return 0;
-    }
+    auto token6 = Dht::listenTo(id, AF_INET6, gcb, filter, query);
 
+    DHT_LOG.d(id, "Added listen : %d -> %d %d %d", token, tokenlocal, token4, token6);
     listeners.emplace(token, std::make_tuple(tokenlocal, token4, token6));
     return token;
 }
@@ -914,7 +916,7 @@ struct OpStatus {
 
 template <typename T>
 struct GetStatus : public OpStatus {
-    T values;
+    std::vector<Sp<T>> values;
     std::vector<Sp<Node>> nodes;
 };
 
@@ -973,16 +975,20 @@ void doneCallbackWrapper(DoneCallback dcb, const std::vector<Sp<Node>>& nodes, G
     }
 }
 
-template <typename T, typename St, typename Cb, typename Av, typename Cv>
-bool callbackWrapper(Cb get_cb, DoneCallback done_cb, const std::vector<Sp<T>>& values,
-    Av add_values, Cv cache_values, GetStatus<St>& op)
+template <typename T, typename Cb>
+bool callbackWrapper(Cb get_cb,
+        DoneCallback done_cb,
+        const std::vector<Sp<T>>& values,
+        std::function<std::vector<Sp<T>>(const std::vector<Sp<T>>&)> add_values,
+        Sp<GetStatus<T>> o)
 {
+    auto& op = *o;
     if (op.status.done)
         return false;
     auto newvals = add_values(values);
     if (not newvals.empty()) {
         op.status.ok = !get_cb(newvals);
-        cache_values(newvals);
+        op.values.insert(op.values.end(), newvals.begin(), newvals.end());
     }
     doneCallbackWrapper(done_cb, {}, op);
     return !op.status.ok;
@@ -994,26 +1000,23 @@ Dht::get(const InfoHash& id, GetCallback getcb, DoneCallback donecb, Value::Filt
     scheduler.syncTime();
 
     auto q = std::make_shared<Query>(Select {}, std::move(where));
-    auto op = std::make_shared<GetStatus<std::map<Value::Id, Sp<Value>>>>();
+    auto op = std::make_shared<GetStatus<Value>>();
     auto f = filter.chain(q->where.getFilter());
 
-    auto gcb = [getcb, donecb, op, f](const std::vector<Sp<Value>>& vals) {
-        auto& o = *op;
-        return callbackWrapper(getcb, donecb, vals, [&o,&f](const std::vector<Sp<Value>>& values) {
-            std::vector<Sp<Value>> newvals {};
-            for (const auto& v : values) {
-                auto it = o.values.find(v->id);
-                if (it == o.values.cend() or (it->second != v && !(*it->second == *v))) {
-                    if (not f or f(*v))
-                        newvals.push_back(v);
-                }
+    auto add_values = [op,f](const std::vector<Sp<Value>>& values) {
+        std::vector<Sp<Value>> newvals {};
+        for (const auto& v : values) {
+            auto it = std::find_if(op->values.cbegin(), op->values.cend(), [&](const Sp<Value>& sv) {
+                return sv == v or *sv == *v;
+            });
+            if (it == op->values.cend()) {
+               if (not f or f(*v))
+                   newvals.push_back(v);
             }
-            return newvals;
-        }, [&o](const std::vector<Sp<Value>>& newvals) {
-            for (const auto& v : newvals)
-                o.values[v->id] = v;
-        }, o);
+        }
+        return newvals;
     };
+    auto gcb = std::bind(callbackWrapper<Value, GetCallback>, getcb, donecb, _1, add_values, op);
 
     /* Try to answer this search locally. */
     gcb(getLocal(id, f));
@@ -1033,39 +1036,36 @@ Dht::get(const InfoHash& id, GetCallback getcb, DoneCallback donecb, Value::Filt
 void Dht::query(const InfoHash& id, QueryCallback cb, DoneCallback done_cb, Query&& q)
 {
     scheduler.syncTime();
-    auto op = std::make_shared<GetStatus<std::vector<Sp<FieldValueIndex>>>>();
-    auto f = q.where.getFilter();
-    auto qcb = [cb, done_cb, op](const std::vector<Sp<FieldValueIndex>>& fields){
-        auto& o = *op;
-        return callbackWrapper(cb, done_cb, fields, [&](const std::vector<Sp<FieldValueIndex>>& fields) {
-            std::vector<Sp<FieldValueIndex>> newvals {};
-            for (const auto& f : fields) {
-                auto it = std::find_if(o.values.cbegin(), o.values.cend(),
-                    [&](const Sp<FieldValueIndex>& sf) {
-                        return sf == f or f->containedIn(*sf);
-                    });
-                if (it == o.values.cend()) {
-                    auto lesser = std::find_if(o.values.begin(), o.values.end(),
-                        [&](const Sp<FieldValueIndex>& sf) {
-                            return sf->containedIn(*f);
-                        });
-                    if (lesser != o.values.end())
-                        o.values.erase(lesser);
-                    newvals.push_back(f);
-                }
-            }
-            return newvals;
-        }, [&](const std::vector<Sp<FieldValueIndex>>& fields){
-            o.values.insert(o.values.end(), fields.begin(), fields.end());
-        }, o);
-    };
+    auto op = std::make_shared<GetStatus<FieldValueIndex>>();
 
-    /* Try to answer this search locally. */
+    auto f = q.where.getFilter();
     auto values = getLocal(id, f);
+    auto add_fields = [=](const std::vector<Sp<FieldValueIndex>>& fields) {
+        std::vector<Sp<FieldValueIndex>> newvals {};
+        for (const auto& f : fields) {
+            auto it = std::find_if(op->values.cbegin(), op->values.cend(),
+                [&](const Sp<FieldValueIndex>& sf) {
+                    return sf == f or f->containedIn(*sf);
+                });
+            if (it == op->values.cend()) {
+                auto lesser = std::find_if(op->values.begin(), op->values.end(),
+                    [&](const Sp<FieldValueIndex>& sf) {
+                        return sf->containedIn(*f);
+                    });
+                if (lesser != op->values.end())
+                    op->values.erase(lesser);
+                newvals.push_back(f);
+            }
+        }
+        return newvals;
+    };
     std::vector<Sp<FieldValueIndex>> local_fields(values.size());
     std::transform(values.begin(), values.end(), local_fields.begin(), [&q](const Sp<Value>& v) {
         return std::make_shared<FieldValueIndex>(*v, q.select);
     });
+    auto qcb = std::bind(callbackWrapper<FieldValueIndex, QueryCallback>, cb, done_cb, _1, add_fields, op);
+
+    /* Try to answer this search locally. */
     qcb(local_fields);
 
     auto sq = std::make_shared<Query>(std::move(q));
@@ -1173,7 +1173,6 @@ Dht::storageChanged(const InfoHash& id, Storage& st, ValueStorage& v, bool newVa
         if (not st.local_listeners.empty()) {
             DHT_LOG.d(id, "[store %s] %lu local listeners", id.toString().c_str(), st.local_listeners.size());
             std::vector<std::pair<ValueCallback, std::vector<Sp<Value>>>> cbs;
-            cbs.reserve(st.local_listeners.size());
             for (const auto& l : st.local_listeners) {
                 std::vector<Sp<Value>> vals;
                 if (not l.second.filter or l.second.filter(*v.data))
@@ -1224,7 +1223,7 @@ Dht::storageStore(const InfoHash& id, const Sp<Value>& value, time_point created
     if (st == store.end()) {
         if (store.size() >= MAX_HASHES)
             return false;
-        auto st_i = store.emplace(id, now);
+        auto st_i = store.emplace(id, Storage(now));
         st = st_i.first;
         if (maintain_storage and st_i.second)
             scheduler.add(st->second.maintenance_time, std::bind(&Dht::dataPersistence, this, id));
@@ -1270,7 +1269,7 @@ Dht::storageAddListener(const InfoHash& id, const Sp<Node>& node, size_t socket_
     if (st == store.end()) {
         if (store.size() >= MAX_HASHES)
             return;
-        st = store.emplace(id, now).first;
+        st = store.emplace(id, Storage(now)).first;
     }
     auto node_listeners = st->second.listeners.emplace(node, std::map<size_t, Listener> {}).first;
     auto l = node_listeners->second.find(socket_id);
@@ -1389,6 +1388,12 @@ Dht::connectivityChanged(sa_family_t af)
     scheduler.edit(nextNodesConfirmation, now);
     buckets(af).connectivityChanged(now);
     network_engine.connectivityChanged(af);
+    for (auto& sp : searches(af))
+        for (auto& sn : sp.second->nodes) {
+            for (auto& ls : sn.listenStatus)
+                sn.node->cancelRequest(ls.second.req);
+            sn.listenStatus.clear();
+        }
     reported_addr.erase(std::remove_if(reported_addr.begin(), reported_addr.end(), [&](const ReportedAddr& addr){
         return addr.second.getFamily() == af;
     }), reported_addr.end());
@@ -1800,7 +1805,7 @@ Dht::neighbourhoodMaintenance(RoutingTable& list)
                 n->toString().c_str(), id.toString().c_str());
         /* Since our node-id is the same in both DHTs, it's probably
            profitable to query both families. */
-        network_engine.sendFindNode(n, id, network_engine.want());
+        network_engine.sendFindNode(n, id, network_engine.want(), nullptr, nullptr);
     }
 
     return true;
@@ -2283,10 +2288,12 @@ Dht::onListen(Sp<Node> node, const InfoHash& hash, const Blob& token, size_t soc
 }
 
 void
-Dht::onListenDone(const Sp<Node>& node, net::RequestAnswer& answer, Sp<Search>& sr)
+Dht::onListenDone(const Sp<Node>& node,
+        net::RequestAnswer& answer,
+        Sp<Search>& sr)
 {
-    // DHT_LOG.d(sr->id, node->id, "[search %s] [node %s] got listen confirmation",
-    //            sr->id.toString().c_str(), node->toString().c_str(), answer.values.size());
+    DHT_LOG.d(sr->id, node->id, "[search %s] [node %s] got listen confirmation",
+                sr->id.toString().c_str(), node->toString().c_str(), answer.values.size());
 
     if (not sr->done) {
         const auto& now = scheduler.time();
@@ -2354,7 +2361,7 @@ Dht::onAnnounce(Sp<Node> n,
             // Allow the value to be edited by the storage policy
             const auto& type = getType(vc->type);
             if (type.storePolicy(hash, vc, node.id, node.getAddr())) {
-                //DHT_LOG.d(hash, node.id, "[store %s] storing %s", hash.toString().c_str(), std::to_string(vc->id).c_str());
+                DHT_LOG.d(hash, node.id, "[store %s] storing %s", hash.toString().c_str(), vc->toString().c_str());
                 storageStore(hash, vc, created, node.getAddr());
             } else {
                 DHT_LOG.d(hash, node.id, "[store %s] rejecting storage of %s",
@@ -2390,23 +2397,6 @@ Dht::storageRefresh(const InfoHash& id, Value::Id vid)
     const auto& now = scheduler.time();
     auto s = store.find(id);
     if (s != store.end()) {
-        // Values like for a permanent put can be refreshed. So, inform remote listeners that the value
-        // need to be refreshed
-        auto& st = s->second;
-        if (not st.listeners.empty()) {
-            DHT_LOG.d(id, "[store %s] %lu remote listeners", id.toString().c_str(), st.listeners.size());
-            std::vector<Value::Id> ids = {vid};
-            for (const auto& node_listeners : st.listeners) {
-                for (const auto& l : node_listeners.second) {
-                    DHT_LOG.w(id, node_listeners.first->id, "[store %s] [node %s] sending refresh",
-                            id.toString().c_str(),
-                            node_listeners.first->toString().c_str());
-                    Blob ntoken = makeToken(node_listeners.first->getAddr(), false);
-                    network_engine.tellListenerRefreshed(node_listeners.first, l.first, id, ntoken, ids);
-                }
-            }
-        }
-
         auto expiration = s->second.refresh(now, vid, types);
         if (expiration != time_point::max())
             scheduler.add(expiration, std::bind(&Dht::expireStorage, this, id));

src/dht_proxy_client.cpp

@@ -53,6 +53,7 @@ struct DhtProxyClient::Listener
     std::thread thread;
     unsigned callbackId;
     Sp<ListenState> state;
+    Sp<proxy::ListenToken> pushNotifToken;
     Sp<Scheduler::Job> refreshJob;
     Listener(ValueCache&& c, Sp<Scheduler::Job>&& j, const Sp<restbed::Request>& r, Value::Filter&& f)
         : cache(std::move(c)),
@@ -80,8 +81,6 @@ DhtProxyClient::DhtProxyClient() {}
 DhtProxyClient::DhtProxyClient(std::function<void()> signal, const std::string& serverHost, const std::string& pushClientId)
 : serverHost_(serverHost), pushClientId_(pushClientId), loopSignal_(signal)
 {
-    if (serverHost_.find("://") == std::string::npos)
-        serverHost_ = proxy::HTTP_PROTO + serverHost_;
     if (!serverHost_.empty())
         startProxy();
 }
@@ -99,7 +98,6 @@ DhtProxyClient::startProxy()
     if (serverHost_.empty()) return;
     DHT_LOG.w("Staring proxy client to %s", serverHost_.c_str());
     nextProxyConfirmation = scheduler.add(scheduler.time(), std::bind(&DhtProxyClient::confirmProxy, this));
-    listenerRestart = std::make_shared<Scheduler::Job>(std::bind(&DhtProxyClient::restartListeners, this));
     loopSignal_();
 }
 
@@ -245,7 +243,7 @@ void
 DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Value::Filter&& f, Where&& w)
 {
     DHT_LOG.d(key, "[search %s]: get", key.to_c_str());
-    restbed::Uri uri(serverHost_ + "/" + key.toString());
+    restbed::Uri uri(proxy::HTTP_PROTO + serverHost_ + "/" + key.toString());
     auto req = std::make_shared<restbed::Request>(uri);
     Value::Filter filter = w.empty() ? f : f.chain(w.getFilter());
 
@@ -258,49 +256,45 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Va
         // will change the connectivity status
         struct GetState{ std::atomic_bool ok {true}; std::atomic_bool stop {false}; };
         auto state = std::make_shared<GetState>();
-        try {
-            restbed::Http::async(req,
-                [=](const std::shared_ptr<restbed::Request>& req,
-                    const std::shared_ptr<restbed::Response>& reply) {
-                auto code = reply->get_status_code();
+        restbed::Http::async(req,
+            [=](const std::shared_ptr<restbed::Request>& req,
+                const std::shared_ptr<restbed::Response>& reply) {
+            auto code = reply->get_status_code();
 
-                if (code == 200) {
-                    try {
-                        while (restbed::Http::is_open(req) and not *finished and not state->stop) {
-                            restbed::Http::fetch("\n", reply);
-                            if (*finished or state->stop)
-                                break;
-                            std::string body;
-                            reply->get_body(body);
-                            reply->set_body(""); // Reset the body for the next fetch
+            if (code == 200) {
+                try {
+                    while (restbed::Http::is_open(req) and not *finished and not state->stop) {
+                        restbed::Http::fetch("\n", reply);
+                        if (*finished or state->stop)
+                            break;
+                        std::string body;
+                        reply->get_body(body);
+                        reply->set_body(""); // Reset the body for the next fetch
 
-                            std::string err;
-                            Json::Value json;
-                            Json::CharReaderBuilder rbuilder;
-                            auto* char_data = reinterpret_cast<const char*>(&body[0]);
-                            auto reader = std::unique_ptr<Json::CharReader>(rbuilder.newCharReader());
-                            if (reader->parse(char_data, char_data + body.size(), &json, &err)) {
-                                auto value = std::make_shared<Value>(json);
-                                if ((not filter or filter(*value)) and cb) {
-                                    std::lock_guard<std::mutex> lock(lockCallbacks);
-                                    callbacks_.emplace_back([cb, value, state]() {
-                                        if (not state->stop and not cb({value}))
-                                            state->stop = true;
-                                    });
-                                    loopSignal_();
-                                }
-                            } else {
-                                state->ok = false;
+                        std::string err;
+                        Json::Value json;
+                        Json::CharReaderBuilder rbuilder;
+                        auto* char_data = reinterpret_cast<const char*>(&body[0]);
+                        auto reader = std::unique_ptr<Json::CharReader>(rbuilder.newCharReader());
+                        if (reader->parse(char_data, char_data + body.size(), &json, &err)) {
+                            auto value = std::make_shared<Value>(json);
+                            if ((not filter or filter(*value)) and cb) {
+                                std::lock_guard<std::mutex> lock(lockCallbacks);
+                                callbacks_.emplace_back([cb, value, state]() {
+                                    if (not state->stop and not cb({value}))
+                                        state->stop = true;
+                                });
+                                loopSignal_();
                             }
+                        } else {
+                            state->ok = false;
                         }
-                    } catch (std::runtime_error& e) { }
-                } else {
-                    state->ok = false;
-                }
-            }).wait();
-        } catch(const std::exception& e) {
-            state->ok = false;
-        }
+                    }
+                } catch (std::runtime_error& e) { }
+            } else {
+                state->ok = false;
+            }
+        }).wait();
         if (donecb) {
             std::lock_guard<std::mutex> lock(lockCallbacks);
             callbacks_.emplace_back([=](){
@@ -364,7 +358,7 @@ void
 DhtProxyClient::doPut(const InfoHash& key, Sp<Value> val, DoneCallback cb, time_point /*created*/, bool permanent)
 {
     DHT_LOG.d(key, "[search %s] performing put of %s", key.to_c_str(), val->toString().c_str());
-    restbed::Uri uri(serverHost_ + "/" + key.toString());
+    restbed::Uri uri(proxy::HTTP_PROTO + serverHost_ + "/" + key.toString());
     auto req = std::make_shared<restbed::Request>(uri);
     req->set_method("POST");
 
@@ -395,36 +389,32 @@ DhtProxyClient::doPut(const InfoHash& key, Sp<Value> val, DoneCallback cb, time_
     o.finished = finished;
     o.thread = std::thread([=](){
         auto ok = std::make_shared<std::atomic_bool>(true);
-        try {
-            restbed::Http::async(req,
-                [ok](const std::shared_ptr<restbed::Request>& /*req*/,
-                            const std::shared_ptr<restbed::Response>& reply) {
-                auto code = reply->get_status_code();
+        restbed::Http::async(req,
+            [ok](const std::shared_ptr<restbed::Request>& /*req*/,
+                        const std::shared_ptr<restbed::Response>& reply) {
+            auto code = reply->get_status_code();
 
-                if (code == 200) {
-                    restbed::Http::fetch("\n", reply);
-                    std::string body;
-                    reply->get_body(body);
-                    reply->set_body(""); // Reset the body for the next fetch
+            if (code == 200) {
+                restbed::Http::fetch("\n", reply);
+                std::string body;
+                reply->get_body(body);
+                reply->set_body(""); // Reset the body for the next fetch
 
-                    try {
-                        std::string err;
-                        Json::Value json;
-                        Json::CharReaderBuilder rbuilder;
-                        auto* char_data = reinterpret_cast<const char*>(&body[0]);
-                        auto reader = std::unique_ptr<Json::CharReader>(rbuilder.newCharReader());
-                        if (not reader->parse(char_data, char_data + body.size(), &json, &err))
-                            *ok = false;
-                    } catch (...) {
+                try {
+                    std::string err;
+                    Json::Value json;
+                    Json::CharReaderBuilder rbuilder;
+                    auto* char_data = reinterpret_cast<const char*>(&body[0]);
+                    auto reader = std::unique_ptr<Json::CharReader>(rbuilder.newCharReader());
+                    if (not reader->parse(char_data, char_data + body.size(), &json, &err))
                         *ok = false;
-                    }
-                } else {
+                } catch (...) {
                     *ok = false;
                 }
-            }).wait();
-        } catch(const std::exception& e) {
-            *ok = false;
-        }
+            } else {
+                *ok = false;
+            }
+        }).wait();
         if (cb) {
             std::lock_guard<std::mutex> lock(lockCallbacks);
             callbacks_.emplace_back([=](){
@@ -527,14 +517,7 @@ DhtProxyClient::getProxyInfos()
     }
     statusThread_ = std::thread([this, serverHost, infoState]{
         try {
-            auto endpointStr = serverHost;
-            auto protocol = std::string(proxy::HTTP_PROTO);
-            auto protocolIdx = serverHost.find("://");
-            if (protocolIdx != std::string::npos) {
-                protocol = endpointStr.substr(0, protocolIdx + 3);
-                endpointStr = endpointStr.substr(protocolIdx + 3);
-            }
-            auto hostAndService = splitPort(endpointStr);
+            auto hostAndService = splitPort(serverHost);
             auto resolved_proxies = SockAddr::resolve(hostAndService.first, hostAndService.second);
             std::vector<std::future<Sp<restbed::Response>>> reqs;
             reqs.reserve(resolved_proxies.size());
@@ -543,9 +526,9 @@ DhtProxyClient::getProxyInfos()
                 if (resolved_proxy.getFamily() == AF_INET6) {
                     // HACK restbed seems to not correctly handle directly http://[ipv6]
                     // See https://github.com/Corvusoft/restbed/issues/290.
-                    server = endpointStr;
+                    server = serverHost;
                 }
-                restbed::Uri uri(protocol + server + "/");
+                restbed::Uri uri(proxy::HTTP_PROTO + server + "/");
                 auto req = std::make_shared<restbed::Request>(uri);
                 if (infoState->cancel)
                     return;
@@ -630,7 +613,7 @@ DhtProxyClient::onProxyInfos(const Json::Value& proxyInfos, sa_family_t family)
     auto newStatus = std::max(statusIpv4_, statusIpv6_);
     if (newStatus == NodeStatus::Connected) {
         if (oldStatus == NodeStatus::Disconnected || oldStatus == NodeStatus::Connecting) {
-            scheduler.edit(listenerRestart, scheduler.time());
+            restartListeners();
         }
         scheduler.edit(nextProxyConfirmation, scheduler.time() + std::chrono::minutes(15));
     }
@@ -668,93 +651,8 @@ DhtProxyClient::listen(const InfoHash& key, ValueCallback cb, Value::Filter filt
         it = searches_.emplace(key, ProxySearch{}).first;
     }
     auto query = std::make_shared<Query>(Select{}, where);
-    auto token = it->second.ops.listen(cb, query, filter, [this, key, filter](Sp<Query> /*q*/, ValueCallback cb, SyncCallback scb) -> size_t {
-        scheduler.syncTime();
-        restbed::Uri uri(serverHost_ + "/" + key.toString());
-        std::lock_guard<std::mutex> lock(searchLock_);
-        auto search = searches_.find(key);
-        if (search == searches_.end()) {
-            DHT_LOG.e(key, "[search %s] listen: search not found", key.to_c_str());
-            return 0;
-        }
-        DHT_LOG.d(key, "[search %s] sending %s", key.to_c_str(), deviceKey_.empty() ? "listen" : "subscribe");
-
-        auto req = std::make_shared<restbed::Request>(uri);
-        auto token = ++listenerToken_;
-        auto l = search->second.listeners.find(token);
-        if (l == search->second.listeners.end()) {
-            auto f = filter;
-            l = search->second.listeners.emplace(token, Listener {
-                ValueCache(cb, std::move(scb)), scheduler.add(time_point::max(), [this, key, token]{
-                    std::lock_guard<std::mutex> lock(searchLock_);
-                    auto s = searches_.find(key);
-                    if (s == searches_.end()) {
-                        return;
-                    }
-                    auto l = s->second.listeners.find(token);
-                    if (l == s->second.listeners.end()) {
-                        return;
-                    }
-                    auto next = l->second.cache.expireValues(scheduler.time());
-                    scheduler.edit(l->second.cacheExpirationJob, next);
-                }), req, std::move(f)
-            }).first;
-        } else {
-            if (l->second.state)
-                l->second.state->cancel = true;
-        }
-
-        auto state = std::make_shared<ListenState>();
-        l->second.state = state;
-        l->second.cb = [this,key,token,state](const std::vector<Sp<Value>>& values, bool expired) {
-            if (state->cancel)
-                return false;
-            std::lock_guard<std::mutex> lock(searchLock_);
-            auto s = searches_.find(key);
-            if (s == searches_.end()) {
-                return false;
-            }
-            auto l = s->second.listeners.find(token);
-            if (l == s->second.listeners.end()) {
-                return false;
-            }
-            const std::vector<Sp<Value>> new_values_empty;
-            std::vector<Value::Id> expired_ids;
-            if (expired) {
-                expired_ids.reserve(values.size());
-                for (const auto& v : values)
-                    expired_ids.emplace_back(v->id);
-            }
-            auto next = l->second.cache.onValues(expired ? new_values_empty : values, std::vector<Value::Id>{}, expired_ids, types, scheduler.time());
-            scheduler.edit(l->second.cacheExpirationJob, next);
-            loopSignal_();
-            return true;
-        };
-
-        auto vcb = l->second.cb;
-        l->second.req = req;
-
-        if (not deviceKey_.empty()) {
-            // Relaunch push listeners even if a timeout is not received (if the proxy crash for any reason)
-            l->second.refreshJob = scheduler.add(scheduler.time() + proxy::OP_TIMEOUT - proxy::OP_MARGIN, [this, key, token, state] {
-                if (state->cancel)
-                    return;
-                std::lock_guard<std::mutex> lock(searchLock_);
-                auto s = searches_.find(key);
-                if (s != searches_.end()) {
-                    auto l = s->second.listeners.find(token);
-                    if (l != s->second.listeners.end()) {
-                        resubscribe(key, l->second);
-                    }
-                }
-            });
-        }
-        l->second.thread = std::thread([this, req, vcb, filter, state]() {
-            sendListen(req, vcb, filter, state,
-                    deviceKey_.empty() ? ListenMethod::LISTEN
-                                    : ListenMethod::SUBSCRIBE);
-        });
-        return token;
+    auto token = it->second.ops.listen(cb, query, filter, [&](Sp<Query> /*q*/, ValueCallback vcb){
+        return doListen(key, vcb, filter);
     });
     return token;
 }
@@ -786,78 +684,195 @@ DhtProxyClient::cancelListen(const InfoHash& key, size_t gtoken) {
     return canceled;
 }
 
-void DhtProxyClient::sendListen(const std::shared_ptr<restbed::Request> &req,
-                                const ValueCallback &cb,
-                                const Value::Filter &filter,
-                                const Sp<ListenState> &state,
-                                ListenMethod method) {
-  auto settings = std::make_shared<restbed::Settings>();
-  if (method != ListenMethod::LISTEN) {
-    req->set_method("SUBSCRIBE");
-  } else {
-    std::chrono::milliseconds timeout(std::numeric_limits<int>::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 (method != ListenMethod::LISTEN)
-      fillBody(req, method == ListenMethod::RESUBSCRIBE);
-#endif
-        restbed::Http::async(req,
-            [this, filter, cb, state](const std::shared_ptr<restbed::Request>& req,
-                                    const std::shared_ptr<restbed::Response>& reply) {
-            auto code = reply->get_status_code();
-            if (code == 200) {
-                try {
-                    while (restbed::Http::is_open(req) and not state->cancel) {
-                        restbed::Http::fetch("\n", reply);
-                        if (state->cancel)
-                            break;
-                        std::string body;
-                        reply->get_body(body);
-                        reply->set_body(""); // Reset the body for the next fetch
 
-                        Json::Value json;
-                        std::string err;
-                        Json::CharReaderBuilder rbuilder;
-                        auto reader = std::unique_ptr<Json::CharReader>(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<Value>(json);
-                            if ((not filter or filter(*value)) and cb) {
-                                std::lock_guard<std::mutex> lock(lockCallbacks);
-                                callbacks_.emplace_back([cb, value, state, expired]() {
-                                    if (not state->cancel and not cb({value}, expired))
-                                        state->cancel = true;
-                                });
-                                loopSignal_();
-                            }
+void
+DhtProxyClient::sendListen(const std::shared_ptr<restbed::Request>& req, const ValueCallback& cb, const Value::Filter& filter, const Sp<ListenState>& state) {
+    auto settings = std::make_shared<restbed::Settings>();
+    std::chrono::milliseconds timeout(std::numeric_limits<int>::max());
+    settings->set_connection_timeout(timeout); // Avoid the client to close the socket after 5 seconds.
+    req->set_method("LISTEN");
+    restbed::Http::async(req,
+        [this, filter, cb, state](const std::shared_ptr<restbed::Request>& req,
+                                  const std::shared_ptr<restbed::Response>& reply) {
+        auto code = reply->get_status_code();
+        if (code == 200) {
+            try {
+                while (restbed::Http::is_open(req) and not state->cancel) {
+                    restbed::Http::fetch("\n", reply);
+                    if (state->cancel)
+                        break;
+                    std::string body;
+                    reply->get_body(body);
+                    reply->set_body(""); // Reset the body for the next fetch
+
+                    Json::Value json;
+                    std::string err;
+                    Json::CharReaderBuilder rbuilder;
+                    auto reader = std::unique_ptr<Json::CharReader>(rbuilder.newCharReader());
+                    if (reader->parse(body.data(), body.data() + body.size(), &json, &err)) {
+                        auto expired = json.get("expired", Json::Value(false)).asBool();
+                        auto value = std::make_shared<Value>(json);
+                        if ((not filter or filter(*value)) and cb)  {
+                            std::lock_guard<std::mutex> 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.w("Listen closed by the proxy server: %s", e.what());
-                        state->ok = false;
-                    }
                 }
-            } else {
-                state->ok = false;
+            } catch (const std::exception& e) {
+                if (not state->cancel) {
+                    DHT_LOG.w("Listen closed by the proxy server: %s", e.what());
+                    state->ok = false;
+                }
             }
-        }, settings).get();
-    } catch (const std::exception& e) {
-        state->ok = false;
-    }
+        } else {
+            state->ok = false;
+        }
+    }, settings).get();
     auto& s = *state;
     if (not s.ok and not s.cancel)
         opFailed();
 }
 
+void
+DhtProxyClient::sendSubscribe(const std::shared_ptr<restbed::Request>& req, const Sp<proxy::ListenToken>& token, const Sp<ListenState>& state) {
+#if OPENDHT_PUSH_NOTIFICATIONS
+    req->set_method("SUBSCRIBE");
+    fillBodyToGetToken(req);
+    restbed::Http::async(req, [this,state, token](const std::shared_ptr<restbed::Request>&,
+                                             const std::shared_ptr<restbed::Response>& reply) {
+        auto code = reply->get_status_code();
+        if (code == 200) {
+            try {
+                restbed::Http::fetch("\n", reply);
+                std::string body;
+                reply->get_body(body);
+
+                std::string err;
+                Json::Value json;
+                Json::CharReaderBuilder rbuilder;
+                auto reader = std::unique_ptr<Json::CharReader>(rbuilder.newCharReader());
+                if (reader->parse(body.data(), body.data() + body.size(), &json, &err)) {
+                    if (not json.isMember("token")) {
+                        state->ok = false;
+                        return;
+                    }
+                    *token = unpackId(json, "token");
+                }
+            } 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();
+    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*/)
+{
+    scheduler.syncTime();
+    restbed::Uri uri(proxy::HTTP_PROTO + serverHost_ + "/" + key.toString());
+    std::lock_guard<std::mutex> lock(searchLock_);
+    auto search = searches_.find(key);
+    if (search == searches_.end()) {
+        DHT_LOG.e(key, "[search %s] listen: search not found", key.to_c_str());
+        return 0;
+    }
+    DHT_LOG.d(key, "[search %s] sending %s", key.to_c_str(), deviceKey_.empty() ? "listen" : "subscribe");
+
+    auto req = std::make_shared<restbed::Request>(uri);
+    auto token = ++listenerToken_;
+    auto l = search->second.listeners.find(token);
+    if (l == search->second.listeners.end()) {
+        auto f = filter;
+        l = search->second.listeners.emplace(token, Listener {
+            ValueCache(cb), scheduler.add(time_point::max(), [this, key, token]{
+                std::lock_guard<std::mutex> lock(searchLock_);
+                auto s = searches_.find(key);
+                if (s == searches_.end()) {
+                    return;
+                }
+                auto l = s->second.listeners.find(token);
+                if (l == s->second.listeners.end()) {
+                    return;
+                }
+                auto next = l->second.cache.expireValues(scheduler.time());
+                scheduler.edit(l->second.cacheExpirationJob, next);
+            }), req, std::move(f)
+        }).first;
+    } else {
+        if (l->second.state)
+            l->second.state->cancel = true;
+    }
+
+    auto state = std::make_shared<ListenState>();
+    l->second.state = state;
+    l->second.cb = [this,key,token,state](const std::vector<Sp<Value>>& values, bool expired) {
+        if (state->cancel)
+            return false;
+        std::lock_guard<std::mutex> lock(searchLock_);
+        auto s = searches_.find(key);
+        if (s == searches_.end()) {
+            return false;
+        }
+        auto l = s->second.listeners.find(token);
+        if (l == s->second.listeners.end()) {
+            return false;
+        }
+        const std::vector<Sp<Value>> new_values_empty;
+        std::vector<Value::Id> expired_ids;
+        if (expired) {
+            expired_ids.reserve(values.size());
+            for (const auto& v : values)
+                expired_ids.emplace_back(v->id);
+        }
+        auto next = l->second.cache.onValues(expired ? new_values_empty : values, std::vector<Value::Id>{}, expired_ids, types, scheduler.time());
+        scheduler.edit(l->second.cacheExpirationJob, next);
+        loopSignal_();
+        return true;
+    };
+
+    auto pushNotifToken = std::make_shared<proxy::ListenToken>(0);
+    auto vcb = l->second.cb;
+    l->second.pushNotifToken = pushNotifToken;
+    l->second.req = req;
+
+    if (not deviceKey_.empty()) {
+        // Relaunch push listeners even if a timeout is not received (if the proxy crash for any reason)
+        l->second.refreshJob = scheduler.add(scheduler.time() + proxy::OP_TIMEOUT - proxy::OP_MARGIN, [this, key, token, state] {
+            if (state->cancel)
+                return;
+            std::lock_guard<std::mutex> lock(searchLock_);
+            auto s = searches_.find(key);
+            if (s != searches_.end()) {
+                auto l = s->second.listeners.find(token);
+                if (l != s->second.listeners.end()) {
+                    resubscribe(key, l->second);
+                }
+            }
+        });
+        l->second.thread = std::thread([this, req,pushNotifToken, state](){
+            sendSubscribe(req,pushNotifToken,state);
+        });
+    } else {
+        l->second.thread = std::thread([this, req, vcb, filter, state]{
+            sendListen(req,vcb,filter,state);
+        });
+    }
+    return token;
+}
+
 bool
 DhtProxyClient::doCancelListen(const InfoHash& key, size_t ltoken)
 {
@@ -881,13 +896,14 @@ DhtProxyClient::doCancelListen(const InfoHash& key, size_t ltoken)
             listener.thread.join();
         }
         // UNSUBSCRIBE
-        restbed::Uri uri(serverHost_ + "/" + key.toString());
+        restbed::Uri uri(proxy::HTTP_PROTO + serverHost_ + "/" + key.toString());
         auto req = std::make_shared<restbed::Request>(uri);
         req->set_method("UNSUBSCRIBE");
         // fill request body
         Json::Value body;
         body["key"] = deviceKey_;
         body["client_id"] = pushClientId_;
+        body["token"] = std::to_string(*listener.pushNotifToken);
         Json::StreamWriterBuilder wbuilder;
         wbuilder["commentStyle"] = "None";
         wbuilder["indentation"] = "";
@@ -895,11 +911,8 @@ DhtProxyClient::doCancelListen(const InfoHash& key, size_t ltoken)
         std::replace(content.begin(), content.end(), '\n', ' ');
         req->set_body(content);
         req->set_header("Content-Length", std::to_string(content.size()));
-        try {
-            restbed::Http::async(req, [](const std::shared_ptr<restbed::Request>&, const std::shared_ptr<restbed::Response>&){});
-        } catch (const std::exception& e) {
-            DHT_LOG.w(key, "[search %s] cancelListen: Http::async failed: %s", key.to_c_str(), e.what());
-        }
+
+        restbed::Http::async(req, [](const std::shared_ptr<restbed::Request>&, const std::shared_ptr<restbed::Response>&){});
     } else {
         // Just stop the request
         if (listener.thread.joinable()) {
@@ -980,7 +993,7 @@ DhtProxyClient::restartListeners()
             state->ok = true;
             auto filter = listener.filter;
             auto cb = listener.cb;
-            restbed::Uri uri(serverHost_ + "/" + search.first.toString());
+            restbed::Uri uri(proxy::HTTP_PROTO + serverHost_ + "/" + search.first.toString());
             auto req = std::make_shared<restbed::Request>(uri);
             req->set_method("LISTEN");
             listener.req = req;
@@ -1021,12 +1034,10 @@ DhtProxyClient::pushNotificationReceived(const std::map<std::string, std::string
                     resubscribe(key, list.second);
             }
         } else {
-            auto keyidx = notification.find("key");
-            InfoHash key(keyidx->second);
-
+            auto token = std::stoull(notification.at("token"));
             for (auto& search: searches_) {
                 for (auto& list : search.second.listeners) {
-                    if (search.first != key || list.second.state->cancel)
+                    if (*list.second.pushNotifToken != token or list.second.state->cancel)
                         continue;
                     DHT_LOG.d(search.first, "[search %s] handling push notification", search.first.to_c_str());
                     auto cb = list.second.cb;
@@ -1052,21 +1063,21 @@ DhtProxyClient::resubscribe(const InfoHash& key, Listener& listener)
     scheduler.syncTime();
     DHT_LOG.d(key, "[search %s] resubscribe push listener", key.to_c_str());
     // Subscribe
-    restbed::Uri uri(serverHost_ + "/" + key.toString());
+    restbed::Uri uri(proxy::HTTP_PROTO + serverHost_ + "/" + key.toString());
     auto req = std::make_shared<restbed::Request>(uri);
     req->set_method("SUBSCRIBE");
 
+    auto pushNotifToken = std::make_shared<proxy::ListenToken>(0);
     auto state = listener.state;
     if (listener.thread.joinable())
         listener.thread.join();
     state->cancel = false;
     state->ok = true;
     listener.req = req;
+    listener.pushNotifToken = pushNotifToken;
     scheduler.edit(listener.refreshJob, scheduler.time() + proxy::OP_TIMEOUT - proxy::OP_MARGIN);
-    auto vcb = listener.cb;
-    auto filter = listener.filter;
-    listener.thread = std::thread([this, req, vcb, filter, state]() {
-        sendListen(req, vcb, filter, state, ListenMethod::RESUBSCRIBE);
+    listener.thread = std::thread([this, req, pushNotifToken, state]() {
+        sendSubscribe(req, pushNotifToken, state);
     });
 #endif
 }
@@ -1086,18 +1097,17 @@ DhtProxyClient::getPushRequest(Json::Value& body) const
 }
 
 void
-DhtProxyClient::fillBody(std::shared_ptr<restbed::Request> req, bool resubscribe)
+DhtProxyClient::fillBodyToGetToken(std::shared_ptr<restbed::Request> req, unsigned token)
 {
     // Fill body with
     // {
     //   "key":"device_key",
+    //   "token": xxx
     // }
     Json::Value body;
     getPushRequest(body);
-    if (!resubscribe) {
-        // This is the first listen, we want to retrieve previous values.
-        body["previous_values"] = true;
-    }
+    if (token > 0)
+        body["token"] = token;
     Json::StreamWriterBuilder wbuilder;
     wbuilder["commentStyle"] = "None";
     wbuilder["indentation"] = "";

src/dht_proxy_server.cpp

@@ -370,6 +370,7 @@ DhtProxyServer::listen(const Sp<restbed::Session>& session)
 #if OPENDHT_PUSH_NOTIFICATIONS
 
 struct DhtProxyServer::Listener {
+    std::shared_ptr<proxy::ListenToken> token;
     std::string clientId;
     std::future<size_t> internalToken;
     Sp<Scheduler::Job> expireJob;
@@ -408,11 +409,12 @@ DhtProxyServer::subscribe(const std::shared_ptr<restbed::Session>& session)
                     s->close(restbed::BAD_REQUEST, "{\"err\":\"No token\"}");
                     return;
                 }
+                auto tokenFromReq = unpackId(root, "token");
                 auto platform = root["platform"].asString();
                 auto isAndroid = platform == "android";
                 auto clientId = root.isMember("client_id") ? root["client_id"].asString() : std::string();
 
-                std::cout << "Subscribe " << infoHash << " client:" << clientId << std::endl;
+                std::cout << "Subscribe " << infoHash << " token:" << tokenFromReq << " client:" << clientId << std::endl;
 
                 {
                     std::lock(schedulerLock_, lockListener_);
@@ -426,40 +428,10 @@ DhtProxyServer::subscribe(const std::shared_ptr<restbed::Session>& session)
                     auto listeners = pushListener->second.listeners.emplace(infoHash, std::vector<Listener>{}).first;
                     for (auto& listener: listeners->second) {
                         if (listener.clientId == clientId) {
+                            *listener.token = tokenFromReq;
                             scheduler_.edit(listener.expireJob, timeout);
                             scheduler_.edit(listener.expireNotifyJob, timeout - proxy::OP_MARGIN);
-                            s->yield(restbed::OK);
-
-                            if (root.isMember("previous_values") &&
-                                root["previous_values"].asBool()) {
-                                dht_->get(
-                                    infoHash,
-                                    [this, s](const Sp<Value> &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<restbed::Session>
-                                                                & /*session*/) {});
-                                        return true;
-                                    },
-                                    [s](bool /*ok* */) {
-                                        // Communication is finished
-                                        if (not s->is_closed()) {
-                                            s->close("{}\n");
-                                        }
-                                    });
-                            } else {
-                                // Communication is finished
-                                if (not s->is_closed()) {
-                                    s->close("{}\n");
-                                }
-                            }
+                            s->close(restbed::OK, "{\"token\": " + std::to_string(tokenFromReq) + "}\n");
                             schedulerCv_.notify_one();
                             return;
                         }
@@ -472,12 +444,15 @@ DhtProxyServer::subscribe(const std::shared_ptr<restbed::Session>& session)
                     pushListener->second.isAndroid = isAndroid;
 
                     // The listener is not found, so add it.
+                    auto token = std::make_shared<proxy::ListenToken>(tokenFromReq);
+                    listener.token = token;
                     listener.internalToken = dht_->listen(infoHash,
-                        [this, infoHash, pushToken, isAndroid, clientId](std::vector<std::shared_ptr<Value>> /*value*/) {
+                        [this, token, infoHash, pushToken, isAndroid, clientId](std::vector<std::shared_ptr<Value>> /*value*/) {
                             // Build message content.
                             Json::Value json;
                             json["key"] = infoHash.toString();
                             json["to"] = clientId;
+                            json["token"] = std::to_string(*token);
                             sendPushNotification(pushToken, json, isAndroid);
                             return true;
                         }
@@ -488,17 +463,18 @@ DhtProxyServer::subscribe(const std::shared_ptr<restbed::Session>& session)
                         }
                     );
                     listener.expireNotifyJob = scheduler_.add(timeout - proxy::OP_MARGIN,
-                        [this, infoHash, pushToken, isAndroid, clientId] {
+                        [this, token, infoHash, pushToken, isAndroid, clientId] {
                             std::cout << "Listener: sending refresh " << infoHash << std::endl;
                             Json::Value json;
                             json["timeout"] = infoHash.toString();
                             json["to"] = clientId;
+                            json["token"] = std::to_string(*token);
                             sendPushNotification(pushToken, json, isAndroid);
                         }
                     );
                 }
                 schedulerCv_.notify_one();
-                s->close(restbed::OK, "{}\n");
+                s->close(restbed::OK, "{\"token\": " + std::to_string(tokenFromReq) + "}\n");
             } catch (...) {
                 s->close(restbed::INTERNAL_SERVER_ERROR, "{\"err\":\"Internal server error\"}");
             }
@@ -573,8 +549,6 @@ DhtProxyServer::cancelPushListen(const std::string& pushToken, const dht::InfoHa
 void
 DhtProxyServer::sendPushNotification(const std::string& token, const Json::Value& json, bool isAndroid) const
 {
-    if (pushServer_.empty())
-        return;
     restbed::Uri uri(proxy::HTTP_PROTO + pushServer_ + "/api/push");
     auto req = std::make_shared<restbed::Request>(uri);
     req->set_method("POST");

src/dhtrunner.cpp

@@ -460,7 +460,6 @@ int bindSocket(const SockAddr& addr, SockAddr& bound)
         setsockopt(sock, IPPROTO_IPV6, IPV6_V6ONLY, (const char*)&set, sizeof(set));
     int rc = bind(sock, addr.get(), addr.getLength());
     if(rc < 0) {
-        rc = errno;
         close(sock);
         throw DhtException("Can't bind socket on " + addr.toString() + " " + strerror(rc));
     }
@@ -521,7 +520,7 @@ DhtRunner::startNetwork(const SockAddr sin4, const SockAddr sin6)
 #ifndef _WIN32
     auto status = pipe(stopfds);
     if (status == -1) {
-        throw DhtException(std::string("Can't open pipe: ") + strerror(errno));
+        throw DhtException("Can't open pipe");
     }
 #else
     udpPipe(stopfds);
@@ -533,29 +532,15 @@ DhtRunner::startNetwork(const SockAddr sin4, const SockAddr sin6)
     s6 = -1;
 
     bound4 = {};
-    if (sin4) {
-        try {
-            s4 = bindSocket(sin4, bound4);
-        } catch (const DhtException& e) {
-            std::cerr << "Can't bind inet socket: " << e.what() << std::endl;
-        }
-    }
+    if (sin4)
+        s4 = bindSocket(sin4, bound4);
 
 #if 1
     bound6 = {};
-    if (sin6) {
-        try {
-            s6 = bindSocket(sin6, bound6);
-        } catch (const DhtException& e) {
-            std::cerr << "Can't bind inet6 socket: " << e.what() << std::endl;
-        }
-    }
+    if (sin6)
+        s6 = bindSocket(sin6, bound6);
 #endif
 
-    if (s4 == -1 && s6 == -1) {
-        throw DhtException("Can't bind socket");
-    }
-
     running_network = true;
     rcv_thread = std::thread([this, stop_readfd]() {
         try {

src/listener.h

@@ -48,4 +48,11 @@ struct LocalListener {
     ValueCallback get_cb;
 };
 
+
+struct SearchListener {
+    Sp<Query> query;
+    Value::Filter filter;
+    ValueCallback get_cb;
+};
+
 }

src/network_engine.cpp

@@ -353,8 +353,9 @@ NetworkEngine::rateLimit(const SockAddr& addr)
         limiter_maintenance = 0;
     }
 
+    auto it = address_rate_limiter.emplace(addr, IpLimiter{});
     // invoke per IP, then global rate limiter
-    return address_rate_limiter[addr].limit(now) and rate_limiter.limit(now);
+    return it.first->second.limit(now) and rate_limiter.limit(now);
 }
 
 bool
@@ -433,8 +434,7 @@ NetworkEngine::processMessage(const uint8_t *buf, size_t buflen, const SockAddr&
     if (msg->type == MessageType::ValueData) {
         auto pmsg_it = partial_messages.find(msg->tid);
         if (pmsg_it == partial_messages.end()) {
-            if (logIncoming_)
-                DHT_LOG.d("Can't find partial message");
+            DHT_LOG.d("Can't find partial message");
             rateLimit(from);
             return;
         }
@@ -474,15 +474,15 @@ NetworkEngine::processMessage(const uint8_t *buf, size_t buflen, const SockAddr&
         process(std::move(msg), from);
     } else {
         // starting partial message session
-        auto k = msg->tid;
-        auto& pmsg = partial_messages[k];
-        if (not pmsg.msg) {
-            pmsg.from = from;
-            pmsg.msg = std::move(msg);
-            pmsg.start = now;
-            pmsg.last_part = now;
-            scheduler.add(now + RX_MAX_PACKET_TIME, std::bind(&NetworkEngine::maintainRxBuffer, this, k));
-            scheduler.add(now + RX_TIMEOUT, std::bind(&NetworkEngine::maintainRxBuffer, this, k));
+        PartialMessage pmsg;
+        pmsg.from = from;
+        pmsg.msg = std::move(msg);
+        pmsg.start = now;
+        pmsg.last_part = now;
+        auto wmsg = partial_messages.emplace(pmsg.msg->tid, std::move(pmsg));
+        if (wmsg.second) {
+            scheduler.add(now + RX_MAX_PACKET_TIME, std::bind(&NetworkEngine::maintainRxBuffer, this, wmsg.first->first));
+            scheduler.add(now + RX_TIMEOUT, std::bind(&NetworkEngine::maintainRxBuffer, this, wmsg.first->first));
         } else
             DHT_LOG.e("Partial message with given TID already exists");
     }
@@ -544,21 +544,19 @@ NetworkEngine::process(std::unique_ptr<ParsedMessage>&& msg, const SockAddr& fro
                 req->reply_time = time_point::min();
                 onError(req, DhtProtocolException {msg->error_code});
             } else {
-                if (logIncoming_)
-                    DHT_LOG.w(msg->id, "[node %s %s] received unknown error message %u",
+                DHT_LOG.w(msg->id, "[node %s %s] received unknown error message %u",
                         msg->id.toString().c_str(), from.toString().c_str(), msg->error_code);
             }
             break;
         }
         case MessageType::Reply:
             if (req) { /* request reply */
-                auto& r = *req;
-                if (r.getType() == MessageType::AnnounceValue or r.getType() == MessageType::Listen)
-                    r.node->authSuccess();
-                r.reply_time = scheduler.time();
+                if (req->getType() == MessageType::AnnounceValue or req->getType() == MessageType::Listen)
+                    req->node->authSuccess();
+                req->reply_time = scheduler.time();
 
                 deserializeNodes(*msg, from);
-                r.setDone(std::move(*msg));
+                req->setDone(std::move(*msg));
                 break;
             } else { /* request socket data */
                 deserializeNodes(*msg, from);
@@ -576,8 +574,7 @@ NetworkEngine::process(std::unique_ptr<ParsedMessage>&& msg, const SockAddr& fro
             switch (msg->type) {
             case MessageType::Ping:
                 ++in_stats.ping;
-                if (logIncoming_)
-                    DHT_LOG.d(node->id, "[node %s] sending pong", node->toString().c_str());
+                DHT_LOG.d(node->id, "[node %s] sending pong", node->toString().c_str());
                 onPing(node);
                 sendPong(from, msg->tid);
                 break;
@@ -598,8 +595,7 @@ NetworkEngine::process(std::unique_ptr<ParsedMessage>&& msg, const SockAddr& fro
                 break;
             }
             case MessageType::AnnounceValue: {
-                if (logIncoming_)
-                    DHT_LOG.d(msg->info_hash, node->id, "[node %s] got 'put' request for %s", node->toString().c_str(), msg->info_hash.toString().c_str());
+                DHT_LOG.d(msg->info_hash, node->id, "[node %s] got 'put' request for %s", node->toString().c_str(), msg->info_hash.toString().c_str());
                 ++in_stats.put;
                 onAnnounce(node, msg->info_hash, msg->token, msg->values, msg->created);
 
@@ -612,15 +608,13 @@ NetworkEngine::process(std::unique_ptr<ParsedMessage>&& msg, const SockAddr& fro
                 break;
             }
             case MessageType::Refresh:
-                if (logIncoming_)
-                    DHT_LOG.d(msg->info_hash, node->id, "[node %s] got 'refresh' request for %s", node->toString().c_str(), msg->info_hash.toString().c_str());
+                DHT_LOG.d(msg->info_hash, node->id, "[node %s] got 'refresh' request for %s", node->toString().c_str(), msg->info_hash.toString().c_str());
                 onRefresh(node, msg->info_hash, msg->token, msg->value_id);
                 /* Same note as above in MessageType::AnnounceValue applies. */
                 sendValueAnnounced(from, msg->tid, msg->value_id);
                 break;
             case MessageType::Listen: {
-                if (logIncoming_)
-                    DHT_LOG.d(msg->info_hash, node->id, "[node %s] got 'listen' request for %s", node->toString().c_str(), msg->info_hash.toString().c_str());
+                DHT_LOG.d(msg->info_hash, node->id, "[node %s] got 'listen' request for %s", node->toString().c_str(), msg->info_hash.toString().c_str());
                 ++in_stats.listen;
                 RequestAnswer answer = onListen(node, msg->info_hash, msg->token, msg->socket_id, std::move(msg->query));
                 auto nnodes = bufferNodes(from.getFamily(), msg->info_hash, msg->want, answer.nodes4, answer.nodes6);

src/op_cache.cpp

@@ -25,7 +25,7 @@ bool
 OpValueCache::onValuesAdded(const std::vector<Sp<Value>>& vals) {
     std::vector<Sp<Value>> newValues;
     for (const auto& v : vals) {
-        auto viop = values.emplace(v->id, v);
+        auto viop = values.emplace(v->id, OpCacheValueStorage{v});
         if (viop.second) {
             newValues.emplace_back(v);
         } else {
@@ -50,7 +50,7 @@ OpValueCache::onValuesExpired(const std::vector<Sp<Value>>& vals) {
     return expiredValues.empty() ? true : callback(expiredValues, true);
 }
 std::vector<Sp<Value>>
-OpValueCache::get(const Value::Filter& filter) const {
+OpValueCache::get(Value::Filter& filter) const {
     std::vector<Sp<Value>> ret;
     if (not filter)
         ret.reserve(values.size());
@@ -108,57 +108,33 @@ OpCache::getExpiration() const {
     return lastRemoved + EXPIRATION;
 }
 
-SearchCache::OpMap::iterator
-SearchCache::getOp(const Sp<Query>& q)
-{
-    // find exact match
-    auto op = ops.find(q);
-    if (op != ops.end())
-        return op;
-    // find satisfying query
-    for (auto it = ops.begin(); it != ops.end(); it++) {
-        if (q->isSatisfiedBy(*it->first)) {
-            return it;
-        }
-    }
-    return ops.end();
-}
-
-SearchCache::OpMap::const_iterator
-SearchCache::getOp(const Sp<Query>& q) const
-{
-    // find exact match
-    auto op = ops.find(q);
-    if (op != ops.cend())
-        return op;
-    // find satisfying query
-    for (auto it = ops.begin(); it != ops.end(); it++) {
-        if (q->isSatisfiedBy(*it->first)) {
-            return it;
-        }
-    }
-    return ops.cend();
-}
-
 size_t
-SearchCache::listen(ValueCallback get_cb, Sp<Query> q, Value::Filter filter, OnListen onListen)
+SearchCache::listen(ValueCallback get_cb, Sp<Query> q, Value::Filter filter, std::function<size_t(Sp<Query>, ValueCallback)> onListen)
 {
     // find exact match
-    auto op = getOp(q);
+    auto op = ops.find(q);
+    if (op == ops.end()) {
+        // find satisfying query
+        for (auto it = ops.begin(); it != ops.end(); it++) {
+            if (q->isSatisfiedBy(*it->first)) {
+                op = it;
+                break;
+            }
+        }
+    }
     if (op == ops.end()) {
         // New query
         op = ops.emplace(q, std::unique_ptr<OpCache>(new OpCache)).first;
         auto& cache = *op->second;
         cache.searchToken = onListen(q, [&](const std::vector<Sp<Value>>& values, bool expired){
             return cache.onValue(values, expired);
-        }, [&](ListenSyncStatus status) {
-            cache.onNodeChanged(status);
         });
     }
     auto token = nextToken_++;
     if (nextToken_ == 0)
         nextToken_++;
-    return op->second->addListener(token, get_cb, q, filter) ? token : 0;
+    op->second->addListener(token, get_cb, q, filter);
+    return token;
 }
 
 bool
@@ -201,22 +177,8 @@ SearchCache::expire(const time_point& now, std::function<void(size_t)> onCancel)
     return ret;
 }
 
-bool
-SearchCache::get(const Value::Filter& f, const Sp<Query>& q, const GetCallback& gcb, const DoneCallback& dcb) const
-{
-    auto op = getOp(q);
-    if (op != ops.end()) {
-        auto vals = op->second->get(f);
-        if ((not vals.empty() and not gcb(vals)) or op->second->isSynced()) {
-            dcb(true, {});
-            return true;
-        }
-    }
-    return false;
-}
-
 std::vector<Sp<Value>>
-SearchCache::get(const Value::Filter& filter) const {
+SearchCache::get(Value::Filter& filter) const {
     if (ops.size() == 1)
         return ops.begin()->second->get(filter);
     std::map<Value::Id, Sp<Value>> c;

src/op_cache.h

@@ -27,7 +27,7 @@ struct OpCacheValueStorage
 {
     Sp<Value> data {};
     unsigned refCount {1};
-    OpCacheValueStorage(Sp<Value> val) : data(val) {}
+    OpCacheValueStorage(Sp<Value> val = {}) : data(val) {}
 };
 
 class OpValueCache {
@@ -37,13 +37,10 @@ public:
         o.callback = {};
     }
 
-    static ValueCallback cacheCallback(ValueCallback&& cb, std::function<void()>&& onCancel) {
+    static ValueCallback cacheCallback(ValueCallback&& cb) {
         auto cache = std::make_shared<OpValueCache>(std::forward<ValueCallback>(cb));
-        return [cache, onCancel](const std::vector<Sp<Value>>& vals, bool expired){
-            auto ret = cache->onValue(vals, expired);
-            if (not ret)
-                onCancel();
-            return ret;
+        return [cache](const std::vector<Sp<Value>>& vals, bool expired){
+            return cache->onValue(vals, expired);
         };
     }
 
@@ -57,18 +54,7 @@ public:
     bool onValuesAdded(const std::vector<Sp<Value>>& vals);
     bool onValuesExpired(const std::vector<Sp<Value>>& vals);
 
-    void onNodeChanged(ListenSyncStatus status) {
-        switch (status) {
-            case ListenSyncStatus::ADDED: nodes++; break;
-            case ListenSyncStatus::REMOVED: nodes--; break;
-            case ListenSyncStatus::SYNCED : syncedNodes++; break;
-            case ListenSyncStatus::UNSYNCED: syncedNodes--; break;
-        }
-    }
-
-    bool isSynced() const { return nodes > 0 and syncedNodes == nodes; }
-
-    std::vector<Sp<Value>> get(const Value::Filter& filter) const;
+    std::vector<Sp<Value>> get(Value::Filter& filter) const;
     Sp<Value> get(Value::Id id) const;
     std::vector<Sp<Value>> getValues() const;
 
@@ -76,8 +62,6 @@ private:
     OpValueCache(const OpValueCache&) = delete;
     OpValueCache& operator=(const OpValueCache&) = delete;
 
-    size_t nodes {0};
-    size_t syncedNodes {0};
     std::map<Value::Id, OpCacheValueStorage> values {};
     ValueCallback callback;
 };
@@ -96,23 +80,13 @@ public:
         cache.onValue(vals, expired);
         return not listeners.empty();
     }
-    void onNodeChanged(ListenSyncStatus status) {
-        cache.onNodeChanged(status);
-    }
 
     void onValuesAdded(const std::vector<Sp<Value>>& vals);
     void onValuesExpired(const std::vector<Sp<Value>>& vals);
 
-    bool addListener(size_t token, ValueCallback cb, Sp<Query> q, Value::Filter filter) {
+    void addListener(size_t token, ValueCallback cb, Sp<Query> q, Value::Filter filter) {
         listeners.emplace(token, LocalListener{q, filter, cb});
-        auto cached = cache.get(filter);
-        if (not cached.empty()) {
-            auto ret = cb(cached, false);
-            if (not ret)
-                listeners.erase(token);
-            return ret;
-        }
-        return true;
+        cb(cache.get(filter), false);
     }
 
     bool removeListener(size_t token, const time_point& now) {
@@ -128,7 +102,7 @@ public:
         return listeners.empty();
     }
 
-    std::vector<Sp<Value>> get(const Value::Filter& filter) const {
+    std::vector<Sp<Value>> get(Value::Filter& filter) const {
         return cache.get(filter);
     }
 
@@ -136,10 +110,6 @@ public:
         return cache.get(id);
     }
 
-    bool isSynced() const {
-        return cache.isSynced();
-    }
-
     bool isExpired(const time_point& now) const {
         return listeners.empty() and (lastRemoved + EXPIRATION < now);
     }
@@ -160,9 +130,7 @@ class SearchCache {
 public:
     SearchCache() {}
     SearchCache(SearchCache&&) = default;
-
-    using OnListen = std::function<size_t(Sp<Query>, ValueCallback, SyncCallback)>;
-    size_t listen(ValueCallback get_cb, Sp<Query> q, Value::Filter filter, OnListen onListen);
+    size_t listen(ValueCallback get_cb, Sp<Query> q, Value::Filter filter, std::function<size_t(Sp<Query>, ValueCallback)> onListen);
 
     bool cancelListen(size_t gtoken, const time_point& now);
     void cancelAll(std::function<void(size_t)> onCancel);
@@ -172,19 +140,14 @@ public:
         return nextExpiration_;
     }
 
-    bool get(const Value::Filter& f, const Sp<Query>& q, const GetCallback& gcb, const DoneCallback& dcb) const;
-    std::vector<Sp<Value>> get(const Value::Filter& filter) const;
+    std::vector<Sp<Value>> get(Value::Filter& filter) const;
     Sp<Value> get(Value::Id id) const;
 
 private:
     SearchCache(const SearchCache&) = delete;
     SearchCache& operator=(const SearchCache&) = delete;
 
-    using OpMap = std::map<Sp<Query>, std::unique_ptr<OpCache>>;
-    OpMap ops {};
-    OpMap::iterator getOp(const Sp<Query>& q);
-    OpMap::const_iterator getOp(const Sp<Query>& q) const;
-
+    std::map<Sp<Query>, std::unique_ptr<OpCache>> ops {};
     size_t nextToken_ {1};
     time_point nextExpiration_ {time_point::max()};
 };

src/search.h

@@ -65,8 +65,7 @@ struct Dht::SearchNode {
         ValueCache cache;
         Sp<Scheduler::Job> cacheExpirationJob {};
         Sp<net::Request> req {};
-        CachedListenStatus(ValueStateCallback&& cb, SyncCallback&& scb)
-         : cache(std::forward<ValueStateCallback>(cb), std::forward<SyncCallback>(scb)) {}
+        CachedListenStatus(ValueStateCallback&& cb) : cache(std::forward<ValueStateCallback>(cb)) {}
         CachedListenStatus(CachedListenStatus&&) = default;
         CachedListenStatus(const CachedListenStatus&) = delete;
         CachedListenStatus& operator=(const CachedListenStatus&) = delete;
@@ -226,13 +225,6 @@ struct Dht::SearchNode {
         }
     }
 
-    void onListenSynced(const Sp<Query>& q, bool synced = true) {
-        auto l = listenStatus.find(q);
-        if (l != listenStatus.end()) {
-            l->second.cache.onSynced(synced);
-        }
-    }
-
     void expireValues(const Sp<Query>& q, Scheduler& scheduler) {
         auto l = listenStatus.find(q);
         if (l != listenStatus.end()) {
@@ -385,12 +377,6 @@ struct Dht::Search {
     std::multimap<time_point, Get> callbacks {};
 
     /* listeners */
-    struct SearchListener {
-        Sp<Query> query;
-        Value::Filter filter;
-        ValueCallback get_cb;
-        SyncCallback sync_cb;
-    };
     std::map<size_t, SearchListener> listeners {};
     size_t listener_token = 1;
 
@@ -491,20 +477,21 @@ struct Dht::Search {
 
     void get(Value::Filter f, const Sp<Query>& q, const QueryCallback& qcb, const GetCallback& gcb, const DoneCallback& dcb, Scheduler& scheduler) {
         if (gcb or qcb) {
-            if (not cache.get(f, q, gcb, dcb)) {
-                const auto& now = scheduler.time();
-                callbacks.emplace(now, Get { now, f, q, qcb, gcb, dcb });
-                scheduler.edit(nextSearchStep, now);
-            }
+            const auto& now = scheduler.time();
+            callbacks.emplace(now, Get { now, f, q, qcb, gcb, dcb });
+            auto values = cache.get(f);
+            if (not values.empty())
+                gcb(values);
+            scheduler.edit(nextSearchStep, now);
         }
     }
 
     size_t listen(ValueCallback cb, Value::Filter f, const Sp<Query>& q, Scheduler& scheduler) {
         //DHT_LOG.e(id, "[search %s IPv%c] listen", id.toString().c_str(), (af == AF_INET) ? '4' : '6');
-        return cache.listen(cb, q, f, [&](const Sp<Query>& q, ValueCallback vcb, SyncCallback scb){
+        return cache.listen(cb, q, f, [&](const Sp<Query>& q, ValueCallback vcb){
             done = false;
             auto token = ++listener_token;
-            listeners.emplace(token, SearchListener{q, f, vcb, scb});
+            listeners.emplace(token, SearchListener{q, f, vcb});
             scheduler.edit(nextSearchStep, scheduler.time());
             return token;
         });

src/securedht.cpp

@@ -68,33 +68,19 @@ ValueType
 SecureDht::secureType(ValueType&& type)
 {
     type.storePolicy = [this,type](InfoHash id, Sp<Value>& v, const InfoHash& nid, const SockAddr& a) {
-        if (v->isSigned()) {
-            if (!v->signatureChecked) {
-                v->signatureChecked = true;
-                v->signatureValid = v->owner and v->owner->checkSignature(v->getToSign(), v->signature);
-            }
-            if (!v->signatureValid) {
-                DHT_LOG.WARN("Signature verification failed");
-                return false;
-            }
-        }
+        if (v->isSigned())
+            return v->checkSignature();
         return type.storePolicy(id, v, nid, a);
     };
     type.editPolicy = [this,type](InfoHash id, const Sp<Value>& o, Sp<Value>& n, const InfoHash& nid, const SockAddr& a) {
-        if (!o->isSigned())
+        if (not o->isSigned())
             return type.editPolicy(id, o, n, nid, a);
-        if (o->owner != n->owner) {
+        if (o->owner != n->owner or not n->isSigned()) {
             DHT_LOG.WARN("Edition forbidden: owner changed.");
             return false;
         }
-        if (!n->signatureChecked) {
-            n->signatureChecked = true;
-            n->signatureValid = o->owner and o->owner->checkSignature(n->getToSign(), n->signature);
-        }
-        if (!n->signatureValid) {
-            DHT_LOG.WARN("Edition forbidden: signature verification failed.");
+        if (not n->checkSignature())
             return false;
-        }
         if (o->seq == n->seq) {
             // If the data is exactly the same,
             // it can be reannounced, possibly by someone else.
@@ -241,35 +227,23 @@ SecureDht::checkValue(const Sp<Value>& v)
 #endif
             return {};
         }
-        if (v->decrypted) {
-            return v->decryptedValue;
-        }
-        v->decrypted = true;
         try {
-            Value decrypted_val (decrypt(*v));
-            if (decrypted_val.recipient == getId()) {
-                if (decrypted_val.owner)
-                    nodesPubKeys_[decrypted_val.owner->getId()] = decrypted_val.owner;
-                v->decryptedValue = std::make_shared<Value>(std::move(decrypted_val));
-                return v->decryptedValue;
+            if (auto decrypted_val = v->decrypt(*key_)) {
+                if (decrypted_val->owner)
+                    nodesPubKeys_[decrypted_val->owner->getId()] = decrypted_val->owner;
+                return decrypted_val;
             }
-            // Ignore values belonging to other people
         } catch (const std::exception& e) {
             DHT_LOG.WARN("Could not decrypt value %s : %s", v->toString().c_str(), e.what());
         }
     }
     // Check signed values
     else if (v->isSigned()) {
-        if (v->signatureChecked) {
-            return v->signatureValid ? v : Sp<Value>{};
-        }
-        v->signatureChecked = true;
-        if (v->owner and v->owner->checkSignature(v->getToSign(), v->signature)) {
-            v->signatureValid = true;
-            nodesPubKeys_[v->owner->getId()] = v->owner;
+        if (v->checkSignature()) {
+            if (v->owner)
+                nodesPubKeys_[v->owner->getId()] = v->owner;
             return v;
-        }
-        else
+        } else
             DHT_LOG.WARN("Signature verification failed for %s", v->toString().c_str());
     }
     // Forward normal values
@@ -284,8 +258,6 @@ SecureDht::getCallbackFilter(ValueCallback cb, Value::Filter&& filter)
 {
     return [=](const std::vector<Sp<Value>>& values, bool expired) {
         std::vector<Sp<Value>> tmpvals {};
-        if (not filter)
-            tmpvals.reserve(values.size());
         for (const auto& v : values) {
             if (auto nv = checkValue(v))
                 if (not filter or filter(*nv))
@@ -303,8 +275,6 @@ SecureDht::getCallbackFilter(GetCallback cb, Value::Filter&& filter)
 {
     return [=](const std::vector<Sp<Value>>& values) {
         std::vector<Sp<Value>> tmpvals {};
-        if (not filter)
-            tmpvals.reserve(values.size());
         for (const auto& v : values) {
             if (auto nv = checkValue(v))
                 if (not filter or filter(*nv))
@@ -368,8 +338,7 @@ SecureDht::putSigned(const InfoHash& hash, Sp<Value> val, DoneCallback callback,
             sign(*val);
             dht_->put(hash, val, callback, time_point::max(), permanent);
         },
-        Value::IdFilter(val->id),
-        std::move(Where().id(val->id))
+        Value::IdFilter(val->id)
     );
 }
 

src/value.cpp

@@ -54,14 +54,12 @@ std::ostream& operator<< (std::ostream& s, const Value& v)
             s << IpServiceAnnouncement(v.data);
         } else if (v.type == CERTIFICATE_TYPE.id) {
             s << "Certificate";
-#ifdef OPENDHT_LOG_CRT_ID
             try {
-                auto h = crypto::Certificate(v.data).getPublicKey().getLongId();
+                InfoHash h = crypto::Certificate(v.data).getPublicKey().getId();
                 s << " with ID " << h;
             } catch (const std::exception& e) {
                 s << " (invalid)";
             }
-#endif
         } else {
             s << "Data (type: " << v.type << " ): ";
             s << std::hex;
@@ -235,6 +233,39 @@ Value::toJson() const
     return val;
 }
 
+bool
+Value::checkSignature()
+{
+    if (!signatureChecked) {
+        signatureChecked = true;
+        if (isSigned()) {
+            signatureValid = owner and owner->checkSignature(getToSign(), signature);
+        }
+    }
+    return signatureValid;
+}
+
+Sp<Value>
+Value::decrypt(const crypto::PrivateKey& key)
+{
+    if (not decrypted) {
+        decrypted = true;
+        if (isEncrypted()) {
+            auto decryptedBlob = key.decrypt(cypher);
+            std::unique_ptr<Value> v {new Value(id)};
+            auto msg = msgpack::unpack((const char*)decryptedBlob.data(), decryptedBlob.size());
+            v->msgpack_unpack_body(msg.get());
+            if (v->recipient != key.getPublicKey().getId())
+                throw crypto::DecryptError("Recipient mismatch");
+            // Ignore values belonging to other people
+            if (not v->owner or not v->owner->checkSignature(v->getToSign(), v->signature))
+                throw crypto::DecryptError("Signature mismatch");
+            decryptedValue = std::move(v);
+        }
+    }
+    return decryptedValue;
+}
+
 uint64_t
 unpackId(const Json::Value& json, const std::string& key) {
     uint64_t ret = 0;
@@ -288,7 +319,7 @@ FieldValue::getLocalFilter() const
         case Value::Field::UserType:
             return Value::UserTypeFilter(std::string {blobValue.begin(), blobValue.end()});
         default:
-            return {};
+            return Value::AllFilter();
     }
 }
 
@@ -497,7 +528,8 @@ template <typename T>
 bool subset(std::vector<T> fds, std::vector<T> qfds)
 {
     for (auto& fd : fds) {
-        if (std::find_if(qfds.begin(), qfds.end(), [&fd](T& _vfd) { return fd == _vfd; }) == qfds.end())
+        auto correspondance = std::find_if(qfds.begin(), qfds.end(), [&fd](T& _vfd) { return fd == _vfd; });
+        if (correspondance == qfds.end())
             return false;
     }
     return true;
@@ -505,9 +537,10 @@ bool subset(std::vector<T> fds, std::vector<T> qfds)
 
 bool Select::isSatisfiedBy(const Select& os) const {
     /* empty, means all values are selected. */
-    return fieldSelection_.empty() ?
-        os.fieldSelection_.empty() :
-        subset(fieldSelection_, os.fieldSelection_);
+    if (fieldSelection_.empty() and not os.fieldSelection_.empty())
+        return false;
+    else
+        return subset(fieldSelection_, os.fieldSelection_);
 }
 
 bool Where::isSatisfiedBy(const Where& ow) const {

src/value_cache.h

@@ -22,32 +22,19 @@
 namespace dht {
 
 using ValueStateCallback = std::function<void(const std::vector<Sp<Value>>&, bool)>;
-enum class ListenSyncStatus { ADDED, SYNCED, UNSYNCED, REMOVED };
-using SyncCallback = std::function<void(ListenSyncStatus)>;
 using CallbackQueue = std::list<std::function<void()>>;
 
 class ValueCache {
 public:
-    ValueCache(ValueStateCallback&& cb, SyncCallback&& scb = {})
-        : callback(std::forward<ValueStateCallback>(cb)), syncCallback(std::move(scb))
-    {
-        if (syncCallback)
-            syncCallback(ListenSyncStatus::ADDED);
-    }
-    ValueCache(ValueCache&& o) : values(std::move(o.values)), callback(std::move(o.callback)), syncCallback(std::move(o.syncCallback)) {
+    ValueCache(ValueStateCallback&& cb) : callback(std::forward<ValueStateCallback>(cb)) {}
+    ValueCache(ValueCache&& o) : values(std::move(o.values)), callback(std::move(o.callback)) {
         o.callback = {};
-        o.syncCallback = {};
     }
 
     ~ValueCache() {
         auto q = clear();
         for (auto& cb: q)
             cb();
-        if (syncCallback) {
-            if (status == ListenSyncStatus::SYNCED)
-                syncCallback(ListenSyncStatus::UNSYNCED);
-            syncCallback(ListenSyncStatus::REMOVED);
-        }
     }
 
     CallbackQueue clear() {
@@ -133,15 +120,6 @@ public:
         return ret;
     }
 
-    void onSynced(bool synced) {
-        auto newStatus = synced ? ListenSyncStatus::SYNCED : ListenSyncStatus::UNSYNCED;
-        if (status != newStatus) {
-            status = newStatus;
-            if (syncCallback)
-                syncCallback(newStatus);
-        }
-    }
-
 private:
     // prevent copy
     ValueCache(const ValueCache&) = delete;
@@ -163,8 +141,6 @@ private:
 
     std::map<Value::Id, CacheValueStorage> values;
     ValueStateCallback callback;
-    SyncCallback syncCallback;
-    ListenSyncStatus status {ListenSyncStatus::UNSYNCED};
 
     CallbackQueue addValues(const std::vector<Sp<Value>>& new_values, const TypeStore& types, const time_point& now) {
         std::vector<Sp<Value>> nvals;

tests/dhtproxytester.cpp

@@ -77,6 +77,7 @@ DhtProxyTester::testGetPut() {
     CPPUNIT_ASSERT(vals.front()->data == val_data);
 }
 
+
 void
 DhtProxyTester::testListen() {
     bool done = false;
@@ -107,7 +108,7 @@ DhtProxyTester::testListen() {
 
     cv.wait_for(lk, std::chrono::seconds(10), [&]{ return done; });
     done = false;
-    // Here values should contains 1 values
+    // Here values should contains 2 values
     CPPUNIT_ASSERT_EQUAL(static_cast<int>(values.size()), 1);
     CPPUNIT_ASSERT(values.front() == firstVal_data);
 
@@ -117,63 +118,9 @@ DhtProxyTester::testListen() {
     auto secondVal_data = secondVal.data;
     nodePeer.put(key, std::move(secondVal));
     cv.wait_for(lk, std::chrono::seconds(10), [&]{ return done; });
-    // Here values should contains 2 values
+    // Here values should contains 3 values
     CPPUNIT_ASSERT_EQUAL(static_cast<int>(values.size()), 2);
     CPPUNIT_ASSERT(values.back() == secondVal_data);
 }
 
-void
-DhtProxyTester::testResubscribeGetValues() {
-    nodeClient->setPushNotificationToken("atlas");
-
-    bool done = false;
-    std::condition_variable cv;
-    std::mutex cv_m;
-    std::unique_lock<std::mutex> lk(cv_m);
-    auto key = dht::InfoHash::get("GLaDOs");
-
-    // If a peer send a value, the listen operation from the client
-    // should retrieve this value
-    dht::Value firstVal {"Hey! It's been a long time. How have you been?"};
-    auto firstVal_data = firstVal.data;
-    nodePeer.put(key, std::move(firstVal), [&](bool) {
-        done = true;
-        cv.notify_all();
-    });
-    cv.wait_for(lk, std::chrono::seconds(10), [&]{ return done; });
-    done = false;
-
-    // Send a first subscribe, the value is sent via a push notification
-    // So ignore values here.
-    nodeClient->listen(key, [&](const std::vector<std::shared_ptr<dht::Value>>&, bool) {
-        return true;
-    });
-    cv.wait_for(lk, std::chrono::seconds(1));
-
-    // Reboot node (to avoid cache)
-    nodeClient->join();
-    nodeClient->run(42242, {}, true);
-    nodeClient->bootstrap(nodePeer.getBound());
-    nodeClient->setProxyServer("127.0.0.1:8080");
-    nodeClient->enableProxy(true);
-    nodeClient->setPushNotificationToken("atlas");
-
-    // For the second subscribe, the proxy will return the value in the body
-    auto values = std::vector<dht::Blob>();
-    nodeClient->listen(key, [&](const std::vector<std::shared_ptr<dht::Value>>& v, bool) {
-        for (const auto& value : v)
-            values.emplace_back(value->data);
-        done = true;
-        cv.notify_all();
-        return true;
-    });
-
-    cv.wait_for(lk, std::chrono::seconds(10), [&]{ return done; });
-    done = false;
-    // Here values should still contains 1 values
-    CPPUNIT_ASSERT_EQUAL(static_cast<int>(values.size()), 1);
-    CPPUNIT_ASSERT(values.front() == firstVal_data);
-
-}
-
 }  // namespace test

tests/dhtproxytester.h

@@ -32,7 +32,6 @@ class DhtProxyTester : public CppUnit::TestFixture {
     CPPUNIT_TEST_SUITE(DhtProxyTester);
     CPPUNIT_TEST(testGetPut);
     CPPUNIT_TEST(testListen);
-    CPPUNIT_TEST(testResubscribeGetValues);
     CPPUNIT_TEST_SUITE_END();
 
  public:
@@ -40,25 +39,19 @@ class DhtProxyTester : public CppUnit::TestFixture {
      * Method automatically called before each test by CppUnit
      * Init nodes
      */
-   void setUp();
+    void setUp();
     /**
      * Method automatically called after each test CppUnit
      */
-   void tearDown();
+    void tearDown();
     /**
      * Test get and put methods
      */
-   void testGetPut();
+    void testGetPut();
     /**
      * Test listen
      */
-   void testListen();
-
-   /**
-    * When a proxy redo a subscribe on the proxy
-    * it should retrieve existant values
-    */
-   void testResubscribeGetValues();
+    void testListen();
 
  private:
     dht::DhtRunner nodePeer {};

tests/dhtrunnertester.cpp

@@ -50,56 +50,15 @@ DhtRunnerTester::testGetPut() {
     auto key = dht::InfoHash::get("123");
     dht::Value val {"hey"};
     auto val_data = val.data;
-    std::promise<bool> p;
-    node2.put(key, std::move(val), [&](bool ok){
-        p.set_value(ok);
-    });
-    CPPUNIT_ASSERT(p.get_future().get());
-    auto vals = node1.get(key).get();
+    node1.put(key, std::move(val));
+    auto vals = node2.get(key).get();
     CPPUNIT_ASSERT(not vals.empty());
     CPPUNIT_ASSERT(vals.front()->data == val_data);
 }
 
 void
 DhtRunnerTester::testListen() {
-    std::atomic_uint valueCount(0);
-    std::atomic_uint putCount(0);
-
-    auto a = dht::InfoHash::get("234");
-    auto b = dht::InfoHash::get("2345");
-    auto c = dht::InfoHash::get("23456");
-    constexpr unsigned N = 32;
-
-    auto ftokena = node1.listen(a, [&](const std::shared_ptr<dht::Value>&){
-        valueCount++;
-        return true;
-    });
-
-    auto ftokenb = node1.listen(b, [&](const std::shared_ptr<dht::Value>&){
-        valueCount++;
-        return false;
-    });
-
-    auto ftokenc = node1.listen(c, [&](const std::shared_ptr<dht::Value>&){
-        valueCount++;
-        return true;
-    });
-
-    for (unsigned i=0; i<N; i++) {
-        node2.put(a, dht::Value("v1"), [&](bool ok) { if (ok) putCount++; });
-        node2.put(b, dht::Value("v2"), [&](bool ok) { if (ok) putCount++; });
-    }
-
-    std::this_thread::sleep_for(std::chrono::milliseconds(100));
-    auto tokena = ftokena.get();
-    auto tokenb = ftokenb.get();
-    auto tokenc = ftokenc.get();
-
-    CPPUNIT_ASSERT(tokena);
-    CPPUNIT_ASSERT(tokenb);
-    CPPUNIT_ASSERT(tokenc);
-    CPPUNIT_ASSERT_EQUAL(N * 2u, putCount.load());
-    CPPUNIT_ASSERT_EQUAL(N + 1u, valueCount.load());
+    // TODO
 }
 
 }  // namespace test

tools/dhtchat.cpp

@@ -64,15 +64,6 @@ main(int argc, char **argv)
     try {
         dht.run(params.port, dht::crypto::generateIdentity("DHT Chat Node"), true, params.network);
 
-        if (params.log) {
-            if (params.syslog)
-                log::enableSyslog(dht, "dhtnode");
-            else if (not params.logfile.empty())
-                log::enableFileLogging(dht, params.logfile);
-            else
-                log::enableLogging(dht);
-        }
-
         if (not params.bootstrap.first.empty())
             dht.bootstrap(params.bootstrap.first.c_str(), params.bootstrap.second.c_str());