dht: optimize get

op cache: optimize cache insertion
filter: cleanup and optimize
2026-11-09 12:57:03 +00:00 · 2019-02-19 01:59:59 -05:00 · 2019-02-19 01:59:59 -05:00 · 2019-02-19 01:59:59 -05:00 · 2019-02-19 01:59:59 -05:00 · 2019-02-19 01:59:59 -05:00
30 changed files with 848 additions and 548 deletions
@@ -1,7 +1,7 @@
 cmake_minimum_required (VERSION 3.1)
 project (opendht)
 set (opendht_VERSION_MAJOR 1)
-set (opendht_VERSION_MINOR 8.0)
+set (opendht_VERSION_MINOR 8.1)
 set (opendht_VERSION ${opendht_VERSION_MAJOR}.${opendht_VERSION_MINOR})
 set (PACKAGE_VERSION ${opendht_VERSION})
 set (VERSION "${opendht_VERSION}")
@@ -70,6 +70,8 @@ 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)
@@ -304,7 +306,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
@@ -325,16 +327,17 @@ 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()
@@ -16,7 +16,9 @@ if ENABLE_TESTS
 SUBDIRS += tests
 endif

+if HAVE_DOXYGEN
 SUBDIRS += doc
+endif

 ACLOCAL_AMFLAGS = -I m4

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

@@ -193,7 +193,9 @@ AM_COND_IF(ENABLE_PROXY_SERVER_IDENTITY,
    [CPPFLAGS+=" -DOPENDHT_PROXY_SERVER_IDENTITY=true"],
    [CPPFLAGS+=" -DOPENDHT_PROXY_SERVER_IDENTITY=false"])

-AC_CONFIG_FILES([doc/Doxyfile doc/Makefile])
+AM_COND_IF([HAVE_DOXYGEN], [
+  AC_CONFIG_FILES([doc/Doxyfile doc/Makefile])
+])

 dnl Configure setup.py if we build the python module
 AM_COND_IF([USE_CYTHON], [
@@ -163,7 +163,7 @@ public:
    /**
     * Get locally stored data for the given hash.
     */
-    std::vector<Sp<Value>> getLocal(const InfoHash& key, Value::Filter f = Value::AllFilter()) const;
+    std::vector<Sp<Value>> getLocal(const InfoHash& key, Value::Filter f = {}) 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 {16384};
+    static constexpr unsigned MAX_HASHES {64 * 1024};

    /* The maximum number of searches we keep data about. */
-    static constexpr unsigned MAX_SEARCHES {16384};
+    static constexpr unsigned MAX_SEARCHES {64 * 1024};

    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 = Value::AllFilter(), const Sp<Query>& q = {});
+    size_t listenTo(const InfoHash& id, sa_family_t af, ValueCallback cb, Value::Filter f = {}, const Sp<Query>& q = {});

    /**
     * Refill the search with good nodes if possible.
@@ -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 = Value::AllFilter()) const = 0;
+    virtual std::vector<Sp<Value>> getLocal(const InfoHash& key, Value::Filter f = {}) const = 0;

    /**
     * Get locally stored data for the given key and value id.
@@ -41,8 +41,6 @@ namespace Json {

 namespace dht {

-class SearchCache;
-
 class OPENDHT_PUBLIC DhtProxyClient final : public DhtInterface {
 public:

@@ -276,12 +274,18 @@ private:

    void opFailed();

-    size_t doListen(const InfoHash& key, ValueCallback, Value::Filter);
    bool doCancelListen(const InfoHash& key, size_t token);

    struct ListenState;
-    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);
+    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 doPut(const InfoHash&, Sp<Value>, DoneCallback, time_point created, bool permanent);

@@ -351,6 +355,8 @@ private:
     */
    void confirmProxy();
    Sp<Scheduler::Job> nextProxyConfirmation {};
+    Sp<Scheduler::Job> listenerRestart {};
+
    /**
     * Relaunch LISTEN requests if the client disconnect/reconnect.
     */
@@ -371,11 +377,11 @@ private:
    const std::function<void()> loopSignal_;

 #if OPENDHT_PUSH_NOTIFICATIONS
-    void fillBodyToGetToken(std::shared_ptr<restbed::Request> request, unsigned token = 0);
+    void fillBody(std::shared_ptr<restbed::Request> request, bool resubscribe);
    void getPushRequest(Json::Value&) const;
 #endif // OPENDHT_PUSH_NOTIFICATIONS

-    bool isDestroying_ {false};
+    std::atomic_bool isDestroying_ {false};
 };

 }
@@ -63,20 +63,20 @@ public:
    DhtRunner();
    virtual ~DhtRunner();

