update to version 1.8.1

This reverts commit 7ebad7d3c5.

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

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

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",

src/dht_proxy_client.cpp

@@ -98,6 +98,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_();
 }
 
@@ -256,45 +257,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([=](){
@@ -389,32 +394,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([=](){
@@ -613,7 +622,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));
     }
@@ -691,47 +700,51 @@ DhtProxyClient::sendListen(const std::shared_ptr<restbed::Request>& req, const V
     std::chrono::milliseconds timeout(std::numeric_limits<int>::max());
     settings->set_connection_timeout(timeout); // Avoid the client to close the socket after 5 seconds.
     req->set_method("LISTEN");
-    restbed::Http::async(req,
-        [this, filter, cb, state](const std::shared_ptr<restbed::Request>& req,
-                                  const std::shared_ptr<restbed::Response>& reply) {
-        auto code = reply->get_status_code();
-        if (code == 200) {
-            try {
-                while (restbed::Http::is_open(req) and not state->cancel) {
-                    restbed::Http::fetch("\n", reply);
-                    if (state->cancel)
-                        break;
-                    std::string body;
-                    reply->get_body(body);
-                    reply->set_body(""); // Reset the body for the next fetch
+    try {
+        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)) {
+                            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;
+                    }
                 }
-            } 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;
             }
-        } else {
-            state->ok = false;
-        }
-    }, settings).get();
+        }, settings).get();
+    } catch (const std::exception& e) {
+        state->ok = false;
+    }
     auto& s = *state;
     if (not s.ok and not s.cancel)
         opFailed();
@@ -741,37 +754,41 @@ 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);
+    try {
+        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")) {
+                    std::string err;
+                    Json::Value json;
+                    Json::CharReaderBuilder rbuilder;
+                    auto reader = std::unique_ptr<Json::CharReader>(rbuilder.newCharReader());
+                    if (reader->parse(body.data(), body.data() + body.size(), &json, &err)) {
+                        if (not json.isMember("token")) {
+                            state->ok = false;
+                            return;
+                        }
+                        *token = unpackId(json, "token");
+                    }
+                } catch (const std::exception& e) {
+                    if (not state->cancel) {
+                        DHT_LOG.e("sendSubscribe: error: %s", e.what());
                         state->ok = false;
-                        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();
+        }).get();
+    } catch(const std::exception& e) {
+        state->ok = false;
+    }
     auto& s = *state;
     if (not s.ok and not s.cancel)
         opFailed();
@@ -911,8 +928,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()) {

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