proxy: resolve url with protocol and port

When a permanent put is refreshed, the node has to tell to its
remote listeners that the value is refreshed.

CMakeLists.txt

@@ -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}")

configure.ac

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

include/opendht/dht.h

@@ -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};
 

include/opendht/dht_proxy_client.h

@@ -280,8 +280,15 @@ private:
     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 +358,8 @@ private:
      */
     void confirmProxy();
     Sp<Scheduler::Job> nextProxyConfirmation {};
+    Sp<Scheduler::Job> listenerRestart {};
+
     /**
      * Relaunch LISTEN requests if the client disconnect/reconnect.
      */
@@ -371,7 +380,7 @@ 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
 

include/opendht/network_engine.h

@@ -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  */

include/opendht/routing_table.h

@@ -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);

python/opendht.pyx

@@ -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):

python/opendht_cpp.pxd

@@ -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
 

python/tools/pingpong.py

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

src/dht.cpp

@@ -125,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,7 +435,7 @@ 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,
@@ -500,7 +503,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 +527,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()) {
@@ -595,9 +598,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 +659,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 +670,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 +831,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 +859,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;
 }
@@ -1223,7 +1216,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 +1262,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 +1381,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 +1792,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;
@@ -2292,8 +2279,8 @@ 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 +2348,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 +2384,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));

src/dht_proxy_client.cpp

@@ -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;
@@ -613,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));
     }
@@ -684,105 +701,83 @@ 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());
+    restbed::Uri uri(serverHost_ + "/" + key.toString());
     std::lock_guard<std::mutex> lock(searchLock_);
     auto search = searches_.find(key);
     if (search == searches_.end()) {
@@ -843,9 +838,7 @@ DhtProxyClient::doListen(const InfoHash& key, ValueCallback cb, Value::Filter fi
         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()) {
@@ -862,14 +855,12 @@ DhtProxyClient::doListen(const InfoHash& key, ValueCallback cb, Value::Filter fi
                 }
             }
         });
-        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);
-        });
     }
+    l->second.thread = std::thread([this, req, vcb, filter, state]() {
+        sendListen(req, vcb, filter, state,
+                deviceKey_.empty() ? ListenMethod::LISTEN
+                                   : ListenMethod::SUBSCRIBE);
+    });
     return token;
 }
 
@@ -896,14 +887,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"] = "";
@@ -911,8 +901,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()) {
@@ -993,7 +986,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;
@@ -1034,10 +1027,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;
@@ -1063,21 +1058,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
 }
@@ -1097,17 +1092,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"] = "";

src/dht_proxy_server.cpp

@@ -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");

src/dhtrunner.cpp

@@ -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));
     }

src/network_engine.cpp

@@ -434,7 +434,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;
         }
@@ -544,19 +545,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 +577,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 +599,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 +613,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);

src/op_cache.cpp

@@ -133,8 +133,7 @@ SearchCache::listen(ValueCallback get_cb, Sp<Query> q, Value::Filter filter, std
     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

src/op_cache.h

@@ -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;
         };
     }
 
@@ -84,9 +87,16 @@ public:
     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) {

src/securedht.cpp

@@ -364,7 +364,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))
     );
 }
 

src/value.cpp

@@ -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;

tests/dhtproxytester.cpp

@@ -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

tests/dhtproxytester.h

@@ -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 {};

tests/dhtrunnertester.cpp

@@ -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

tools/dhtchat.cpp

@@ -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());