-    void get(InfoHash id, GetCallbackSimple cb, DoneCallback donecb={}, Value::Filter f = Value::AllFilter(), Where w = {}) {
+    void get(InfoHash id, GetCallbackSimple cb, DoneCallback donecb={}, Value::Filter f = {}, Where w = {}) {
        get(id, bindGetCb(cb), donecb, f, w);
    }

-    void get(InfoHash id, GetCallbackSimple cb, DoneCallbackSimple donecb={}, Value::Filter f = Value::AllFilter(), Where w = {}) {
+    void get(InfoHash id, GetCallbackSimple cb, DoneCallbackSimple donecb={}, Value::Filter f = {}, 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 = Value::AllFilter(), Where w = {}) {
+    void get(InfoHash id, GetCallback cb, DoneCallbackSimple donecb={}, Value::Filter f = {}, Where w = {}) {
        get(id, cb, bindDoneCb(donecb), f, w);
    }
-    void get(const std::string& key, GetCallback vcb, DoneCallbackSimple dcb={}, Value::Filter f = Value::AllFilter(), Where w = {});
+    void get(const std::string& key, GetCallback vcb, DoneCallbackSimple dcb={}, Value::Filter f = {}, Where w = {});

    template <class T>
    void get(InfoHash hash, std::function<bool(std::vector<T>&&)> cb, DoneCallbackSimple dcb={})
@@ -290,9 +290,9 @@ public:
     */
    Sp<Request> sendFindNode(Sp<Node> n,
                             const InfoHash& hash,
-                             want_t want,
-                             RequestCb&& on_done,
-                             RequestExpiredCb&& on_expired);
+                             want_t want = -1,
+                             RequestCb&& on_done = {},
+                             RequestExpiredCb&& on_expired = {});
    /**
     * Send a "get" request to a given node.
     *
@@ -531,6 +531,8 @@ private:
    std::set<SockAddr> blacklist {};

    Scheduler& scheduler;
+
+    bool logIncoming_ {false};
 };

 } /* namespace net  */
@@ -42,6 +42,11 @@ 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> {
@@ -76,7 +81,7 @@ public:
    void connectivityChanged(const time_point& now) {
        grow_time = now;
        for (auto& b : *this)
-            b.time = time_point::min();
+            b.connectivityChanged();
    }

    bool onNewNode(const Sp<Node>& node, int comfirm, const time_point& now, const InfoHash& myid, net::NetworkEngine& ne);
@@ -71,7 +71,7 @@ public:
        return secureType(std::move(tmp_type));
    }

-    void registerType(const ValueType& type) {
+    void registerType(const ValueType& type) override {
        if (dht_)
            dht_->registerType(secureType(type));
    }
@@ -89,14 +89,14 @@ public:
     * If the signature can't be checked, or if the data can't be decrypted, it is not returned.
     * Public, non-signed & non-encrypted data is retransmitted as-is.
     */
-    void get(const InfoHash& id, GetCallback cb, DoneCallback donecb={}, Value::Filter&& = {}, Where&& w = {});
-    void get(const InfoHash& id, GetCallback cb, DoneCallbackSimple donecb={}, Value::Filter&& f = {}, Where&& w = {}) {
+    void get(const InfoHash& id, GetCallback cb, DoneCallback donecb={}, Value::Filter&& = {}, Where&& w = {}) override;
+    void get(const InfoHash& id, GetCallback cb, DoneCallbackSimple donecb={}, Value::Filter&& f = {}, Where&& w = {}) override {
        get(id, cb, bindDoneCb(donecb), std::forward<Value::Filter>(f), std::forward<Where>(w));
    }
-    void get(const InfoHash& key, GetCallbackSimple cb, DoneCallback donecb={}, Value::Filter&& f={}, Where&& w = {}) {
+    void get(const InfoHash& key, GetCallbackSimple cb, DoneCallback donecb={}, Value::Filter&& f={}, Where&& w = {}) override {
        get(key, bindGetCb(cb), donecb, std::forward<Value::Filter>(f), std::forward<Where>(w));
    }
-    void get(const InfoHash& key, GetCallbackSimple cb, DoneCallbackSimple donecb, Value::Filter&& f={}, Where&& w = {}) {
+    void get(const InfoHash& key, GetCallbackSimple cb, DoneCallbackSimple donecb, Value::Filter&& f={}, Where&& w = {}) override {
        get(key, bindGetCb(cb), bindDoneCb(donecb), std::forward<Value::Filter>(f), std::forward<Where>(w));
    }

@@ -148,99 +148,100 @@ public:
    /**
     * SecureDht to Dht proxy
     */
-    void shutdown(ShutdownCallback cb) {
+    void shutdown(ShutdownCallback cb) override {
        dht_->shutdown(cb);
    }
-    void dumpTables() const {
+    void dumpTables() const override {
        dht_->dumpTables();
    }
-    inline const InfoHash& getNodeId() const { return dht_->getNodeId(); }
-    std::pair<size_t, size_t> getStoreSize() const {
+    inline const InfoHash& getNodeId() const override { return dht_->getNodeId(); }
+
+    std::pair<size_t, size_t> getStoreSize() const override {
        return dht_->getStoreSize();
    }
-    std::string getStorageLog() const {
+    std::string getStorageLog() const override {
        return dht_->getStorageLog();
    }
-    std::string getStorageLog(const InfoHash& h) const {
+    std::string getStorageLog(const InfoHash& h) const override {
        return dht_->getStorageLog(h);
    }
-    void setStorageLimit(size_t limit = DEFAULT_STORAGE_LIMIT) {
+    void setStorageLimit(size_t limit = DEFAULT_STORAGE_LIMIT) override {
        dht_->setStorageLimit(limit);
    }
-    std::vector<NodeExport> exportNodes() {
+    std::vector<NodeExport> exportNodes() override {
        return dht_->exportNodes();
    }
-    std::vector<ValuesExport> exportValues() const {
+    std::vector<ValuesExport> exportValues() const override {
        return dht_->exportValues();
    }
-    void importValues(const std::vector<ValuesExport>& v) {
+    void importValues(const std::vector<ValuesExport>& v) override {
        dht_->importValues(v);
    }
-    NodeStats getNodesStats(sa_family_t af) const {
+    NodeStats getNodesStats(sa_family_t af) const override {
        return dht_->getNodesStats(af);
    }
-    std::vector<unsigned> getNodeMessageStats(bool in = false) {
+    std::vector<unsigned> getNodeMessageStats(bool in = false) override {
        return dht_->getNodeMessageStats(in);
    }
-    std::string getRoutingTablesLog(sa_family_t af) const {
+    std::string getRoutingTablesLog(sa_family_t af) const override {
        return dht_->getRoutingTablesLog(af);
    }
-    std::string getSearchesLog(sa_family_t af) const {
+    std::string getSearchesLog(sa_family_t af) const override {
        return dht_->getSearchesLog(af);
    }
-    std::string getSearchLog(const InfoHash& h, sa_family_t af = AF_UNSPEC) const {
+    std::string getSearchLog(const InfoHash& h, sa_family_t af = AF_UNSPEC) const override {
        return dht_->getSearchLog(h, af);
    }
-    std::vector<SockAddr> getPublicAddress(sa_family_t family = 0) {
+    std::vector<SockAddr> getPublicAddress(sa_family_t family = 0) override {
        return dht_->getPublicAddress(family);
    }
-    time_point periodic(const uint8_t *buf, size_t buflen, const SockAddr& sa) {
+    time_point periodic(const uint8_t *buf, size_t buflen, const SockAddr& sa) override {
        return dht_->periodic(buf, buflen, sa);
    }
-    time_point periodic(const uint8_t *buf, size_t buflen, const sockaddr* from, socklen_t fromlen) {
+    time_point periodic(const uint8_t *buf, size_t buflen, const sockaddr* from, socklen_t fromlen) override {
        return dht_->periodic(buf, buflen, from, fromlen);
    }
-    NodeStatus getStatus(sa_family_t af) const {
+    NodeStatus getStatus(sa_family_t af) const override {
        return dht_->getStatus(af);
    }
-    NodeStatus getStatus() const {
+    NodeStatus getStatus() const override {
        return dht_->getStatus();
    }
-    bool isRunning(sa_family_t af = 0) const {
+    bool isRunning(sa_family_t af = 0) const override {
        return dht_->isRunning(af);
    }
-    const ValueType& getType(ValueType::Id type_id) const {
+    const ValueType& getType(ValueType::Id type_id) const override {
        return dht_->getType(type_id);
    }
-    void insertNode(const InfoHash& id, const SockAddr& sa) {
+    void insertNode(const InfoHash& id, const SockAddr& sa) override {
        dht_->insertNode(id, sa);
    }
-    void insertNode(const InfoHash& id, const sockaddr* sa, socklen_t salen) {
+    void insertNode(const InfoHash& id, const sockaddr* sa, socklen_t salen) override {
        dht_->insertNode(id, sa, salen);
    }
-    void insertNode(const NodeExport& n) {
+    void insertNode(const NodeExport& n) override {
        dht_->insertNode(n);
    }
-    void pingNode(const sockaddr* sa, socklen_t salen, DoneCallbackSimple&& cb={}) {
+    void pingNode(const sockaddr* sa, socklen_t salen, DoneCallbackSimple&& cb={}) override {
        dht_->pingNode(sa, salen, std::move(cb));
    }
-    void query(const InfoHash& key, QueryCallback cb, DoneCallback done_cb = {}, Query&& q = {}) {
+    void query(const InfoHash& key, QueryCallback cb, DoneCallback done_cb = {}, Query&& q = {}) override {
        dht_->query(key, cb, done_cb, std::move(q));
    }
-    void query(const InfoHash& key, QueryCallback cb, DoneCallbackSimple done_cb = {}, Query&& q = {}) {
+    void query(const InfoHash& key, QueryCallback cb, DoneCallbackSimple done_cb = {}, Query&& q = {}) override {
        dht_->query(key, cb, done_cb, std::move(q));
    }
-    std::vector<Sp<Value>> getLocal(const InfoHash& key, Value::Filter f = Value::AllFilter()) const {
+    std::vector<Sp<Value>> getLocal(const InfoHash& key, Value::Filter f = Value::AllFilter()) const override {
        return dht_->getLocal(key, f);
    }
-    Sp<Value> getLocalById(const InfoHash& key, Value::Id vid) const {
+    Sp<Value> getLocalById(const InfoHash& key, Value::Id vid) const override {
        return dht_->getLocalById(key, vid);
    }
    void put(const InfoHash& key,
            Sp<Value> v,
            DoneCallback cb=nullptr,
            time_point created=time_point::max(),
-            bool permanent = false)
+            bool permanent = false) override
    {
        dht_->put(key, v, cb, created, permanent);
    }
@@ -248,7 +249,7 @@ public:
            const Sp<Value>& v,
            DoneCallbackSimple cb,
            time_point created=time_point::max(),
-            bool permanent = false)
+            bool permanent = false) override
    {
        dht_->put(key, v, cb, created, permanent);
    }
@@ -257,7 +258,7 @@ public:
            Value&& v,
            DoneCallback cb=nullptr,
            time_point created=time_point::max(),
-            bool permanent = false)
+            bool permanent = false) override
    {
        dht_->put(key, std::move(v), cb, created, permanent);
    }
@@ -265,32 +266,32 @@ public:
            Value&& v,
            DoneCallbackSimple cb,
            time_point created=time_point::max(),
-            bool permanent = false)
+            bool permanent = false) override
    {
        dht_->put(key, std::move(v), cb, created, permanent);
    }
-    std::vector<Sp<Value>> getPut(const InfoHash& h) {
+    std::vector<Sp<Value>> getPut(const InfoHash& h) override {
        return dht_->getPut(h);
    }
-    Sp<Value> getPut(const InfoHash& h, const Value::Id& vid) {
+    Sp<Value> getPut(const InfoHash& h, const Value::Id& vid) override {
        return dht_->getPut(h, vid);
    }
-    bool cancelPut(const InfoHash& h, const Value::Id& vid) {
+    bool cancelPut(const InfoHash& h, const Value::Id& vid) override {
        return dht_->cancelPut(h, vid);
    }

-    size_t listen(const InfoHash& key, ValueCallback, Value::Filter={}, Where={});
-    size_t listen(const InfoHash& key, GetCallback cb, Value::Filter = {}, Where w = {});
-    size_t listen(const InfoHash& key, GetCallbackSimple cb, Value::Filter f={}, Where w = {}) {
+    size_t listen(const InfoHash& key, ValueCallback, Value::Filter={}, Where={}) override;
+    size_t listen(const InfoHash& key, GetCallback cb, Value::Filter = {}, Where w = {}) override;
+    size_t listen(const InfoHash& key, GetCallbackSimple cb, Value::Filter f={}, Where w = {}) override {
        return listen(key, bindGetCb(cb), f, w);
    }
-    bool cancelListen(const InfoHash& h, size_t token) {
+    bool cancelListen(const InfoHash& h, size_t token) override {
        return dht_->cancelListen(h, token);
    }
-    void connectivityChanged(sa_family_t af) {
+    void connectivityChanged(sa_family_t af) override {
        dht_->connectivityChanged(af);
    }
-    void connectivityChanged() {
+    void connectivityChanged() override {
        dht_->connectivityChanged();
    }

@@ -298,7 +299,7 @@ public:
        forward_all_ = forward;
    }

-    void setPushNotificationToken(const std::string& token = "") {
+    void setPushNotificationToken(const std::string& token = "") override {
        dht_->setPushNotificationToken(token);
    }

@@ -306,7 +307,7 @@ public:
     * Call linked callback with push_notification
     * @param notification to process
     */
-    void pushNotificationReceived(const std::map<std::string, std::string>& notification) {
+    void pushNotificationReceived(const std::map<std::string, std::string>& notification) override {
        dht_->pushNotificationReceived(notification);
    }

@@ -319,7 +320,7 @@ public:
    /**
     * Only print logs related to the given InfoHash (if given), or disable filter (if zeroes).
     */
-    void setLogFilter(const InfoHash& f) {
+    void setLogFilter(const InfoHash& f) override {
        DHT_LOG.setFilter(f);
        dht_->setLogFilter(f);
    }
@@ -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 f2;
-            if (not f2) return f1;
+            if (not f1) return std::move(f1);
+            if (not f2) return std::move(f2);
            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 AllFilter();
+            if (not f1 or not f2) return {};
            return [f1,f2](const Value& v) {
                return f1(v) or f2(v);
            };
@@ -990,7 +990,7 @@ template <typename T,
 Value::Filter
 getFilterSet()
 {
-    return Value::AllFilter();
+    return {};
 }

 template <class T>
@@ -139,6 +139,12 @@ 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):
@@ -77,6 +77,9 @@ 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

@@ -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))
@@ -125,19 +125,16 @@ Dht::trySearchInsert(const Sp<Node>& node)
    bool inserted {false};

    // insert forward
-    auto it = closest;
-    while (it != srs.end()) {
+    for (auto it = closest; it != srs.end(); it++) {
        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
-    it = closest;
-    while (it != srs.begin()) {
+    for (auto it = closest; it != srs.begin();) {
        --it;
        auto& s = *it->second;
        if (s.insertNode(node, now)) {
@@ -167,8 +164,14 @@ Dht::reportedAddr(const SockAddr& addr)
 void
 Dht::onNewNode(const Sp<Node>& node, int confirm)
 {
-    if (buckets(node->getFamily()).onNewNode(node, confirm, scheduler.time(), myid, network_engine) or 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) {
        trySearchInsert(node);
+        if (wasEmpty) {
+            scheduler.edit(nextNodesConfirmation, now + std::chrono::seconds(1));
+        }
    }
 }

@@ -279,7 +282,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.w(id, sn->node->id, "[search %s] [node %s] sending %s",
+                DHT_LOG.d(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,
@@ -298,7 +301,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.w(sr->id, n->node->id, "[search %s] [node %s] sending %s",
+    DHT_LOG.d(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,
@@ -432,24 +435,20 @@ 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.w(sr->id, sn->node->id, "[search %s] [node %s] sending 'put' (vid: %d)",
+                DHT_LOG.d(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);
-                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);
+                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
+                };
            } 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] = std::make_pair(network_engine.sendRefreshValue(sn->node,
-                                                sr->id,
-                                                a.value->id,
-                                                sn->token,
-                                                onDone,
-                                                onExpired), next_refresh_time);
+                sn->acked[a.value->id] = {
+                    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);
@@ -500,7 +499,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.w(sr->id, n.node->id, "[search %s] [node %s] sending %s",
+            DHT_LOG.d(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,
@@ -524,8 +523,8 @@ Dht::searchSynchedNodeListen(const Sp<Search>& sr, SearchNode& n)
        auto list_token = l.first;
        if (n.getListenTime(query) > scheduler.time())
            continue;
-        DHT_LOG.w(sr->id, n.node->id, "[search %s] [node %s] sending 'listen'",
-                sr->id.toString().c_str(), n.node->toString().c_str());
+        // DHT_LOG.d(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()) {
@@ -537,6 +536,13 @@ 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;
@@ -554,8 +560,10 @@ 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);
                }
            },
@@ -595,9 +603,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)
@@ -656,18 +664,7 @@ Dht::searchStep(Sp<Search> sr)
            sr->setDone();
    }

-    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());*/
-    }
+    while (sr->currentlySolicitedNodeCount() < MAX_REQUESTED_SEARCH_NODES and searchSendGetValues(sr));

    if (sr->getNumberOfConsecutiveBadNodes() >= std::min(sr->nodes.size(),
                                                             static_cast<size_t>(SEARCH_MAX_BAD_NODES)))
@@ -678,10 +675,6 @@ 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) {
@@ -843,14 +836,16 @@ 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));
+    auto gcb = OpValueCache::cacheCallback(std::move(cb), [this, id, token]{
+        cancelListen(id, token);
+    });

    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, Storage(scheduler.time() + MAX_STORAGE_MAINTENANCE_EXPIRE_TIME)).first;
+        st = store.emplace(id, 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);
@@ -869,9 +864,12 @@ Dht::listen(const InfoHash& id, ValueCallback cb, Value::Filter f, Where where)
    }

    auto token4 = Dht::listenTo(id, AF_INET, gcb, filter, query);
-    auto token6 = Dht::listenTo(id, AF_INET6, 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;
+    }

-    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;
 }
@@ -916,7 +914,7 @@ struct OpStatus {

 template <typename T>
 struct GetStatus : public OpStatus {
-    std::vector<Sp<T>> values;
+    T values;
    std::vector<Sp<Node>> nodes;
 };

@@ -975,20 +973,16 @@ void doneCallbackWrapper(DoneCallback dcb, const std::vector<Sp<Node>>& nodes, G
    }
 }

-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)
+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)
 {
-    auto& op = *o;
    if (op.status.done)
        return false;
    auto newvals = add_values(values);
    if (not newvals.empty()) {
        op.status.ok = !get_cb(newvals);
-        op.values.insert(op.values.end(), newvals.begin(), newvals.end());
+        cache_values(newvals);
    }
    doneCallbackWrapper(done_cb, {}, op);
    return !op.status.ok;
@@ -1000,23 +994,26 @@ 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<Value>>();
+    auto op = std::make_shared<GetStatus<std::map<Value::Id, Sp<Value>>>>();
    auto f = filter.chain(q->where.getFilter());

-    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);
+    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);
+                }
            }
-        }
-        return newvals;
+            return newvals;
+        }, [&o](const std::vector<Sp<Value>>& newvals) {
+            for (const auto& v : newvals)
+                o.values[v->id] = v;
+        }, o);
    };
-    auto gcb = std::bind(callbackWrapper<Value, GetCallback>, getcb, donecb, _1, add_values, op);

    /* Try to answer this search locally. */
    gcb(getLocal(id, f));
@@ -1036,36 +1033,39 @@ 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<FieldValueIndex>>();
-
+    auto op = std::make_shared<GetStatus<std::vector<Sp<FieldValueIndex>>>>();
    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(),
+    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->containedIn(*f);
+                        return sf == f or f->containedIn(*sf);
                    });
-                if (lesser != op->values.end())
-                    op->values.erase(lesser);
-                newvals.push_back(f);
+                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;
+            return newvals;
+        }, [&](const std::vector<Sp<FieldValueIndex>>& fields){
+            o.values.insert(o.values.end(), fields.begin(), fields.end());
+        }, o);
    };
+
+    /* Try to answer this search locally. */
+    auto values = getLocal(id, f);
    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,6 +1173,7 @@ 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))
@@ -1223,7 +1224,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, Storage(now));
+        auto st_i = store.emplace(id, now);
        st = st_i.first;
        if (maintain_storage and st_i.second)
            scheduler.add(st->second.maintenance_time, std::bind(&Dht::dataPersistence, this, id));
@@ -1269,7 +1270,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, Storage(now)).first;
+        st = store.emplace(id, now).first;
    }
    auto node_listeners = st->second.listeners.emplace(node, std::map<size_t, Listener> {}).first;
    auto l = node_listeners->second.find(socket_id);
@@ -1388,12 +1389,6 @@ 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());
@@ -1805,7 +1800,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(), nullptr, nullptr);
+        network_engine.sendFindNode(n, id, network_engine.want());
    }

    return true;
@@ -2288,12 +2283,10 @@ 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();
@@ -2361,7 +2354,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(), vc->toString().c_str());
+                //DHT_LOG.d(hash, node.id, "[store %s] storing %s", hash.toString().c_str(), std::to_string(vc->id).c_str());
                storageStore(hash, vc, created, node.getAddr());
            } else {
                DHT_LOG.d(hash, node.id, "[store %s] rejecting storage of %s",
@@ -2397,6 +2390,23 @@ 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));
@@ -53,7 +53,6 @@ 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)),
@@ -81,6 +80,8 @@ 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();
 }
@@ -98,6 +99,7 @@ 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_();
 }

@@ -243,7 +245,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(proxy::HTTP_PROTO + serverHost_ + "/" + key.toString());
+    restbed::Uri uri(serverHost_ + "/" + key.toString());
    auto req = std::make_shared<restbed::Request>(uri);
    Value::Filter filter = w.empty() ? f : f.chain(w.getFilter());

@@ -256,45 +258,49 @@ 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>();
-        restbed::Http::async(req,
-            [=](const std::shared_ptr<restbed::Request>& req,
-                const std::shared_ptr<restbed::Response>& reply) {
-            auto code = reply->get_status_code();
+        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();

-            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_();
+                            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;
                            }
-                        } else {
-                            state->ok = false;
                        }
-                    }
-                } catch (std::runtime_error& e) { }
-            } else {
-                state->ok = false;
-            }
-        }).wait();
+                    } catch (std::runtime_error& e) { }
+                } else {
+                    state->ok = false;
+                }
+            }).wait();
+        } catch(const std::exception& e) {
+            state->ok = false;
+        }
        if (donecb) {
            std::lock_guard<std::mutex> lock(lockCallbacks);
            callbacks_.emplace_back([=](){
@@ -358,7 +364,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(proxy::HTTP_PROTO + serverHost_ + "/" + key.toString());
+    restbed::Uri uri(serverHost_ + "/" + key.toString());
    auto req = std::make_shared<restbed::Request>(uri);
    req->set_method("POST");

@@ -389,32 +395,36 @@ 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);
-        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();
+        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();

-            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))
+                    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 (...) {
                        *ok = false;
-                } catch (...) {
+                    }
+                } else {
                    *ok = false;
                }
