wip

proxy server, fix push listeners and avoid flooding nodes

CMakeLists.txt

@@ -1,7 +1,7 @@
 cmake_minimum_required (VERSION 3.1)
 project (opendht)
 set (opendht_VERSION_MAJOR 1)
-set (opendht_VERSION_MINOR 7.0)
+set (opendht_VERSION_MINOR 7.1)
 set (opendht_VERSION ${opendht_VERSION_MAJOR}.${opendht_VERSION_MINOR})
 set (PACKAGE_VERSION ${opendht_VERSION})
 set (VERSION "${opendht_VERSION}")
@@ -121,6 +121,7 @@ list (APPEND opendht_SOURCES
     src/node.cpp
     src/value.cpp
     src/dht.cpp
+    src/op_cache.cpp
     src/storage.h
     src/listener.h
     src/search.h
@@ -203,6 +204,9 @@ else ()
 endif ()
 
 if (OPENDHT_PROXY_SERVER OR OPENDHT_PROXY_CLIENT)
+  list (APPEND opendht_HEADERS
+    include/opendht/proxy.h
+  )
   list (APPEND opendht_SOURCES
     src/base64.h
     src/base64.cpp

configure.ac

@@ -1,7 +1,7 @@
 dnl define macros
 m4_define([opendht_major_version], 1)
 m4_define([opendht_minor_version], 7)
-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_proxy_server.h

@@ -213,7 +213,13 @@ private:
      */
     void sendPushNotification(const std::string& key, const Json::Value& json, bool isAndroid) const;
 
-    void cancelPushListen(const std::string& pushToken, const InfoHash& key, proxy::ListenToken token);
+    /**
+     * Remove a push listener between a client and a hash
+     * @param pushToken
+     * @param key
+     * @param clientId
+     */
+    void cancelPushListen(const std::string& pushToken, const InfoHash& key, const std::string& clientId);
 
 
 #endif //OPENDHT_PUSH_NOTIFICATIONS

include/opendht/dhtrunner.h

@@ -407,7 +407,7 @@ public:
     /**
      * Insert a push notification to process for OpenDHT
      */
-    void pushNotificationReceived(const std::map<std::string, std::string>& data) const;
+    void pushNotificationReceived(const std::map<std::string, std::string>& data);
 
     /* Proxy server mothods */
     void forwardAllMessages(bool forward);

include/opendht/proxy.h

@@ -22,8 +22,8 @@
 namespace dht {
 namespace proxy {
 
-constexpr const std::chrono::seconds OP_TIMEOUT {1 * 60 * 60}; // onw hour
-constexpr const std::chrono::seconds OP_MARGIN {5 * 60}; // 5 minutes
+constexpr const std::chrono::minutes OP_TIMEOUT {1 * 10}; // one hour
+constexpr const std::chrono::minutes OP_MARGIN {5}; // 5 minutes
 constexpr const char* const HTTP_PROTO {"http://"};
 using ListenToken = uint64_t;
 

src/Makefile.am

@@ -11,6 +11,7 @@ libopendht_la_SOURCES  = \
         search.h \
         value_cache.h \
         op_cache.h \
+        op_cache.cpp \
         net.h \
         parsed_message.h \
         node_cache.cpp \
@@ -55,7 +56,7 @@ nobase_include_HEADERS = \
         ../include/opendht/rng.h
 
 if ENABLE_PROXY_SERVER
-libopendht_la_SOURCES += base64.h base64.cpp dht_proxy_server.cpp
+libopendht_la_SOURCES += dht_proxy_server.cpp
 nobase_include_HEADERS += ../include/opendht/dht_proxy_server.h
 endif
 
@@ -64,6 +65,11 @@ libopendht_la_SOURCES += dht_proxy_client.cpp
 nobase_include_HEADERS += ../include/opendht/dht_proxy_client.h ../include/opendht/dht_interface.h
 endif
 
+if PROXY_CLIENT_OR_SERVER
+libopendht_la_SOURCES += base64.h base64.cpp
+nobase_include_HEADERS += ../include/opendht/proxy.h
+endif
+
 if ENABLE_INDEXATION
 libopendht_la_SOURCES += indexation/pht.cpp
 nobase_include_HEADERS += ../include/opendht/indexation/pht.h

src/dht.cpp

@@ -874,10 +874,10 @@ Dht::cancelListen(const InfoHash& id, size_t token)
     if (st != store.end() && tokenlocal)
         st->second.local_listeners.erase(tokenlocal);
 
-    auto searches_cancel_listen = [&id](std::map<InfoHash, Sp<Search>>& srs, size_t token) {
+    auto searches_cancel_listen = [this,&id](std::map<InfoHash, Sp<Search>>& srs, size_t token) {
         auto srp = srs.find(id);
         if (srp != srs.end() and token)
-            srp->second->cancelListen(token);
+            srp->second->cancelListen(token, scheduler);
     };
     searches_cancel_listen(searches4, std::get<1>(it->second));
     searches_cancel_listen(searches6, std::get<2>(it->second));

src/dht_proxy_client.cpp

@@ -64,6 +64,7 @@ struct PermanentPut {
 
 struct DhtProxyClient::ProxySearch {
     SearchCache ops {};
+    Sp<Scheduler::Job> opExpirationJob;
     std::map<size_t, Listener> listeners {};
     std::map<Value::Id, PermanentPut> puts {};
 };
@@ -90,6 +91,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));
+    loopSignal_();
 }
 
 DhtProxyClient::~DhtProxyClient()
@@ -266,7 +268,7 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Va
                         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)) && cb) {
+                            if ((not filter or filter(*value)) and cb) {
                                 std::lock_guard<std::mutex> lock(lockCallbacks);
                                 callbacks_.emplace_back([cb, value, finished]() {
                                     if (not *finished and not cb({value}))
@@ -323,15 +325,18 @@ DhtProxyClient::put(const InfoHash& key, Sp<Value> val, DoneCallback cb, time_po
         auto nextRefresh = scheduler.time() + proxy::OP_TIMEOUT;
         search->second.puts.erase(id);
         search->second.puts.emplace(id, PermanentPut {val, scheduler.add(nextRefresh, [this, key, id]{
+            DHT_LOG.w(key, "[search %s]: PermanentPut refresh callback vid %llu", key.to_c_str(), id);
+            std::lock_guard<std::mutex> lock(searchLock_);
             auto s = searches_.find(key);
-            if (s == searches_.end())
-                return;
-            auto p = s->second.puts.find(id);
-            if (p == s->second.puts.end())
-                return;
-            const auto& now = scheduler.time();
-            doPut(key, p->second.value, {}, now, true);
-            scheduler.edit(p->second.refreshJob, now + proxy::OP_TIMEOUT);
+            if (s != searches_.end()) {
+                auto p = s->second.puts.find(id);
+                if (p != s->second.puts.end()) {
+                    doPut(key, p->second.value, [this](bool ok, const std::vector<Sp<Node>>&) {
+                        DHT_LOG.w("PermanentPut refresh %s", (ok ? "ok" : "failed"));
+                    }, time_point::max(), true);
+                    scheduler.edit(p->second.refreshJob, scheduler.time() + proxy::OP_TIMEOUT);
+                }
+            }
         })});
     }
     doPut(key, val, std::move(cb), created, permanent);
@@ -626,13 +631,29 @@ DhtProxyClient::listen(const InfoHash& key, ValueCallback cb, Value::Filter filt
 
 bool
 DhtProxyClient::cancelListen(const InfoHash& key, size_t gtoken) {
+    scheduler.syncTime();
     DHT_LOG.d(key, "[search %s]: cancelListen %zu", key.to_c_str(), gtoken);
     auto it = searches_.find(key);
     if (it == searches_.end())
         return false;
-    return it->second.ops.cancelListen(gtoken, [&](size_t ltoken){
-        doCancelListen(key, ltoken);
-    });
+    auto& ops = it->second.ops;
+    bool canceled = ops.cancelListen(gtoken, scheduler.time());
+    if (not it->second.opExpirationJob) {
+        it->second.opExpirationJob = scheduler.add(time_point::max(), [this,key](){
+            auto it = searches_.find(key);
+            if (it != searches_.end()) {
+                auto next = it->second.ops.expire(scheduler.time(), [this,key](size_t ltoken){
+                    doCancelListen(key, ltoken);
+                });
+                if (next != time_point::max()) {
+                    scheduler.edit(it->second.opExpirationJob, next);
+                }
+            }
+        });
+    }
+    scheduler.edit(it->second.opExpirationJob, ops.getExpiration());
+    loopSignal_();
+    return canceled;
 }
 
 size_t
@@ -695,6 +716,7 @@ DhtProxyClient::doListen(const InfoHash& key, ValueCallback cb, Value::Filter fi
         }
         auto next = l->second.cache.onValues(expired ? new_values_empty : values, std::vector<Value::Id>{}, expired_ids, types, scheduler.time());
         scheduler.edit(l->second.cacheExpirationJob, next);
+        loopSignal_();
         return true;
     };
     auto pushNotifToken = std::make_shared<proxy::ListenToken>(0);
@@ -747,7 +769,7 @@ DhtProxyClient::doListen(const InfoHash& key, ValueCallback cb, Value::Filter fi
                                 auto* char_data = reinterpret_cast<const char*>(&body[0]);
                                 if (reader->parse(char_data, char_data + body.size(), &json, &err)) {
                                     auto value = std::make_shared<Value>(json);
-                                    if (not filter or filter(*value))  {
+                                    if ((not filter or filter(*value)) and vcb)  {
                                         std::lock_guard<std::mutex> lock(lockCallbacks);
                                         callbacks_.emplace_back([vcb, value, state]() {
                                             if (not state->cancel and not vcb({value}, false))
@@ -761,7 +783,7 @@ DhtProxyClient::doListen(const InfoHash& key, ValueCallback cb, Value::Filter fi
                             }
                         }
                     } catch (std::runtime_error&) {
-                        state->ok = false;
+                        // NOTE: Http::close() can occurs here. Ignore this.
                     }
                 } else {
                     state->ok = false;
@@ -792,6 +814,7 @@ DhtProxyClient::doCancelListen(const InfoHash& key, size_t ltoken)
     DHT_LOG.d(key, "[search %s] cancel listen", key.to_c_str());
 
     auto& listener = it->second;
+    listener.state->cancel = true;
     if (not deviceKey_.empty()) {
         // First, be sure to have a token
         if (listener.thread.joinable()) {
@@ -805,7 +828,7 @@ DhtProxyClient::doCancelListen(const InfoHash& key, size_t ltoken)
         Json::Value body;
         body["key"] = deviceKey_;
         body["client_id"] = pushClientId_;
-        body["token"] = std::to_string(ltoken);
+        body["token"] = std::to_string(*listener.pushNotifToken);
         Json::StreamWriterBuilder wbuilder;
         wbuilder["commentStyle"] = "None";
         wbuilder["indentation"] = "";
@@ -818,8 +841,7 @@ DhtProxyClient::doCancelListen(const InfoHash& key, size_t ltoken)
     } else {
         // Just stop the request
         if (listener.thread.joinable()) {
-            // Close connection to stop listener?
-            listener.state->cancel = true;
+            // Close connection to stop listener
             if (listener.req)
                 restbed::Http::close(listener.req);
             listener.thread.join();
@@ -864,16 +886,23 @@ DhtProxyClient::restartListeners()
     for (auto& search: searches_) {
     for (auto& l: search.second.listeners) {
         auto& listener = l.second;
-        if (listener.thread.joinable())
+        auto state = listener.state;
+        if (listener.thread.joinable()) {
+            state->cancel = true;
+            if (listener.req)
+                restbed::Http::close(listener.req);
             listener.thread.join();
+        }
         // Redo listen
+        state->cancel = false;
+        state->ok = true;
         auto filter = listener.filter;
         auto cb = listener.cb;
         restbed::Uri uri(proxy::HTTP_PROTO + serverHost_ + "/" + search.first.toString());
         auto req = std::make_shared<restbed::Request>(uri);
         req->set_method("LISTEN");
         listener.req = req;
-        listener.thread = std::thread([this, filter, cb, req]()
+        listener.thread = std::thread([this, filter, cb, req, state]()
             {
                 auto settings = std::make_shared<restbed::Settings>();
                 std::chrono::milliseconds timeout(std::numeric_limits<int>::max());
@@ -881,14 +910,16 @@ DhtProxyClient::restartListeners()
 
                 auto ok = std::make_shared<std::atomic_bool>(true);
                 restbed::Http::async(req,
-                    [this, filter, cb, ok](const std::shared_ptr<restbed::Request>& 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)) {
+                            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
@@ -896,24 +927,16 @@ DhtProxyClient::restartListeners()
                                 Json::Value json;
                                 std::string err;
                                 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)) {
+                                if (reader->parse(body.data(), body.data() + body.size(), &json, &err)) {
                                     auto value = std::make_shared<Value>(json);
-                                    if ((not filter or filter(*value)) && cb) {
-                                        auto okCb = std::make_shared<std::promise<bool>>();
-                                        auto futureCb = okCb->get_future();
-                                        {
-                                            std::lock_guard<std::mutex> lock(lockCallbacks);
-                                            callbacks_.emplace_back([cb, value, okCb](){
-                                                okCb->set_value(cb({value}, false));
-                                            });
-                                            loopSignal_();
-                                        }
-                                        futureCb.wait();
-                                        if (!futureCb.get()) {
-                                            return;
-                                        }
+                                    if ((not filter or filter(*value)) and cb)  {
+                                        std::lock_guard<std::mutex> lock(lockCallbacks);
+                                        callbacks_.emplace_back([cb, value, state]() {
+                                            if (not state->cancel and not cb({value}, false))
+                                                state->cancel = true;
+                                        });
+                                        loopSignal_();
                                     }
                                 }
                             }
@@ -921,10 +944,12 @@ DhtProxyClient::restartListeners()
                             // NOTE: Http::close() can occurs here. Ignore this.
                         }
                     } else {
-                        *ok = false;
+                        state->ok = false;
                     }
                 }, settings).get();
-                if (!ok) opFailed();
+                auto& s = *state;
+                if (not s.ok and not s.cancel)
+                    opFailed();
             }
         );
     }
@@ -936,6 +961,10 @@ DhtProxyClient::pushNotificationReceived(const std::map<std::string, std::string
 {
 #if OPENDHT_PUSH_NOTIFICATIONS
     scheduler.syncTime();
+    DHT_LOG.d("Got push notification");
+    for (const auto& n : notification) {
+        DHT_LOG.d("   %s -> %s", n.first.c_str(), n.second.c_str());
+    }
     try {
         std::lock_guard<std::mutex> lock(searchLock_);
         auto timeout = notification.find("timeout");
@@ -946,8 +975,10 @@ DhtProxyClient::pushNotificationReceived(const std::map<std::string, std::string
             if (vidIt != notification.end()) {
                 // Refresh put
                 auto vid = std::stoull(vidIt->second);
-                auto put = search.puts.at(vid);
+                auto& put = search.puts.at(vid);
+                DHT_LOG.d("refresh PermanentPut: %s %llx %p %s", key.to_c_str(), vid, put.refreshJob.get(), put.refreshJob ? (put.refreshJob->do_ ? "set":"not set") : "(null job)");
                 scheduler.edit(put.refreshJob, scheduler.time());
+                loopSignal_();
             } else {
                 // Refresh listen
                 for (auto& list : search.listeners)
@@ -957,12 +988,12 @@ DhtProxyClient::pushNotificationReceived(const std::map<std::string, std::string
             auto token = std::stoull(notification.at("token"));
             for (auto& search: searches_) {
                 for (auto& list : search.second.listeners) {
-                    if (*list.second.pushNotifToken != token)
+                    if (*list.second.pushNotifToken != token or list.second.state->cancel)
                         continue;
                     DHT_LOG.d(search.first, "[search %s] handling push notification", search.first.to_c_str());
                     auto cb = list.second.cb;
                     auto filter = list.second.filter;
-                    get(search.first, [cb](const std::vector<Sp<Value>>& vals){
+                    get(search.first, [cb](const std::vector<Sp<Value>>& vals) {
                         cb(vals, false);
                         return true;
                     }, DoneCallbackSimple{}, std::move(filter));
@@ -987,18 +1018,18 @@ DhtProxyClient::resubscribe(const InfoHash& key, Listener& listener)
     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;
-    listener.thread = std::thread([=]()
-    {
+    listener.thread = std::thread([this, req, pushNotifToken, state]() {
         fillBodyToGetToken(req);
         auto settings = std::make_shared<restbed::Settings>();
-        auto ok = std::make_shared<std::atomic_bool>(true);
         restbed::Http::async(req,
-            [pushNotifToken, ok](const std::shared_ptr<restbed::Request>&,
+            [pushNotifToken, state](const std::shared_ptr<restbed::Request>&,
                                        const std::shared_ptr<restbed::Response>& reply) {
             auto code = reply->get_status_code();
             if (code == 200) {
@@ -1010,9 +1041,8 @@ DhtProxyClient::resubscribe(const InfoHash& key, Listener& listener)
                     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)) {
+                    if (reader->parse(body.data(), body.data() + body.size(), &json, &err)) {
                         if (!json.isMember("token")) return;
                         *pushNotifToken = unpackId(json, "token");
                     }
@@ -1020,10 +1050,12 @@ DhtProxyClient::resubscribe(const InfoHash& key, Listener& listener)
                     // NOTE: Http::close() can occurs here. Ignore this.
                 }
             } else {
-                *ok = false;
+                state->ok = false;
             }
         }, settings).get();
-        if (!ok) opFailed();
+        auto& s = *state;
+        if (not s.ok and not s.cancel)
+            opFailed();
     });
 #endif
 }

src/dht_proxy_server.cpp

@@ -330,6 +330,7 @@ struct DhtProxyServer::Listener {
     std::string clientId;
     std::future<size_t> internalToken;
     Sp<Scheduler::Job> expireJob;
+    Sp<Scheduler::Job> expireNotifyJob;
 };
 struct DhtProxyServer::PushListener {
     std::map<InfoHash, std::vector<Listener>> listeners;
@@ -372,6 +373,8 @@ DhtProxyServer::subscribe(const std::shared_ptr<restbed::Session>& session)
                     std::lock(schedulerLock_, lockListener_);
                     std::lock_guard<std::mutex> lk1(lockListener_, std::adopt_lock);
                     std::lock_guard<std::mutex> lk2(schedulerLock_, std::adopt_lock);
+                    scheduler_.syncTime();
+                    auto timeout = scheduler_.time() + proxy::OP_TIMEOUT;
                     // Check if listener is already present and refresh timeout if launched
                     // One push listener per pushToken.infoHash.clientId
                     auto pushListener = pushListeners_.emplace(pushToken, PushListener{}).first;
@@ -379,7 +382,8 @@ DhtProxyServer::subscribe(const std::shared_ptr<restbed::Session>& session)
                     for (auto& listener: listeners->second) {
                         if (listener.clientId == clientId) {
                             *listener.token = tokenFromReq;
-                            scheduler_.edit(listener.expireJob, scheduler_.time() + proxy::OP_TIMEOUT);
+                            scheduler_.edit(listener.expireJob, timeout);
+                            scheduler_.edit(listener.expireNotifyJob, timeout - proxy::OP_MARGIN);
                             s->close(restbed::OK, "{\"token\": " + std::to_string(tokenFromReq) + "}\n");
                             schedulerCv_.notify_one();
                             return;
@@ -387,6 +391,7 @@ DhtProxyServer::subscribe(const std::shared_ptr<restbed::Session>& session)
                     }
                     listeners->second.emplace_back(Listener{});
                     auto& listener = listeners->second.back();
+                    listener.clientId = clientId;
 
                     // New listener
                     pushListener->second.isAndroid = isAndroid;
@@ -405,10 +410,14 @@ DhtProxyServer::subscribe(const std::shared_ptr<restbed::Session>& session)
                             return true;
                         }
                     );
-                    listener.expireJob = scheduler_.add(scheduler_.time() + proxy::OP_TIMEOUT,
+                    listener.expireJob = scheduler_.add(timeout,
+                        [this, clientId, infoHash, pushToken] {
+                            cancelPushListen(pushToken, infoHash, clientId);
+                        }
+                    );
+                    listener.expireNotifyJob = scheduler_.add(timeout - proxy::OP_MARGIN,
                         [this, token, infoHash, pushToken, isAndroid, clientId] {
-                            cancelPushListen(pushToken, infoHash, *token);
-                            // Send a push notification to inform the client that this listen has timeout
+                            std::cout << "Listener: sending refresh " << infoHash << std::endl;
                             Json::Value json;
                             json["timeout"] = infoHash.toString();
                             json["to"] = clientId;
@@ -452,10 +461,9 @@ DhtProxyServer::unsubscribe(const std::shared_ptr<restbed::Session>& session)
                 }
                 auto pushToken = root["key"].asString();
                 if (pushToken.empty()) return;
-                auto token = unpackId(root, "token");
-                if (token == 0) return;
+                auto clientId = root["client_id"].asString();
 
-                cancelPushListen(pushToken, infoHash, token);
+                cancelPushListen(pushToken, infoHash, clientId);
                 s->close(restbed::OK);
             } catch (...) {
                 s->close(restbed::INTERNAL_SERVER_ERROR, "{\"err\":\"Internal server error\"}");
@@ -465,8 +473,9 @@ DhtProxyServer::unsubscribe(const std::shared_ptr<restbed::Session>& session)
 }
 
 void
-DhtProxyServer::cancelPushListen(const std::string& pushToken, const dht::InfoHash& key, proxy::ListenToken token)
+DhtProxyServer::cancelPushListen(const std::string& pushToken, const dht::InfoHash& key, const std::string& clientId)
 {
+    std::cout << "cancelPushListen: " << key << " clientId:" << clientId << std::endl;
     std::lock_guard<std::mutex> lock(lockListener_);
     auto pushListener = pushListeners_.find(pushToken);
     if (pushListener == pushListeners_.end())
@@ -475,7 +484,7 @@ DhtProxyServer::cancelPushListen(const std::string& pushToken, const dht::InfoHa
     if (listeners == pushListener->second.listeners.end())
         return;
     for (auto listener = listeners->second.begin(); listener != listeners->second.end();) {
-        if (*listener->token == token) {
+        if (listener->clientId == clientId) {
             if (dht_)
                 dht_->cancelListen(key, std::move(listener->internalToken));
             listener = listeners->second.erase(listener);
@@ -579,8 +588,8 @@ DhtProxyServer::put(const std::shared_ptr<restbed::Session>& session)
                         if (reader->parse(char_data, char_data + b.size(), &root, &err)) {
                             // Build the Value from json
                             auto value = std::make_shared<Value>(root);
-                            auto permanent = root.isMember("permanent");
-                            std::cout << "Got put " << infoHash << " " << value << " " << (permanent ? "permanent" : "") << std::endl;
+                            bool permanent = root.isMember("permanent");
+                            std::cout << "Got put " << infoHash << " " << *value << " " << (permanent ? "permanent" : "") << std::endl;
 
                             if (permanent) {
                                 std::string pushToken, clientId, platform;
@@ -590,35 +599,41 @@ DhtProxyServer::put(const std::shared_ptr<restbed::Session>& session)
                                     clientId = pVal["client_id"].asString();
                                     platform = pVal["platform"].asString();
                                 }
+                                std::cout << "permanent: pushToken " << pushToken << " clientId " << clientId << " platform " << platform << std::endl;
                                 bool isAndroid = platform == "android";
                                 std::unique_lock<std::mutex> lock(schedulerLock_);
                                 scheduler_.syncTime();
-                                auto sPuts = puts_.emplace(infoHash, SearchPuts{}).first;
                                 auto timeout = scheduler_.time() + proxy::OP_TIMEOUT;
                                 auto vid = value->id;
+                                auto sPuts = puts_.emplace(infoHash, SearchPuts{}).first;
                                 auto r = sPuts->second.puts.emplace(vid, PermanentPut{});
+                                auto& pput = r.first->second;
                                 if (r.second) {
-                                    r.first->second.expireJob = scheduler_.add(timeout, [this, infoHash, vid]{
-                                        std::cout << "Permanent put expired" << infoHash << std::endl;
+                                    std::cout << "New ppush entry" << std::endl;
+                                    pput.expireJob = scheduler_.add(timeout, [this, infoHash, vid]{
+                                        std::cout << "Permanent put expired: " << infoHash << " " << vid << std::endl;
                                         cancelPut(infoHash, vid);
                                     });
 #if OPENDHT_PUSH_NOTIFICATIONS
-                                    if (not pushToken.empty())
-                                        r.first->second.expireNotifyJob = scheduler_.add(timeout - proxy::OP_MARGIN,
+                                    if (not pushToken.empty()) {
+                                        std::cout << "Adding refresh job" << std::endl;
+                                        pput.expireNotifyJob = scheduler_.add(timeout - proxy::OP_MARGIN,
                                             [this, infoHash, vid, pushToken, clientId, isAndroid]
                                         {
-                                            std::cout << "Permanent put refresh" << infoHash << std::endl;
+                                            std::cout << "Permanent put refresh: " << infoHash << " " << std::to_string(vid) << " " << clientId << std::endl;
                                             Json::Value json;
                                             json["timeout"] = infoHash.toString();
                                             json["to"] = clientId;
                                             json["vid"] = std::to_string(vid);
                                             sendPushNotification(pushToken, json, isAndroid);
                                         });
+                                    }
 #endif
                                 } else {
-                                    scheduler_.edit(r.first->second.expireJob, timeout);
-                                    if (r.first->second.expireNotifyJob)
-                                        scheduler_.edit(r.first->second.expireNotifyJob, timeout - proxy::OP_MARGIN);
+                                    std::cout << "Refreshing put jobs" << std::endl;
+                                    scheduler_.edit(pput.expireJob, timeout);
+                                    if (pput.expireNotifyJob)
+                                        scheduler_.edit(pput.expireNotifyJob, timeout - proxy::OP_MARGIN);
                                 }
                                 lock.unlock();
                                 schedulerCv_.notify_one();
@@ -643,7 +658,8 @@ DhtProxyServer::put(const std::shared_ptr<restbed::Session>& session)
                 } else {
                     s->close(restbed::SERVICE_UNAVAILABLE, "{\"err\":\"Incorrect DhtRunner\"}");
                 }
-            } catch (...) {
+            } catch (const std::exception& e) {
+                std::cout << "Error performing put: " << e.what() << std::endl;
                 s->close(restbed::INTERNAL_SERVER_ERROR, "{\"err\":\"Internal server error\"}");
             }
         }

src/dhtrunner.cpp

@@ -600,8 +600,8 @@ void
 DhtRunner::cancelListen(InfoHash h, size_t token)
 {
     std::lock_guard<std::mutex> lck(storage_mtx);
+    pending_ops.emplace([=](SecureDht& dht) {
 #if OPENDHT_PROXY_CLIENT
-    pending_ops.emplace([=](SecureDht&) {
         auto it = listeners_.find(token);
         if (it == listeners_.end()) return;
         if (it->second.tokenClassicDht)
@@ -609,23 +609,31 @@ DhtRunner::cancelListen(InfoHash h, size_t token)
         if (it->second.tokenProxyDht and dht_via_proxy_)
             dht_via_proxy_->cancelListen(h, it->second.tokenProxyDht);
         listeners_.erase(it);
-    });
 #else
-    pending_ops.emplace([=](SecureDht& dht) {
         dht.cancelListen(h, token);
-    });
 #endif // OPENDHT_PROXY_CLIENT
+    });
     cv.notify_all();
 }
 
 void
-DhtRunner::cancelListen(InfoHash h, std::shared_future<size_t> token)
+DhtRunner::cancelListen(InfoHash h, std::shared_future<size_t> ftoken)
 {
     {
         std::lock_guard<std::mutex> lck(storage_mtx);
         pending_ops.emplace([=](SecureDht& dht) {
-            auto tk = token.get();
-            dht.cancelListen(h, tk);
+            auto token = ftoken.get();
+#if OPENDHT_PROXY_CLIENT
+            auto it = listeners_.find(token);
+            if (it == listeners_.end()) return;
+            if (it->second.tokenClassicDht)
+                dht_->cancelListen(h, it->second.tokenClassicDht);
+            if (it->second.tokenProxyDht and dht_via_proxy_)
+                dht_via_proxy_->cancelListen(h, it->second.tokenProxyDht);
+            listeners_.erase(it);
+#else
+            dht.cancelListen(h, token);
+#endif // OPENDHT_PROXY_CLIENT
         });
     }
     cv.notify_all();
@@ -958,11 +966,17 @@ DhtRunner::setPushNotificationToken(const std::string& token) {
 }
 
 void
-DhtRunner::pushNotificationReceived(const std::map<std::string, std::string>& data) const
+DhtRunner::pushNotificationReceived(const std::map<std::string, std::string>& data)
 {
 #if OPENDHT_PROXY_CLIENT && OPENDHT_PUSH_NOTIFICATIONS
-    if (dht_via_proxy_)
-        dht_via_proxy_->pushNotificationReceived(data);
+    {
+        std::lock_guard<std::mutex> lck(storage_mtx);
+        pending_ops_prio.emplace([=](SecureDht& dht) {
+            if (dht_via_proxy_)
+                dht_via_proxy_->pushNotificationReceived(data);
+        });
+    }
+    cv.notify_all();
 #endif
 }
 

src/listener.h

@@ -20,6 +20,7 @@
 
 #include "value.h"
 #include "utils.h"
+#include "callbacks.h"
 
 namespace dht {
 

src/op_cache.cpp

@@ -0,0 +1,204 @@
+/*
+ *  Copyright (C) 2018 Savoir-faire Linux Inc.
+ *  Author(s) : Adrien Béraud <adrien.beraud@savoirfairelinux.com>
+ *
+ *  This program is free software; you can redistribute it and/or modify
+ *  it under the terms of the GNU General Public License as published by
+ *  the Free Software Foundation; either version 3 of the License, or
+ *  (at your option) any later version.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+#include "op_cache.h"
+
+namespace dht {
+
+constexpr const std::chrono::seconds OpCache::EXPIRATION;
+
+bool
+OpValueCache::onValuesAdded(const std::vector<Sp<Value>>& vals) {
+    std::vector<Sp<Value>> newValues;
+    for (const auto& v : vals) {
+        auto viop = values.emplace(v->id, OpCacheValueStorage{v});
+        if (viop.second) {
+            newValues.emplace_back(v);
+        } else {
+            viop.first->second.refCount++;
+        }
+    }
+    return newValues.empty() ? true : callback(newValues, false);
+}
+bool
+OpValueCache::onValuesExpired(const std::vector<Sp<Value>>& vals) {
+    std::vector<Sp<Value>> expiredValues;
+    for (const auto& v : vals) {
+        auto vit = values.find(v->id);
+        if (vit != values.end()) {
+            vit->second.refCount--;
+            if (not vit->second.refCount) {
+                expiredValues.emplace_back(std::move(vit->second.data));
+                values.erase(vit);
+            }
+        }
+    }
+    return expiredValues.empty() ? true : callback(expiredValues, true);
+}
+std::vector<Sp<Value>>
+OpValueCache::get(Value::Filter& filter) const {
+    std::vector<Sp<Value>> ret;
+    if (not filter)
+        ret.reserve(values.size());
+    for (const auto& v : values)
+        if (not filter or filter(*v.second.data))
+            ret.emplace_back(v.second.data);
+    return ret;
+}
+
+Sp<Value>
+OpValueCache::get(Value::Id id) const {
+    auto v = values.find(id);
+    if (v == values.end())
+        return {};
+    return v->second.data;
+}
+
+std::vector<Sp<Value>>
+OpValueCache::getValues() const {
+    std::vector<Sp<Value>> ret;
+    ret.reserve(values.size());
+    for (const auto& v : values)
+        ret.emplace_back(v.second.data);
+    return ret;
+}
+
+void
+OpCache::onValuesAdded(const std::vector<Sp<Value>>& vals) {
+    if (not listeners.empty()) {
+        std::vector<LocalListener> list;
+        list.reserve(listeners.size());
+        for (const auto& l : listeners)
+            list.emplace_back(l.second);
+        for (auto& l : list)
+            l.get_cb(l.filter.filter(vals), false);
+    }
+}
+
+void
+OpCache::onValuesExpired(const std::vector<Sp<Value>>& vals) {
+    if (not listeners.empty()) {
+        std::vector<LocalListener> list;
+        list.reserve(listeners.size());
+        for (const auto& l : listeners)
+            list.emplace_back(l.second);
+        for (auto& l : list)
+            l.get_cb(l.filter.filter(vals), true);
+    }
+}
+
+time_point
+OpCache::getExpiration() const {
+    if (not listeners.empty())
+        return time_point::max();
+    return lastRemoved + EXPIRATION;
+}
+
+size_t
+SearchCache::listen(ValueCallback get_cb, Sp<Query> q, Value::Filter filter, std::function<size_t(Sp<Query>, ValueCallback)> onListen)
+{
+    // find exact match
+    auto op = ops.find(q);
+    if (op == ops.end()) {
+        // find satisfying query
+        for (auto it = ops.begin(); it != ops.end(); it++) {
+            if (q->isSatisfiedBy(*it->first)) {
+                op = it;
+                break;
+            }
+        }
+    }
+    if (op == ops.end()) {
+        // New query
+        op = ops.emplace(q, std::unique_ptr<OpCache>(new OpCache)).first;
+        auto& cache = *op->second;
+        cache.searchToken = onListen(q, [&](const std::vector<Sp<Value>>& values, bool expired){
+            return cache.onValue(values, expired);
+        });
+    }
+    auto token = nextToken_++;
+    if (nextToken_ == 0)
+        nextToken_++;
+    op->second->addListener(token, get_cb, q, filter);
+    return token;
+}
+
+bool
+SearchCache::cancelListen(size_t gtoken, const time_point& now) {
+    for (auto& op : ops) {
+        if (op.second->removeListener(gtoken, now)) {
+            nextExpiration_ = std::min(nextExpiration_, op.second->getExpiration());
+            return true;
+        }
+    }
+    return false;
+}
+
+void
+SearchCache::cancelAll(std::function<void(size_t)> onCancel) {
+    for (auto& op : ops) {
+        auto cache = std::move(op.second);
+        cache->removeAll();
+        onCancel(cache->searchToken);
+    }
+    ops.clear();
+}
+
+time_point
+SearchCache::expire(const time_point& now, std::function<void(size_t)> onCancel) {
+    nextExpiration_ = time_point::max();
+    auto ret = nextExpiration_;
+    for (auto it = ops.begin(); it != ops.end();) {
+        auto expiration = it->second->getExpiration();
+        if (expiration < now) {
+            auto cache = std::move(it->second);
+            it = ops.erase(it);
+            onCancel(cache->searchToken);
+        } else {
+            nextExpiration_ = std::min(nextExpiration_, expiration);
+            ret = nextExpiration_;
+            ++it;
+        }
+    }
+    return ret;
+}
+
+std::vector<Sp<Value>>
+SearchCache::get(Value::Filter& filter) const {
+    if (ops.size() == 1)
+        return ops.begin()->second->get(filter);
+    std::map<Value::Id, Sp<Value>> c;
+    for (const auto& op : ops) {
+        for (const auto& v : op.second->get(filter))
+            c.emplace(v->id, v);
+    }
+    std::vector<Sp<Value>> ret;
+    ret.reserve(c.size());
+    for (auto& v : c)
+        ret.emplace_back(std::move(v.second));
+    return ret;
+}
+
+Sp<Value>
+SearchCache::get(Value::Id id) const {
+    for (const auto& op : ops)
+        if (auto v = op.second->get(id))
+            return v;
+    return {};
+}
+
+}

src/op_cache.h

@@ -33,6 +33,9 @@ struct OpCacheValueStorage
 class OpValueCache {
 public:
     OpValueCache(ValueCallback&& cb) : callback(std::forward<ValueCallback>(cb)) {}
+    OpValueCache(OpValueCache&& o) : values(std::move(o.values)), callback(std::move(o.callback)) {
+        o.callback = {};
+    }
 
     static ValueCallback cacheCallback(ValueCallback&& cb) {
         auto cache = std::make_shared<OpValueCache>(std::forward<ValueCallback>(cb));
@@ -48,58 +51,17 @@ public:
             return onValuesAdded(vals);
     }
 
-    bool onValuesAdded(const std::vector<Sp<Value>>& vals) {
-        std::vector<Sp<Value>> newValues;
-        for (const auto& v : vals) {
-            auto viop = values.emplace(v->id, OpCacheValueStorage{v});
-            if (viop.second) {
-                newValues.emplace_back(v);
-            } else {
-                viop.first->second.refCount++;
-            }
-        }
-        return newValues.empty() ? true : callback(newValues, false);
-    }
-    bool onValuesExpired(const std::vector<Sp<Value>>& vals) {
-        std::vector<Sp<Value>> expiredValues;
-        for (const auto& v : vals) {
-            auto vit = values.find(v->id);
-            if (vit != values.end()) {
-                vit->second.refCount--;
-                if (not vit->second.refCount) {
-                    expiredValues.emplace_back(std::move(vit->second.data));
-                    values.erase(vit);
-                }
-            }
-        }
-        return expiredValues.empty() ? true : callback(expiredValues, true);
-    }
-    std::vector<Sp<Value>> get(Value::Filter& filter) const {
-        std::vector<Sp<Value>> ret;
-        if (not filter)
-            ret.reserve(values.size());
-        for (const auto& v : values)
-            if (not filter or filter(*v.second.data))
-                ret.emplace_back(v.second.data);
-        return ret;
-    }
+    bool onValuesAdded(const std::vector<Sp<Value>>& vals);
+    bool onValuesExpired(const std::vector<Sp<Value>>& vals);
 
-    Sp<Value> get(Value::Id id) const {
-        auto v = values.find(id);
-        if (v == values.end())
-            return {};
-        return v->second.data;
-    }
-
-    std::vector<Sp<Value>> getValues() const {
-        std::vector<Sp<Value>> ret;
-        ret.reserve(values.size());
-        for (const auto& v : values)
-            ret.emplace_back(v.second.data);
-        return ret;
-    }
+    std::vector<Sp<Value>> get(Value::Filter& filter) const;
+    Sp<Value> get(Value::Id id) const;
+    std::vector<Sp<Value>> getValues() const;
 
 private:
+    OpValueCache(const OpValueCache&) = delete;
+    OpValueCache& operator=(const OpValueCache&) = delete;
+
     std::map<Value::Id, OpCacheValueStorage> values {};
     ValueCallback callback;
 };
@@ -119,33 +81,16 @@ public:
         return not listeners.empty();
     }
 
-    void onValuesAdded(const std::vector<Sp<Value>>& vals) {
-        if (not listeners.empty()) {
-            std::vector<LocalListener> list;
-            list.reserve(listeners.size());
-            for (const auto& l : listeners)
-                list.emplace_back(l.second);
-            for (auto& l : list)
-                l.get_cb(l.filter.filter(vals), false);
-        }
-    }
-    void onValuesExpired(const std::vector<Sp<Value>>& vals) {
-        if (not listeners.empty()) {
-            std::vector<LocalListener> list;
-            list.reserve(listeners.size());
-            for (const auto& l : listeners)
-                list.emplace_back(l.second);
-            for (auto& l : list)
-                l.get_cb(l.filter.filter(vals), true);
-        }
-    }
+    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) {
         listeners.emplace(token, LocalListener{q, filter, cb});
         cb(cache.get(filter), false);
     }
 
-    bool removeListener(size_t token) {
+    bool removeListener(size_t token, const time_point& now) {
+        lastRemoved = now;
         return listeners.erase(token) > 0;
     }
 
@@ -165,89 +110,46 @@ public:
         return cache.get(id);
     }
 
+    bool isExpired(const time_point& now) const {
+        return listeners.empty() and (lastRemoved + EXPIRATION < now);
+    }
+    time_point getExpiration() const;
+
     size_t searchToken;
 private:
+    constexpr static const std::chrono::seconds EXPIRATION {60};
+    OpCache(const OpCache&) = delete;
+    OpCache& operator=(const OpCache&) = delete;
+
     OpValueCache cache;
     std::map<size_t, LocalListener> listeners;
+    time_point lastRemoved {clock::now()};
 };
 
 class SearchCache {
 public:
-    size_t listen(ValueCallback get_cb, Sp<Query> q, Value::Filter filter, std::function<size_t(Sp<Query>, ValueCallback)> onListen) {
-        // find exact match
-        auto op = ops.find(q);
-        if (op == ops.end()) {
-            // find satisfying query
-            for (auto it = ops.begin(); it != ops.end(); it++) {
-                if (q->isSatisfiedBy(*it->first)) {
-                    op = it;
-                    break;
-                }
-            }
-        }
-        if (op == ops.end()) {
-            // New query
-            op = ops.emplace(q, std::unique_ptr<OpCache>(new OpCache)).first;
-            auto& cache = *op->second;
-            cache.searchToken = onListen(q, [&](const std::vector<Sp<Value>>& values, bool expired){
-                return cache.onValue(values, expired);
-            });
-        }
-        auto token = nextToken_++;
-        if (nextToken_ == 0)
-            nextToken_++;
-        op->second->addListener(token, get_cb, q, filter);
-        return token;
+    SearchCache() {}
+    SearchCache(SearchCache&&) = default;
+    size_t listen(ValueCallback get_cb, Sp<Query> q, Value::Filter filter, std::function<size_t(Sp<Query>, ValueCallback)> onListen);
+
+    bool cancelListen(size_t gtoken, const time_point& now);
+    void cancelAll(std::function<void(size_t)> onCancel);
+
+    time_point expire(const time_point& now, std::function<void(size_t)> onCancel);
+    time_point getExpiration() const {
+        return nextExpiration_;
     }
 
-    bool cancelListen(size_t gtoken, std::function<void(size_t)> onCancel) {
-        for (auto it = ops.begin(); it != ops.end(); it++) {
-            if (it->second->removeListener(gtoken)) {
-                if (it->second->isDone()) {
-                    auto cache = std::move(it->second);
-                    ops.erase(it);
-                    onCancel(cache->searchToken);
-                }
-                return true;
-            }
-        }
-        return false;
-    }
-
-    void cancelAll(std::function<void(size_t)> onCancel) {
-        for (auto& op : ops) {
-            auto cache = std::move(op.second);
-            cache->removeAll();
-            onCancel(cache->searchToken);
-        }
-        ops.clear();
-    }
-
-    std::vector<Sp<Value>> get(Value::Filter& filter) const {
-        if (ops.size() == 1)
-            return ops.begin()->second->get(filter);
-        std::map<Value::Id, Sp<Value>> c;
-        for (const auto& op : ops) {
-            for (const auto& v : op.second->get(filter))
-                c.emplace(v->id, v);
-        }
-        std::vector<Sp<Value>> ret;
-        ret.reserve(c.size());
-        for (auto& v : c)
-            ret.emplace_back(std::move(v.second));
-        return ret;
-    }
-
-    Sp<Value> get(Value::Id id) const {
-        for (const auto& op : ops)
-            if (auto v = op.second->get(id))
-                return v;
-        return {};
-    }
+    std::vector<Sp<Value>> get(Value::Filter& filter) const;
+    Sp<Value> get(Value::Id id) const;
 
 private:
+    SearchCache(const SearchCache&) = delete;
+    SearchCache& operator=(const SearchCache&) = delete;
+
     std::map<Sp<Query>, std::unique_ptr<OpCache>> ops {};
     size_t nextToken_ {1};
+    time_point nextExpiration_ {time_point::max()};
 };
 
 }

src/search.h

@@ -383,8 +383,11 @@ struct Dht::Search {
 
     /* Cache */
     SearchCache cache;
+    Sp<Scheduler::Job> opExpirationJob;
 
     ~Search() {
+        if (opExpirationJob)
+            opExpirationJob->cancel();
         for (auto& get : callbacks) {
             get.second.done_cb(false, {});
             get.second.done_cb = {};
@@ -483,21 +486,27 @@ struct Dht::Search {
         });
     }
 
-    void cancelListen(size_t token) {
-        cache.cancelListen(token, [&](size_t t){
-            Sp<Query> query;
-            const auto& ll = listeners.find(t);
-            if (ll != listeners.cend()) {
-                query = ll->second.query;
-                listeners.erase(ll);
-            }
-            for (auto& sn : nodes) {
-                if (listeners.empty())
-                    sn.cancelListen();
-                else if (query)
-                    sn.cancelListen(query);
-            }
-        });
+    void cancelListen(size_t token, Scheduler& scheduler) {
+        cache.cancelListen(token, scheduler.time());
+        if (not opExpirationJob)
+            opExpirationJob = scheduler.add(time_point::max(), [this,&scheduler]{
+                auto nextExpire = cache.expire(scheduler.time(), [&](size_t t){
+                    Sp<Query> query;
+                    const auto& ll = listeners.find(t);
+                    if (ll != listeners.cend()) {
+                        query = ll->second.query;
+                        listeners.erase(ll);
+                    }
+                    for (auto& sn : nodes) {
+                        if (listeners.empty())
+                            sn.cancelListen();
+                        else if (query)
+                            sn.cancelListen(query);
+                    }
+                });
+                scheduler.edit(opExpirationJob, nextExpire);
+            });
+        scheduler.edit(opExpirationJob, cache.getExpiration());
     }
 
     /**

tools/dhtnode.cpp

@@ -89,7 +89,8 @@ void print_help() {
               << "  pp <key> <str>        Put string value at <key> (persistent version)." << std::endl
               << "  cpp <key> <id>        Cancel persistent put operation for <key> and value <id>." << std::endl
               << "  s <key> <str>         Put string value at <key>, signed with our generated private key." << std::endl
-              << "  e <key> <dest> <str>  Put string value at <key>, encrypted for <dest> with its public key (if found)." << std::endl;
+              << "  e <key> <dest> <str>  Put string value at <key>, encrypted for <dest> with its public key (if found)." << std::endl
+              << "  cc                    Trigger connectivity changed signal." << std::endl;
 
 #ifdef OPENDHT_INDEXATION
     std::cout << std::endl << "Indexation operations on the DHT:" << std::endl
@@ -179,11 +180,8 @@ void cmd_loop(std::shared_ptr<DhtRunner>& dht, dht_params& params
             iss >> idstr;
             try {
                 auto addr = splitPort(idstr);
-                if (not addr.first.empty() and addr.second.empty()){
-                    std::stringstream ss;
-                    ss << DHT_DEFAULT_PORT;
-                    addr.second = ss.str();
-                }
+                if (not addr.first.empty() and addr.second.empty())
+                    addr.second = std::to_string(DHT_DEFAULT_PORT);
                 dht->bootstrap(addr.first.c_str(), addr.second.c_str());
             } catch (const std::exception& e) {
                 std::cerr << e.what() << std::endl;
@@ -199,6 +197,8 @@ void cmd_loop(std::shared_ptr<DhtRunner>& dht, dht_params& params
                 log::disableLogging(*dht);
             dht->setLogFilter(filter);
             continue;
+        } else if (op == "cc") {
+            dht->connectivityChanged();
         }
 #if OPENDHT_PROXY_SERVER
         else if (op == "pst") {