-            } else {
-                *ok = false;
-            }
-        }).wait();
+            }).wait();
+        } catch(const std::exception& e) {
+            *ok = false;
+        }
        if (cb) {
            std::lock_guard<std::mutex> lock(lockCallbacks);
            callbacks_.emplace_back([=](){
@@ -517,7 +527,14 @@ DhtProxyClient::getProxyInfos()
    }
    statusThread_ = std::thread([this, serverHost, infoState]{
        try {
-            auto hostAndService = splitPort(serverHost);
+            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 resolved_proxies = SockAddr::resolve(hostAndService.first, hostAndService.second);
            std::vector<std::future<Sp<restbed::Response>>> reqs;
            reqs.reserve(resolved_proxies.size());
@@ -526,9 +543,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 = serverHost;
+                    server = endpointStr;
                }
-                restbed::Uri uri(proxy::HTTP_PROTO + server + "/");
+                restbed::Uri uri(protocol + server + "/");
                auto req = std::make_shared<restbed::Request>(uri);
                if (infoState->cancel)
                    return;
@@ -578,6 +595,8 @@ DhtProxyClient::getProxyInfos()
 void
 DhtProxyClient::onProxyInfos(const Json::Value& proxyInfos, sa_family_t family)
 {
+    if (isDestroying_)
+        return;
    std::lock_guard<std::mutex> l(lockCurrentProxyInfos_);
    auto oldStatus = std::max(statusIpv4_, statusIpv6_);
    auto& status = family == AF_INET ? statusIpv4_ : statusIpv6_;
@@ -611,7 +630,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) {
-            restartListeners();
+            scheduler.edit(listenerRestart, scheduler.time());
        }
        scheduler.edit(nextProxyConfirmation, scheduler.time() + std::chrono::minutes(15));
    }
@@ -649,8 +668,93 @@ 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, [&](Sp<Query> /*q*/, ValueCallback vcb){
-        return doListen(key, vcb, filter);
+    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;
    });
    return token;
 }
@@ -682,193 +786,76 @@ 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) {
-    auto settings = std::make_shared<restbed::Settings>();
+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.
+    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
+  }
+  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)) {
-                        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_();
+                        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_();
+                            }
                        }
                    }
-                }
-            } 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;
-        }
-    }, 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")) {
+                } catch (const std::exception& e) {
+                    if (not state->cancel) {
+                        DHT_LOG.w("Listen closed by the proxy server: %s", e.what());
                        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;
            }
-        } else {
-            state->ok = false;
-        }
-    }).get();
+        }, settings).get();
+    } catch (const std::exception& e) {
+        state->ok = false;
+    }
    auto& s = *state;
    if (not s.ok and not s.cancel)
        opFailed();
-#endif
-}
-
-size_t
-DhtProxyClient::doListen(const InfoHash& key, ValueCallback cb, Value::Filter filter/*, Where where*/)
-{
-    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
@@ -894,14 +881,13 @@ DhtProxyClient::doCancelListen(const InfoHash& key, size_t ltoken)
            listener.thread.join();
        }
        // UNSUBSCRIBE
-        restbed::Uri uri(proxy::HTTP_PROTO + serverHost_ + "/" + key.toString());
+        restbed::Uri uri(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"] = "";
@@ -909,8 +895,11 @@ 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()));
-
-        restbed::Http::async(req, [](const std::shared_ptr<restbed::Request>&, const std::shared_ptr<restbed::Response>&){});
+        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());
+        }
    } else {
        // Just stop the request
        if (listener.thread.joinable()) {
@@ -951,6 +940,8 @@ DhtProxyClient::getConnectivityStatus()
 void
 DhtProxyClient::restartListeners()
 {
+    if (isDestroying_) return;
+    std::lock_guard<std::mutex> lock(searchLock_);
    DHT_LOG.d("Refresh permanent puts");
    for (auto& search : searches_) {
        for (auto& put : search.second.puts) {
@@ -974,7 +965,6 @@ DhtProxyClient::restartListeners()
        return;
    }
    DHT_LOG.d("Restarting listeners");
-    std::lock_guard<std::mutex> lock(searchLock_);
    for (auto& search: searches_) {
        for (auto& l: search.second.listeners) {
            auto& listener = l.second;
@@ -990,7 +980,7 @@ DhtProxyClient::restartListeners()
            state->ok = true;
            auto filter = listener.filter;
            auto cb = listener.cb;
-            restbed::Uri uri(proxy::HTTP_PROTO + serverHost_ + "/" + search.first.toString());
+            restbed::Uri uri(serverHost_ + "/" + search.first.toString());
            auto req = std::make_shared<restbed::Request>(uri);
            req->set_method("LISTEN");
            listener.req = req;
@@ -1031,10 +1021,12 @@ DhtProxyClient::pushNotificationReceived(const std::map<std::string, std::string
                    resubscribe(key, list.second);
            }
        } else {
-            auto token = std::stoull(notification.at("token"));
+            auto keyidx = notification.find("key");
+            InfoHash key(keyidx->second);
+
            for (auto& search: searches_) {
                for (auto& list : search.second.listeners) {
-                    if (*list.second.pushNotifToken != token or list.second.state->cancel)
+                    if (search.first != key || 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;
@@ -1060,21 +1052,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(proxy::HTTP_PROTO + serverHost_ + "/" + key.toString());
+    restbed::Uri uri(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);
-    listener.thread = std::thread([this, req, pushNotifToken, state]() {
-        sendSubscribe(req, pushNotifToken, state);
+    auto vcb = listener.cb;
+    auto filter = listener.filter;
+    listener.thread = std::thread([this, req, vcb, filter, state]() {
+        sendListen(req, vcb, filter, state, ListenMethod::RESUBSCRIBE);
    });
 #endif
 }
@@ -1094,17 +1086,18 @@ DhtProxyClient::getPushRequest(Json::Value& body) const
 }

 void
-DhtProxyClient::fillBodyToGetToken(std::shared_ptr<restbed::Request> req, unsigned token)
+DhtProxyClient::fillBody(std::shared_ptr<restbed::Request> req, bool resubscribe)
 {
    // Fill body with
    // {
    //   "key":"device_key",
-    //   "token": xxx
    // }
    Json::Value body;
    getPushRequest(body);
-    if (token > 0)
-        body["token"] = token;
+    if (!resubscribe) {
+        // This is the first listen, we want to retrieve previous values.
+        body["previous_values"] = true;
+    }
    Json::StreamWriterBuilder wbuilder;
    wbuilder["commentStyle"] = "None";
    wbuilder["indentation"] = "";
@@ -370,7 +370,6 @@ 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;
@@ -409,12 +408,11 @@ 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 << " token:" << tokenFromReq << " client:" << clientId << std::endl;
+                std::cout << "Subscribe " << infoHash << " client:" << clientId << std::endl;

                {
                    std::lock(schedulerLock_, lockListener_);
@@ -428,10 +426,40 @@ 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->close(restbed::OK, "{\"token\": " + std::to_string(tokenFromReq) + "}\n");
+                            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");
+                                }
+                            }
                            schedulerCv_.notify_one();
                            return;
                        }
@@ -444,15 +472,12 @@ 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, token, infoHash, pushToken, isAndroid, clientId](std::vector<std::shared_ptr<Value>> /*value*/) {
+                        [this, 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;
                        }
@@ -463,18 +488,17 @@ DhtProxyServer::subscribe(const std::shared_ptr<restbed::Session>& session)
                        }
                    );
                    listener.expireNotifyJob = scheduler_.add(timeout - proxy::OP_MARGIN,
-                        [this, token, infoHash, pushToken, isAndroid, clientId] {
+                        [this, 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, "{\"token\": " + std::to_string(tokenFromReq) + "}\n");
+                s->close(restbed::OK, "{}\n");
            } catch (...) {
                s->close(restbed::INTERNAL_SERVER_ERROR, "{\"err\":\"Internal server error\"}");
            }
@@ -549,6 +573,8 @@ 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");
@@ -37,7 +37,7 @@
 #include <winsock2.h>
 #include <ws2tcpip.h>
 #define close(x) closesocket(x)
-#define write(...) _write(__VA_ARGS__)
+#define write(s, b, f) send(s, b, (int)strlen(b), 0)
 #endif

 namespace dht {
@@ -460,6 +460,7 @@ 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));
    }
@@ -520,7 +521,7 @@ DhtRunner::startNetwork(const SockAddr sin4, const SockAddr sin6)
 #ifndef _WIN32
    auto status = pipe(stopfds);
    if (status == -1) {
-        throw DhtException("Can't open pipe");
+        throw DhtException(std::string("Can't open pipe: ") + strerror(errno));
    }
 #else
    udpPipe(stopfds);
@@ -532,15 +533,29 @@ DhtRunner::startNetwork(const SockAddr sin4, const SockAddr sin6)
    s6 = -1;

    bound4 = {};
-    if (sin4)
-        s4 = bindSocket(sin4, bound4);
+    if (sin4) {
+        try {
+            s4 = bindSocket(sin4, bound4);
+        } catch (const DhtException& e) {
+            std::cerr << "Can't bind inet socket: " << e.what() << std::endl;
+        }
+    }

 #if 1
    bound6 = {};
-    if (sin6)
-        s6 = bindSocket(sin6, bound6);
+    if (sin6) {
+        try {
+            s6 = bindSocket(sin6, bound6);
+        } catch (const DhtException& e) {
+            std::cerr << "Can't bind inet6 socket: " << e.what() << std::endl;
+        }
+    }
 #endif

+    if (s4 == -1 && s6 == -1) {
+        throw DhtException("Can't bind socket");
+    }
+
    running_network = true;
    rcv_thread = std::thread([this, stop_readfd]() {
        try {
@@ -48,11 +48,4 @@ struct LocalListener {
    ValueCallback get_cb;
 };

-
-struct SearchListener {
-    Sp<Query> query;
-    Value::Filter filter;
-    ValueCallback get_cb;
-};
-
 }
@@ -353,9 +353,8 @@ NetworkEngine::rateLimit(const SockAddr& addr)
        limiter_maintenance = 0;
    }

-    auto it = address_rate_limiter.emplace(addr, IpLimiter{});
    // invoke per IP, then global rate limiter
-    return it.first->second.limit(now) and rate_limiter.limit(now);
+    return address_rate_limiter[addr].limit(now) and rate_limiter.limit(now);
 }

 bool
@@ -434,7 +433,8 @@ 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()) {
-            DHT_LOG.d("Can't find partial message");
+            if (logIncoming_)
+                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
-        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));
+        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));
        } else
            DHT_LOG.e("Partial message with given TID already exists");
    }
@@ -544,19 +544,21 @@ NetworkEngine::process(std::unique_ptr<ParsedMessage>&& msg, const SockAddr& fro
                req->reply_time = time_point::min();
                onError(req, DhtProtocolException {msg->error_code});
            } else {
-                DHT_LOG.w(msg->id, "[node %s %s] received unknown error message %u",
+                if (logIncoming_)
+                    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 */
-                if (req->getType() == MessageType::AnnounceValue or req->getType() == MessageType::Listen)
-                    req->node->authSuccess();
-                req->reply_time = scheduler.time();
+                auto& r = *req;
+                if (r.getType() == MessageType::AnnounceValue or r.getType() == MessageType::Listen)
+                    r.node->authSuccess();
+                r.reply_time = scheduler.time();

                deserializeNodes(*msg, from);
-                req->setDone(std::move(*msg));
+                r.setDone(std::move(*msg));
                break;
            } else { /* request socket data */
                deserializeNodes(*msg, from);
@@ -574,7 +576,8 @@ NetworkEngine::process(std::unique_ptr<ParsedMessage>&& msg, const SockAddr& fro
            switch (msg->type) {
            case MessageType::Ping:
                ++in_stats.ping;
-                DHT_LOG.d(node->id, "[node %s] sending pong", node->toString().c_str());
+                if (logIncoming_)
+                    DHT_LOG.d(node->id, "[node %s] sending pong", node->toString().c_str());
                onPing(node);
                sendPong(from, msg->tid);
                break;
@@ -595,7 +598,8 @@ NetworkEngine::process(std::unique_ptr<ParsedMessage>&& msg, const SockAddr& fro
                break;
            }
            case MessageType::AnnounceValue: {
-                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());
+                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());
                ++in_stats.put;
                onAnnounce(node, msg->info_hash, msg->token, msg->values, msg->created);

@@ -608,13 +612,15 @@ NetworkEngine::process(std::unique_ptr<ParsedMessage>&& msg, const SockAddr& fro
                break;
            }
            case MessageType::Refresh:
-                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());
+                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());
                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: {
-                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());
+                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());
                ++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);
@@ -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, OpCacheValueStorage{v});
+        auto viop = values.emplace(v->id, 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(Value::Filter& filter) const {
+OpValueCache::get(const Value::Filter& filter) const {
    std::vector<Sp<Value>> ret;
    if (not filter)
        ret.reserve(values.size());
@@ -108,33 +108,57 @@ OpCache::getExpiration() const {
    return lastRemoved + EXPIRATION;
 }

-size_t
-SearchCache::listen(ValueCallback get_cb, Sp<Query> q, Value::Filter filter, std::function<size_t(Sp<Query>, ValueCallback)> onListen)
+SearchCache::OpMap::iterator
+SearchCache::getOp(const Sp<Query>& q)
 {
    // find exact match
    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())
+        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)
+{
+    // find exact match
+    auto op = getOp(q);
    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_++;
-    op->second->addListener(token, get_cb, q, filter);
-    return token;
+    return op->second->addListener(token, get_cb, q, filter) ? token : 0;
 }

 bool
@@ -177,8 +201,22 @@ 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(Value::Filter& filter) const {
+SearchCache::get(const Value::Filter& filter) const {
    if (ops.size() == 1)
        return ops.begin()->second->get(filter);
    std::map<Value::Id, Sp<Value>> c;
@@ -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,10 +37,13 @@ public:
        o.callback = {};
    }

-    static ValueCallback cacheCallback(ValueCallback&& cb) {
+    static ValueCallback cacheCallback(ValueCallback&& cb, std::function<void()>&& onCancel) {
        auto cache = std::make_shared<OpValueCache>(std::forward<ValueCallback>(cb));
-        return [cache](const std::vector<Sp<Value>>& vals, bool expired){
-            return cache->onValue(vals, expired);
+        return [cache, onCancel](const std::vector<Sp<Value>>& vals, bool expired){
+            auto ret = cache->onValue(vals, expired);
+            if (not ret)
+                onCancel();
+            return ret;
        };
    }

@@ -54,7 +57,18 @@ public:
    bool onValuesAdded(const std::vector<Sp<Value>>& vals);
    bool onValuesExpired(const std::vector<Sp<Value>>& vals);

-    std::vector<Sp<Value>> get(Value::Filter& filter) const;
+    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;
    Sp<Value> get(Value::Id id) const;
    std::vector<Sp<Value>> getValues() const;

@@ -62,6 +76,8 @@ 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;
 };
@@ -80,13 +96,23 @@ 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);

-    void addListener(size_t token, ValueCallback cb, Sp<Query> q, Value::Filter filter) {
+    bool addListener(size_t token, ValueCallback cb, Sp<Query> q, Value::Filter filter) {
        listeners.emplace(token, LocalListener{q, filter, cb});
-        cb(cache.get(filter), false);
+        auto cached = cache.get(filter);
+        if (not cached.empty()) {
+            auto ret = cb(cached, false);
+            if (not ret)
+                listeners.erase(token);
+            return ret;
+        }
+        return true;
    }

    bool removeListener(size_t token, const time_point& now) {
@@ -102,7 +128,7 @@ public:
        return listeners.empty();
    }

-    std::vector<Sp<Value>> get(Value::Filter& filter) const {
+    std::vector<Sp<Value>> get(const Value::Filter& filter) const {
        return cache.get(filter);
    }

@@ -110,6 +136,10 @@ 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);
    }
@@ -130,7 +160,9 @@ class SearchCache {
 public:
    SearchCache() {}
    SearchCache(SearchCache&&) = default;
-    size_t listen(ValueCallback get_cb, Sp<Query> q, Value::Filter filter, std::function<size_t(Sp<Query>, ValueCallback)> onListen);
+
+    using OnListen = std::function<size_t(Sp<Query>, ValueCallback, SyncCallback)>;
+    size_t listen(ValueCallback get_cb, Sp<Query> q, Value::Filter filter, OnListen onListen);

    bool cancelListen(size_t gtoken, const time_point& now);
    void cancelAll(std::function<void(size_t)> onCancel);
@@ -140,14 +172,19 @@ public:
        return nextExpiration_;
    }

-    std::vector<Sp<Value>> get(Value::Filter& filter) const;
+    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;
    Sp<Value> get(Value::Id id) const;

 private:
    SearchCache(const SearchCache&) = delete;
    SearchCache& operator=(const SearchCache&) = delete;

-    std::map<Sp<Query>, std::unique_ptr<OpCache>> ops {};
+    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;
+
    size_t nextToken_ {1};
    time_point nextExpiration_ {time_point::max()};
 };
@@ -65,7 +65,8 @@ struct Dht::SearchNode {
        ValueCache cache;
        Sp<Scheduler::Job> cacheExpirationJob {};
        Sp<net::Request> req {};
-        CachedListenStatus(ValueStateCallback&& cb) : cache(std::forward<ValueStateCallback>(cb)) {}
+        CachedListenStatus(ValueStateCallback&& cb, SyncCallback&& scb)
+         : cache(std::forward<ValueStateCallback>(cb), std::forward<SyncCallback>(scb)) {}
        CachedListenStatus(CachedListenStatus&&) = default;
        CachedListenStatus(const CachedListenStatus&) = delete;
        CachedListenStatus& operator=(const CachedListenStatus&) = delete;
@@ -225,6 +226,13 @@ 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()) {
@@ -377,6 +385,12 @@ 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;

@@ -477,21 +491,20 @@ 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) {
-            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);
+            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);
+            }
        }
    }

    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){
+        return cache.listen(cb, q, f, [&](const Sp<Query>& q, ValueCallback vcb, SyncCallback scb){
            done = false;
            auto token = ++listener_token;
-            listeners.emplace(token, SearchListener{q, f, vcb});
+            listeners.emplace(token, SearchListener{q, f, vcb, scb});
            scheduler.edit(nextSearchStep, scheduler.time());
            return token;
        });
@@ -284,6 +284,8 @@ 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))
@@ -301,6 +303,8 @@ 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))
@@ -364,7 +368,8 @@ 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)
+        Value::IdFilter(val->id),
+        std::move(Where().id(val->id))
    );
 }

@@ -54,12 +54,14 @@ 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 {
-                InfoHash h = crypto::Certificate(v.data).getPublicKey().getId();
+                auto h = crypto::Certificate(v.data).getPublicKey().getLongId();
                s << " with ID " << h;
            } catch (const std::exception& e) {
                s << " (invalid)";
            }
+#endif
        } else {
            s << "Data (type: " << v.type << " ): ";
            s << std::hex;
@@ -286,7 +288,7 @@ FieldValue::getLocalFilter() const
        case Value::Field::UserType:
            return Value::UserTypeFilter(std::string {blobValue.begin(), blobValue.end()});
        default:
-            return Value::AllFilter();
+            return {};
    }
 }

@@ -495,8 +497,7 @@ template <typename T>
 bool subset(std::vector<T> fds, std::vector<T> qfds)
 {
    for (auto& fd : fds) {
-        auto correspondance = std::find_if(qfds.begin(), qfds.end(), [&fd](T& _vfd) { return fd == _vfd; });
-        if (correspondance == qfds.end())
+        if (std::find_if(qfds.begin(), qfds.end(), [&fd](T& _vfd) { return fd == _vfd; }) == qfds.end())
            return false;
    }
    return true;
@@ -504,10 +505,9 @@ bool subset(std::vector<T> fds, std::vector<T> qfds)

 bool Select::isSatisfiedBy(const Select& os) const {
    /* empty, means all values are selected. */
-    if (fieldSelection_.empty() and not os.fieldSelection_.empty())
-        return false;
-    else
-        return subset(fieldSelection_, os.fieldSelection_);
+    return fieldSelection_.empty() ?
+        os.fieldSelection_.empty() :
+        subset(fieldSelection_, os.fieldSelection_);
 }

 bool Where::isSatisfiedBy(const Where& ow) const {
@@ -22,19 +22,32 @@
 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) : callback(std::forward<ValueStateCallback>(cb)) {}
-    ValueCache(ValueCache&& o) : values(std::move(o.values)), callback(std::move(o.callback)) {
+    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)) {
        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() {
@@ -120,6 +133,15 @@ 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;
@@ -141,6 +163,8 @@ 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;
@@ -77,7 +77,6 @@ DhtProxyTester::testGetPut() {
    CPPUNIT_ASSERT(vals.front()->data == val_data);
 }

-
 void
 DhtProxyTester::testListen() {
    bool done = false;
@@ -108,7 +107,7 @@ DhtProxyTester::testListen() {

    cv.wait_for(lk, std::chrono::seconds(10), [&]{ return done; });
    done = false;
-    // Here values should contains 2 values
+    // Here values should contains 1 values
    CPPUNIT_ASSERT_EQUAL(static_cast<int>(values.size()), 1);
    CPPUNIT_ASSERT(values.front() == firstVal_data);

@@ -118,9 +117,63 @@ 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 3 values
+    // Here values should contains 2 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
@@ -32,6 +32,7 @@ class DhtProxyTester : public CppUnit::TestFixture {
    CPPUNIT_TEST_SUITE(DhtProxyTester);
    CPPUNIT_TEST(testGetPut);
    CPPUNIT_TEST(testListen);
+    CPPUNIT_TEST(testResubscribeGetValues);
    CPPUNIT_TEST_SUITE_END();

 public:
@@ -39,19 +40,25 @@ 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();
+   void testListen();
+
+   /**
+    * When a proxy redo a subscribe on the proxy
+    * it should retrieve existant values
+    */
+   void testResubscribeGetValues();

 private:
    dht::DhtRunner nodePeer {};
@@ -50,15 +50,56 @@ DhtRunnerTester::testGetPut() {
    auto key = dht::InfoHash::get("123");
    dht::Value val {"hey"};
    auto val_data = val.data;
-    node1.put(key, std::move(val));
-    auto vals = node2.get(key).get();
+    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();
    CPPUNIT_ASSERT(not vals.empty());
    CPPUNIT_ASSERT(vals.front()->data == val_data);
 }

 void
 DhtRunnerTester::testListen() {
-    // TODO
+    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());
 }

 }  // namespace test
@@ -64,6 +64,15 @@ 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());
Author	SHA1	Message	Date
Adrien Béraud	ba2e179abe	dht: optimize get	2019-02-19 01:59:59 -05:00
Adrien Béraud	a5d29c904c	op cache: optimize cache insertion	2019-02-19 01:59:59 -05:00
Adrien Béraud	989f32eb56	filter: cleanup and optimize	2019-02-19 01:59:59 -05:00
Adrien Béraud	b66fa73c56	test: include cppunit headers	2019-02-19 01:59:59 -05:00
Adrien Béraud	1ca181db1d	value: simplify expression	2019-02-19 01:59:59 -05:00
Adrien Béraud	1f3bf697c3	search: use syncStatus for get	2019-02-19 01:59:59 -05:00
Adrien Béraud	a1f8cc906c	proxy client: use SyncCallback	2019-02-19 01:59:59 -05:00
Adrien Béraud	35c6a57526	dht: add synced status	2019-02-19 01:59:59 -05:00
Adrien Béraud	39e69f813c	network engine: cleanup	2019-02-19 01:59:59 -05:00
Sébastien Blin	8045c2cbf2	dhtrunner: avoid startNetwork failure if only one binding is failing	2019-02-19 01:44:47 -05:00
Sébastien Blin	79a2f4280f	proxy: resolve url with protocol and port	2019-02-14 11:04:11 -05:00
Sébastien Blin	769ce45740	proxy_client: support prototype in serverHost	2019-01-22 07:56:27 -08:00
Sébastien Blin	75a14d3608	proxy: only get old values at first subscribe	2019-01-18 22:53:24 -08:00
Sébastien Blin	d523e9f259	tests: test retrieve values when resubscribe	2019-01-14 10:53:08 -08:00
Sébastien Blin	3a3933e6d5	dhtproxy: remove sendSubscribe	2019-01-14 10:53:08 -08:00
Sébastien Blin	70f09ded3f	dhtproxy: avoid notification after subscribe to retrieve values	2019-01-14 10:53:08 -08:00
Sébastien Blin	824a1bf5fa	proxy_server: return old values when resubscribe	2019-01-14 10:53:08 -08:00
Sébastien Blin	e5b56cc0a5	dht: link to tellListenerRefreshed When a permanent put is refreshed, the node has to tell to its remote listeners that the value is refreshed.	2019-01-09 13:34:51 -08:00
Sébastien Blin	36cf70f2a7	proxy: remove useless token for push listeners Since we authentificate listeners with the clientId and the hash and we use only one listener by hash/clientId, we doesn't need to store a token always equals to 0. Note: also avoid to get multiple hashes for one push notification.	2018-12-30 13:35:18 -05:00
Adrien Béraud	459c1344aa	update to version 1.8.1	2018-12-28 11:38:15 -05:00
Adrien Béraud	c0e376e15a	dht: reduce log verbosity	2018-12-28 01:41:10 -05:00
Adrien Béraud	71b9672b96	value: don't print certificate id in toString by default	2018-12-28 01:38:45 -05:00
Adrien Béraud	086f37a71a	op cache: remove listener if callback returned false	2018-12-28 01:38:02 -05:00
Adrien Béraud	87e76a5e4e	tests: reverse nodes	2018-12-26 13:03:36 -05:00
Adrien Béraud	7f0a5dfda3	dht: increase max number of searches and hashes	2018-12-26 10:28:55 -05:00
Adrien Béraud	6d7f8004bf	connectivity change: reset routing table instead of searches	2018-12-26 10:28:30 -05:00
Adrien Béraud	8168c39b63	dht: cleanup	2018-12-26 01:22:37 -05:00
Adrien Béraud	d0fdbe5340	dhtrunner: print proper message on bind error	2018-12-26 00:58:43 -05:00
Adrien Béraud	c7604fbb8d	tests: wait for put before performing get	2018-12-26 00:56:49 -05:00
Adrien Béraud	56773baa98	unit tests: add listen test	2018-12-24 19:09:32 -05:00
Adrien Béraud	e6cf67c575	dht: clear local listener if callback returns false	2018-12-24 18:37:06 -05:00
Adrien Béraud	00ebc5e2bf	op cache: check callback return value	2018-12-24 18:36:29 -05:00
Adrien Béraud	ccef94c11b	Revert "storage: make non-movable" This reverts commit `7ebad7d3c5`.	2018-12-24 17:16:15 -05:00
Adrien Béraud	0066be49b9	python: add more bindings for SockAddr	2018-12-24 16:55:25 -05:00
Adrien Béraud	592ed4ee90	op cache: don't call callback if no value	2018-12-24 15:50:41 -05:00
Adrien Béraud	353a6f9db6	op cache: add onCancel callback	2018-12-24 15:41:17 -05:00
Adrien Béraud	7ebad7d3c5	storage: make non-movable	2018-12-24 15:28:03 -05:00
Adrien Béraud	14f6e6e5f3	proxy client: catch exceptions in restbed::Http::async	2018-12-22 23:45:37 -05:00
Adrien Béraud	9ce5b30931	securedht: use query to check for existing value	2018-12-22 11:42:45 -05:00
Adrien Béraud	92979fa96a	dhtchat: add logging support	2018-12-21 19:55:27 -05:00
Adrien Béraud	0c1e7d2763	dht: run node confirmation if bucket was empty	2018-12-19 14:07:55 -05:00
Adrien Béraud	14a744f6cd	dht: log cleanup	2018-12-19 14:07:55 -05:00
Adrien Béraud	3b26684a87	network engine: add default values for sendFindNodes	2018-12-19 14:07:55 -05:00
Sébastien Blin	0a764ee440	Merge pull request #361 from savoirfairelinux/schedule_restart proxy client: schedule listener restart	2018-12-18 16:07:39 -05:00
Adrien Béraud	7dda5a3d5a	proxy client: schedule listener restart	2018-12-18 10:10:14 -05:00
Adrien Béraud	306d573223	proxy client: fix race condition	2018-12-17 16:41:00 -05:00
Sébastien Blin	20f4355b73	Merge pull request #356 from savoirfairelinux/build securedht: add override	2018-12-11 14:52:07 -05:00
Adrien Béraud	a2d7cd56a8	Merge pull request #358 from atraczyk/master replace _write with send when building for windows	2018-12-11 14:48:26 -05:00
Andreas Traczyk	5cd53f3624	replace _write with send when building for windows	2018-12-11 14:37:14 -05:00
Adrien Béraud	32eea55f10	autotools: don't build doc if doxygen not found	2018-12-11 14:04:35 -05:00
Adrien Béraud	f2f913f93f	securedht: add override for relevant methods	2018-12-11 14:04:35 -05:00