network engine: handle EPERM

travis: run CI with ASAN

.travis.yml

@@ -37,7 +37,7 @@ script:
     # proxy builds
     if [[ "$OPENDHT_TEST_JOB" != *"opendht.llvm"* ]] && [[ "$OPENDHT_TEST_JOB" != *"opendht.classic"* ]]; then
       docker build -t opendht-proxy -f docker/DockerfileTravisProxy .;
-      options='';
+      options='-DOPENDHT_SANITIZE=On ';
       if [[ "$OPENDHT_TEST_JOB" == *"opendht.proxyserver"* ]] || [[ "$OPENDHT_TEST_JOB" == *"opendht.push"* ]]; then
         options+='-DOPENDHT_PROXY_SERVER=ON ';
       else

CMakeLists.txt

@@ -1,7 +1,7 @@
 cmake_minimum_required (VERSION 3.1)
 project (opendht)
 set (opendht_VERSION_MAJOR 1)
-set (opendht_VERSION_MINOR 7.1)
+set (opendht_VERSION_MINOR 7.2)
 set (opendht_VERSION ${opendht_VERSION_MAJOR}.${opendht_VERSION_MINOR})
 set (PACKAGE_VERSION ${opendht_VERSION})
 set (VERSION "${opendht_VERSION}")
@@ -304,13 +304,22 @@ install (FILES ${CMAKE_CURRENT_BINARY_DIR}/opendhtConfigVersion.cmake DESTINATIO
 IF(OPENDHT_TESTS)
   FIND_PACKAGE(Cppunit REQUIRED)
   # unit testing
-    add_executable(opendht_unit_tests
-      tests/tests_runner.cpp
+    list (APPEND test_FILES
       tests/infohashtester.h
       tests/infohashtester.cpp
       tests/dhtrunnertester.h
       tests/dhtrunnertester.cpp
     )
+    if (OPENDHT_PROXY_SERVER AND OPENDHT_PROXY_CLIENT)
+      list (APPEND test_FILES
+        tests/dhtproxytester.h
+        tests/dhtproxytester.cpp
+      )
+    endif()
+    add_executable(opendht_unit_tests
+      tests/tests_runner.cpp
+      ${test_FILES}
+    )
     if (OPENDHT_SHARED)
       TARGET_LINK_LIBRARIES(opendht_unit_tests opendht)
     else ()

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], 1)
+m4_define([opendht_patch_version], 2)
 m4_define([opendht_version],
 		  [opendht_major_version.opendht_minor_version.opendht_patch_version])
 

include/opendht/default_types.h

@@ -114,6 +114,8 @@ public:
     ImMessage() {}
     ImMessage(dht::Value::Id id, std::string&& m, long d = 0)
         : id(id), msg(std::move(m)), date(d) {}
+    ImMessage(dht::Value::Id id, std::string&& dt, std::string&& m, long d = 0)
+        : id(id), msg(std::move(m)), datatype(std::move(dt)), date(d) {}
 
     virtual void unpackValue(const Value& v) override {
         to = v.recipient;
@@ -123,10 +125,11 @@ public:
     dht::InfoHash to;
     dht::Value::Id id {0};
     std::string msg;
+    std::string datatype;
     long date {0};
     ImStatus status {ImStatus::NONE};
 
-    MSGPACK_DEFINE_MAP(id, msg, date, status)
+    MSGPACK_DEFINE_MAP(id, msg, date, status, datatype)
 };
 
 class OPENDHT_PUBLIC TrustRequest : public EncryptedValue<TrustRequest>

include/opendht/dht_proxy_client.h

@@ -269,6 +269,7 @@ private:
      * Get informations from the proxy node
      * @return the JSON returned by the proxy
      */
+    struct InfoState;
     void getProxyInfos();
     void onProxyInfos(const Json::Value& val, sa_family_t family);
     SockAddr parsePublicAddress(const Json::Value& val);
@@ -278,6 +279,10 @@ private:
     size_t doListen(const InfoHash& key, ValueCallback, Value::Filter);
     bool doCancelListen(const InfoHash& key, size_t token);
 
+    struct ListenState;
+    void sendListen(const std::shared_ptr<restbed::Request>& request, const ValueCallback&, const Value::Filter& filter, const Sp<ListenState>& state);
+    void sendSubscribe(const std::shared_ptr<restbed::Request>& request, const Sp<proxy::ListenToken>&, const Sp<ListenState>& state);
+
     void doPut(const InfoHash&, Sp<Value>, DoneCallback, time_point created, bool permanent);
 
     /**
@@ -296,7 +301,7 @@ private:
     std::string serverHost_;
     std::string pushClientId_;
 
-    std::atomic_flag ongoingStatusUpdate_ = ATOMIC_FLAG_INIT;
+    mutable std::mutex lockCurrentProxyInfos_;
     NodeStatus statusIpv4_ {NodeStatus::Disconnected};
     NodeStatus statusIpv6_ {NodeStatus::Disconnected};
     NodeStats stats4_ {};
@@ -336,8 +341,9 @@ private:
     std::vector<std::function<void()>> callbacks_;
     std::mutex lockCallbacks;
 
+    Sp<InfoState> infoState_;
     std::thread statusThread_;
-    mutable std::mutex lockCurrentProxyInfos_;
+    mutable std::mutex statusLock_;
 
     Scheduler scheduler;
     /**

include/opendht/log_enable.h

@@ -43,8 +43,23 @@ inline void NOLOG(char const*, va_list) {}
 struct LogMethod {
     LogMethod() = default;
 
+    LogMethod(LogMethod&& l) : func(std::move(l.func)) {}
+    LogMethod(const LogMethod& l) : func(l.func) {}
+
+    LogMethod& operator=(dht::LogMethod&& l) {
+        func = std::forward<LogMethod>(l.func);
+        return *this;
+    }
+    LogMethod& operator=(const dht::LogMethod& l) {
+        func = l.func;
+        return *this;
+    }
+
     template<typename T>
-    LogMethod(T&& t) : func(std::forward<T>(t)) {}
+    explicit LogMethod(T&& t) : func(std::forward<T>(t)) {}
+
+    template<typename T>
+    LogMethod(const T& t) : func(t) {}
 
     void operator()(char const* format, ...) const {
         va_list args;

include/opendht/securedht.h

@@ -315,7 +315,7 @@ public:
         DHT_LOG.DEBUG = debug;
         DHT_LOG.WARN = warn;
         DHT_LOG.ERR = error;
-        dht_->setLoggers(std::forward<LogMethod>(error), std::forward<LogMethod>(warn), std::forward<LogMethod>(debug));
+        dht_->setLoggers(error, warn, debug);
     }
 
     /**

src/dht.cpp

@@ -516,7 +516,7 @@ Dht::searchSynchedNodeListen(const Sp<Search>& sr, SearchNode& n)
             });
         }
         auto prev_req = r != n.listenStatus.end() ? r->second.req : nullptr;
-        r->second.req = network_engine.sendListen(n.node, sr->id, *query, n.token, prev_req,
+        auto new_req = network_engine.sendListen(n.node, sr->id, *query, n.token, prev_req,
             [this,ws,query](const net::Request& req, net::RequestAnswer&& answer) mutable
             { /* on done */
                 if (auto sr = ws.lock()) {
@@ -545,6 +545,11 @@ Dht::searchSynchedNodeListen(const Sp<Search>& sr, SearchNode& n)
                 }
             }
         );
+        // Here the request may have failed and the CachedListenStatus removed
+        r = n.listenStatus.find(query);
+        if (r != n.listenStatus.end()) {
+            r->second.req = new_req;
+        }
     }
 }
 
@@ -1535,10 +1540,7 @@ Dht::dumpSearch(const Search& sr, std::ostream& out) const
                     if (ack == n.acked.end() or not ack->second.first) {
                         out << ' ';
                     } else {
-                        if (ack->second.first->completed())
-                            out << 'a';
-                        else if (ack->second.first->pending())
-                            out << 'f';
+                        out << ack->second.first->getStateChar();
                     }
                 }
                 out << "] ";

src/dht_proxy_client.cpp

@@ -33,12 +33,18 @@
 
 namespace dht {
 
+struct DhtProxyClient::InfoState {
+    std::atomic_uint ipv4 {0}, ipv6 {0};
+    std::atomic_bool cancel {false};
+};
+
+struct DhtProxyClient::ListenState {
+    std::atomic_bool ok {true};
+    std::atomic_bool cancel {false};
+};
+
 struct DhtProxyClient::Listener
 {
-    struct State {
-        std::atomic_bool ok {true};
-        std::atomic_bool cancel {false};
-    };
     ValueCache cache;
     Sp<Scheduler::Job> cacheExpirationJob {};
     ValueCallback cb;
@@ -46,7 +52,7 @@ struct DhtProxyClient::Listener
     Sp<restbed::Request> req;
     std::thread thread;
     unsigned callbackId;
-    Sp<State> state;
+    Sp<ListenState> state;
     Sp<proxy::ListenToken> pushNotifToken;
     Sp<Scheduler::Job> refreshJob;
     Listener(ValueCache&& c, Sp<Scheduler::Job>&& j, const Sp<restbed::Request>& r, Value::Filter&& f)
@@ -99,6 +105,8 @@ DhtProxyClient::~DhtProxyClient()
     isDestroying_ = true;
     cancelAllOperations();
     cancelAllListeners();
+    if (infoState_)
+        infoState_->cancel = true;
     if (statusThread_.joinable())
         statusThread_.join();
 }
@@ -322,15 +330,16 @@ DhtProxyClient::put(const InfoHash& key, Sp<Value> val, DoneCallback cb, time_po
         std::lock_guard<std::mutex> lock(searchLock_);
         auto id = val->id;
         auto search = searches_.emplace(key, ProxySearch{}).first;
-        auto nextRefresh = scheduler.time() + proxy::OP_TIMEOUT;
+        auto nextRefresh = scheduler.time() + proxy::OP_TIMEOUT - proxy::OP_MARGIN;
         search->second.puts.erase(id);
         search->second.puts.emplace(id, PermanentPut {val, scheduler.add(nextRefresh, [this, key, id]{
+            std::lock_guard<std::mutex> lock(searchLock_);
             auto s = searches_.find(key);
             if (s != searches_.end()) {
                 auto p = s->second.puts.find(id);
                 if (p != s->second.puts.end()) {
                     doPut(key, p->second.value, {}, time_point::max(), true);
-                    scheduler.edit(p->second.refreshJob, scheduler.time() + proxy::OP_TIMEOUT);
+                    scheduler.edit(p->second.refreshJob, scheduler.time() + proxy::OP_TIMEOUT - proxy::OP_MARGIN);
                 }
             }
         })});
@@ -471,9 +480,12 @@ void
 DhtProxyClient::getProxyInfos()
 {
     DHT_LOG.d("Requesting proxy server node information");
+    std::lock_guard<std::mutex> l(statusLock_);
 
-    if (ongoingStatusUpdate_.test_and_set())
-        return;
+    auto infoState = std::make_shared<InfoState>();
+    if (infoState_)
+        infoState_->cancel = true;
+    infoState_ = infoState;
 
     {
         std::lock_guard<std::mutex> l(lockCurrentProxyInfos_);
@@ -488,11 +500,15 @@ DhtProxyClient::getProxyInfos()
 
     // Try to contact the proxy and set the status to connected when done.
     // will change the connectivity status
-    if (statusThread_.joinable())
-        statusThread_.join();
-    statusThread_ = std::thread([this, serverHost]{
-        struct SuccessCount {std::atomic_uint ipv4 {0}, ipv6 {0};};
-        auto successCount = std::make_shared<SuccessCount>();
+    if (statusThread_.joinable()) {
+        try {
+            statusThread_.detach();
+            statusThread_ = {};
+        } catch (const std::exception& e) {
+            DHT_LOG.e("Error detaching thread: %s", e.what());
+        }
+    }
+    statusThread_ = std::thread([this, serverHost, infoState]{
         try {
             auto hostAndService = splitPort(serverHost);
             auto resolved_proxies = SockAddr::resolve(hostAndService.first, hostAndService.second);
@@ -507,8 +523,10 @@ DhtProxyClient::getProxyInfos()
                 }
                 restbed::Uri uri(proxy::HTTP_PROTO + server + "/");
                 auto req = std::make_shared<restbed::Request>(uri);
+                if (infoState->cancel)
+                    return;
                 reqs.emplace_back(restbed::Http::async(req,
-                    [this, resolved_proxy, successCount](
+                    [this, resolved_proxy, infoState](
                                 const std::shared_ptr<restbed::Request>&,
                                 const std::shared_ptr<restbed::Response>& reply)
                 {
@@ -516,6 +534,8 @@ DhtProxyClient::getProxyInfos()
                     Json::Value proxyInfos;
                     if (code == 200) {
                         restbed::Http::fetch("\n", reply);
+                        auto& state = *infoState;
+                        if (state.cancel) return;
                         std::string body;
                         reply->get_body(body);
 
@@ -524,13 +544,14 @@ DhtProxyClient::getProxyInfos()
                         auto reader = std::unique_ptr<Json::CharReader>(rbuilder.newCharReader());
                         try {
                             reader->parse(body.data(), body.data() + body.size(), &proxyInfos, &err);
-                        } catch (const std::exception& e) {
-                            DHT_LOG.w("Error parsing proxy infos: %s", e.what());
+                        } catch (...) {
+                            return;
                         }
                         auto family = resolved_proxy.getFamily();
-                        if      (family == AF_INET)  successCount->ipv4++;
-                        else if (family == AF_INET6) successCount->ipv6++;
-                        onProxyInfos(proxyInfos, family);
+                        if      (family == AF_INET)  state.ipv4++;
+                        else if (family == AF_INET6) state.ipv6++;
+                        if (not state.cancel)
+                            onProxyInfos(proxyInfos, family);
                     }
                 }));
             }
@@ -540,22 +561,24 @@ DhtProxyClient::getProxyInfos()
         } catch (const std::exception& e) {
             DHT_LOG.e("Error sending proxy info request: %s", e.what());
         }
-        if (successCount->ipv4 == 0) onProxyInfos(Json::Value{}, AF_INET);
-        if (successCount->ipv6 == 0) onProxyInfos(Json::Value{}, AF_INET6);
-        ongoingStatusUpdate_.clear();
+        const auto& state = *infoState;
+        if (state.cancel) return;
+        if (state.ipv4 == 0) onProxyInfos(Json::Value{}, AF_INET);
+        if (state.ipv6 == 0) onProxyInfos(Json::Value{}, AF_INET6);
     });
 }
 
 void
 DhtProxyClient::onProxyInfos(const Json::Value& proxyInfos, sa_family_t family)
 {
-    DHT_LOG.d("Got proxy infos %s", family == AF_INET ? "IPv4" : "IPv6");
     std::lock_guard<std::mutex> l(lockCurrentProxyInfos_);
     auto oldStatus = std::max(statusIpv4_, statusIpv6_);
     auto& status = family == AF_INET ? statusIpv4_ : statusIpv6_;
     if (not proxyInfos.isMember("node_id")) {
+        DHT_LOG.e("Proxy info request failed for %s", family == AF_INET ? "IPv4" : "IPv6");
         status = NodeStatus::Disconnected;
     } else {
+        DHT_LOG.d("Got proxy reply for %s", family == AF_INET ? "IPv4" : "IPv6");
         try {
             myid = InfoHash(proxyInfos["node_id"].asString());
             stats4_ = NodeStats(proxyInfos["ipv4"]);
@@ -652,14 +675,104 @@ DhtProxyClient::cancelListen(const InfoHash& key, size_t gtoken) {
     return canceled;
 }
 
+
+void
+DhtProxyClient::sendListen(const std::shared_ptr<restbed::Request>& req, const ValueCallback& cb, const Value::Filter& filter, const Sp<ListenState>& state) {
+    auto settings = std::make_shared<restbed::Settings>();
+    std::chrono::milliseconds timeout(std::numeric_limits<int>::max());
+    settings->set_connection_timeout(timeout); // Avoid the client to close the socket after 5 seconds.
+    req->set_method("LISTEN");
+    restbed::Http::async(req,
+        [this, filter, cb, state](const std::shared_ptr<restbed::Request>& req,
+                                  const std::shared_ptr<restbed::Response>& reply) {
+        auto code = reply->get_status_code();
+        if (code == 200) {
+            try {
+                while (restbed::Http::is_open(req) and not state->cancel) {
+                    restbed::Http::fetch("\n", reply);
+                    if (state->cancel)
+                        break;
+                    std::string body;
+                    reply->get_body(body);
+                    reply->set_body(""); // Reset the body for the next fetch
+
+                    Json::Value json;
+                    std::string err;
+                    Json::CharReaderBuilder rbuilder;
+                    auto reader = std::unique_ptr<Json::CharReader>(rbuilder.newCharReader());
+                    if (reader->parse(body.data(), body.data() + body.size(), &json, &err)) {
+                        auto 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->cancel and not cb({value}, false))
+                                    state->cancel = true;
+                            });
+                            loopSignal_();
+                        }
+                    }
+                }
+            } catch (const std::exception& e) {
+                if (not state->cancel) {
+                    DHT_LOG.w("Listen closed by the proxy server: %s", e.what());
+                    state->ok = false;
+                }
+            }
+        } else {
+            state->ok = false;
+        }
+    }, settings).get();
+    auto& s = *state;
+    if (not s.ok and not s.cancel)
+        opFailed();
+}
+
+void
+DhtProxyClient::sendSubscribe(const std::shared_ptr<restbed::Request>& req, const Sp<proxy::ListenToken>& token, const Sp<ListenState>& state) {
+#if OPENDHT_PUSH_NOTIFICATIONS
+    req->set_method("SUBSCRIBE");
+    fillBodyToGetToken(req);
+    restbed::Http::async(req, [this,state, token](const std::shared_ptr<restbed::Request>&,
+                                             const std::shared_ptr<restbed::Response>& reply) {
+        auto code = reply->get_status_code();
+        if (code == 200) {
+            try {
+                restbed::Http::fetch("\n", reply);
+                std::string body;
+                reply->get_body(body);
+
+                std::string err;
+                Json::Value json;
+                Json::CharReaderBuilder rbuilder;
+                auto reader = std::unique_ptr<Json::CharReader>(rbuilder.newCharReader());
+                if (reader->parse(body.data(), body.data() + body.size(), &json, &err)) {
+                    if (not json.isMember("token")) {
+                        state->ok = false;
+                        return;
+                    }
+                    *token = unpackId(json, "token");
+                }
+            } catch (const std::exception& e) {
+                if (not state->cancel) {
+                    DHT_LOG.e("sendSubscribe: error: %s", e.what());
+                    state->ok = false;
+                }
+            }
+        } else {
+            state->ok = false;
+        }
+    }).get();
+    auto& s = *state;
+    if (not s.ok and not s.cancel)
+        opFailed();
+#endif
+}
+
 size_t
 DhtProxyClient::doListen(const InfoHash& key, ValueCallback cb, Value::Filter filter/*, Where where*/)
 {
     scheduler.syncTime();
     restbed::Uri uri(proxy::HTTP_PROTO + serverHost_ + "/" + key.toString());
-    auto req = std::make_shared<restbed::Request>(uri);
-    req->set_method(deviceKey_.empty() ? "LISTEN" : "SUBSCRIBE");
-
     std::lock_guard<std::mutex> lock(searchLock_);
     auto search = searches_.find(key);
     if (search == searches_.end()) {
@@ -668,6 +781,7 @@ DhtProxyClient::doListen(const InfoHash& key, ValueCallback cb, Value::Filter fi
     }
     DHT_LOG.d(key, "[search %s] sending %s", key.to_c_str(), deviceKey_.empty() ? "listen" : "subscribe");
 
+    auto req = std::make_shared<restbed::Request>(uri);
     auto token = ++listenerToken_;
     auto l = search->second.listeners.find(token);
     if (l == search->second.listeners.end()) {
@@ -687,9 +801,12 @@ DhtProxyClient::doListen(const InfoHash& key, ValueCallback cb, Value::Filter fi
                 scheduler.edit(l->second.cacheExpirationJob, next);
             }), req, std::move(f)
         }).first;
+    } else {
+        if (l->second.state)
+            l->second.state->cancel = true;
     }
 
-    auto state = std::make_shared<Listener::State>();
+    auto state = std::make_shared<ListenState>();
     l->second.state = state;
     l->second.cb = [this,key,token,state](const std::vector<Sp<Value>>& values, bool expired) {
         if (state->cancel)
@@ -715,82 +832,34 @@ DhtProxyClient::doListen(const InfoHash& key, ValueCallback cb, Value::Filter fi
         loopSignal_();
         return true;
     };
+
     auto pushNotifToken = std::make_shared<proxy::ListenToken>(0);
     auto vcb = l->second.cb;
     l->second.pushNotifToken = pushNotifToken;
-    l->second.thread = std::thread([this,req,filter,vcb,pushNotifToken,state]()
-        {
-            auto settings = std::make_shared<restbed::Settings>();
-            if (deviceKey_.empty()) {
-                std::chrono::milliseconds timeout(std::numeric_limits<int>::max());
-                settings->set_connection_timeout(timeout); // Avoid the client to close the socket after 5 seconds.
-            }
-#if OPENDHT_PUSH_NOTIFICATIONS
-            else
-                fillBodyToGetToken(req);
-#endif
+    l->second.req = req;
 
-            restbed::Http::async(req,
-                [this, filter, vcb, pushNotifToken, 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 {
-                        std::string err;
-                        Json::Value json;
-                        Json::CharReaderBuilder rbuilder;
-                        auto reader = std::unique_ptr<Json::CharReader>(rbuilder.newCharReader());
-                        if (!deviceKey_.empty()) {
-                            restbed::Http::fetch("\n", reply);
-                            if (state->cancel)
-                                return;
-                            std::string body;
-                            reply->get_body(body);
-
-                            auto* char_data = reinterpret_cast<const char*>(&body[0]);
-                            if (reader->parse(char_data, char_data + body.size(), &json, &err)) {
-                                *pushNotifToken = unpackId(json, "token");
-                            } else {
-                                state->ok = false;
-                            }
-                        } else {
-                            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
-
-                                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)) and vcb)  {
-                                        std::lock_guard<std::mutex> lock(lockCallbacks);
-                                        callbacks_.emplace_back([vcb, value, state]() {
-                                            if (not state->cancel and not vcb({value}, false))
-                                                state->cancel = true;
-                                        });
-                                        loopSignal_();
-                                    }
-                                } else {
-                                    state->ok = false;
-                                }
-                            }
-                        }
-                    } catch (std::runtime_error&) {
-                        // NOTE: Http::close() can occurs here. Ignore this.
-                    }
-                } else {
-                    state->ok = false;
+    if (not deviceKey_.empty()) {
+        // Relaunch push listeners even if a timeout is not received (if the proxy crash for any reason)
+        l->second.refreshJob = scheduler.add(scheduler.time() + proxy::OP_TIMEOUT - proxy::OP_MARGIN, [this, key, token, state] {
+            if (state->cancel)
+                return;
+            std::lock_guard<std::mutex> lock(searchLock_);
+            auto s = searches_.find(key);
+            if (s != searches_.end()) {
+                auto l = s->second.listeners.find(token);
+                if (l != s->second.listeners.end()) {
+                    resubscribe(key, l->second);
                 }
-            }, settings).get();
-            auto& s = *state;
-            if (not s.ok and not s.cancel) {
-                opFailed();
             }
-        }
-    );
+        });
+        l->second.thread = std::thread([this, req,pushNotifToken, state](){
+            sendSubscribe(req,pushNotifToken,state);
+        });
+    } else {
+        l->second.thread = std::thread([this, req, vcb, filter, state]{
+            sendListen(req,vcb,filter,state);
+        });
+    }
     return token;
 }
 
@@ -880,75 +949,28 @@ DhtProxyClient::restartListeners()
     DHT_LOG.d("Restarting listeners");
     std::lock_guard<std::mutex> lock(searchLock_);
     for (auto& search: searches_) {
-    for (auto& l: search.second.listeners) {
-        auto& listener = l.second;
-        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, state]()
-            {
-                auto settings = std::make_shared<restbed::Settings>();
-                std::chrono::milliseconds timeout(std::numeric_limits<int>::max());
-                settings->set_connection_timeout(timeout); // Avoid the client to close the socket after 5 seconds.
-
-                auto ok = std::make_shared<std::atomic_bool>(true);
-                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 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->cancel and not cb({value}, false))
-                                                state->cancel = true;
-                                        });
-                                        loopSignal_();
-                                    }
-                                }
-                            }
-                        } catch (std::runtime_error&) {
-                            // NOTE: Http::close() can occurs here. Ignore this.
-                        }
-                    } else {
-                        state->ok = false;
-                    }
-                }, settings).get();
-                auto& s = *state;
-                if (not s.ok and not s.cancel)
-                    opFailed();
+        for (auto& l: search.second.listeners) {
+            auto& listener = l.second;
+            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, req, cb, filter, state]() {
+                sendListen(req, cb, filter, state);
+            });
+        }
     }
 }
 
@@ -957,6 +979,12 @@ DhtProxyClient::pushNotificationReceived(const std::map<std::string, std::string
 {
 #if OPENDHT_PUSH_NOTIFICATIONS
     scheduler.syncTime();
+    {
+        // If a push notification is received, the proxy is up and running
+        std::lock_guard<std::mutex> l(lockCurrentProxyInfos_);
+        statusIpv4_ = NodeStatus::Connected;
+        statusIpv6_ = NodeStatus::Connected;
+    }
     try {
         std::lock_guard<std::mutex> lock(searchLock_);
         auto timeout = notification.find("timeout");
@@ -967,7 +995,7 @@ 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);
                 scheduler.edit(put.refreshJob, scheduler.time());
                 loopSignal_();
             } else {
@@ -1002,6 +1030,7 @@ DhtProxyClient::resubscribe(const InfoHash& key, Listener& listener)
 {
 #if OPENDHT_PUSH_NOTIFICATIONS
     if (deviceKey_.empty()) return;
+    scheduler.syncTime();
     DHT_LOG.d(key, "[search %s] resubscribe push listener", key.to_c_str());
     // Subscribe
     restbed::Uri uri(proxy::HTTP_PROTO + serverHost_ + "/" + key.toString());
@@ -1016,37 +1045,9 @@ DhtProxyClient::resubscribe(const InfoHash& key, Listener& listener)
     state->ok = true;
     listener.req = req;
     listener.pushNotifToken = pushNotifToken;
+    scheduler.edit(listener.refreshJob, scheduler.time() + proxy::OP_TIMEOUT - proxy::OP_MARGIN);
     listener.thread = std::thread([this, req, pushNotifToken, state]() {
-        fillBodyToGetToken(req);
-        auto settings = std::make_shared<restbed::Settings>();
-        restbed::Http::async(req,
-            [pushNotifToken, state](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 (!json.isMember("token")) return;
-                        *pushNotifToken = unpackId(json, "token");
-                    }
-                } catch (std::runtime_error&) {
-                    // NOTE: Http::close() can occurs here. Ignore this.
-                }
-            } else {
-                state->ok = false;
-            }
-        }, settings).get();
-        auto& s = *state;
-        if (not s.ok and not s.cancel)
-            opFailed();
+        sendSubscribe(req, pushNotifToken, state);
     });
 #endif
 }

src/dht_proxy_server.cpp

@@ -361,7 +361,10 @@ DhtProxyServer::subscribe(const std::shared_ptr<restbed::Session>& session)
                     return;
                 }
                 auto pushToken = root["key"].asString();
-                if (pushToken.empty()) return;
+                if (pushToken.empty()) {
+                    s->close(restbed::BAD_REQUEST, "{\"err\":\"No token\"}");
+                    return;
+                }
                 auto tokenFromReq = unpackId(root, "token");
                 auto platform = root["platform"].asString();
                 auto isAndroid = platform == "android";
@@ -588,8 +591,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;
@@ -606,28 +609,30 @@ DhtProxyServer::put(const std::shared_ptr<restbed::Session>& session)
                                 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;
+                                    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()) {
+                                        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 << " " << vid << 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);
+                                    scheduler_.edit(pput.expireJob, timeout);
+                                    if (pput.expireNotifyJob)
+                                        scheduler_.edit(pput.expireNotifyJob, timeout - proxy::OP_MARGIN);
                                 }
                                 lock.unlock();
                                 schedulerCv_.notify_one();
@@ -652,7 +657,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

@@ -247,10 +247,10 @@ void
 DhtRunner::setLoggers(LogMethod error, LogMethod warn, LogMethod debug) {
     std::lock_guard<std::mutex> lck(dht_mtx);
     if (dht_)
-        dht_->setLoggers(std::forward<LogMethod>(error), std::forward<LogMethod>(warn), std::forward<LogMethod>(debug));
+        dht_->setLoggers(error, warn, debug);
 #if OPENDHT_PROXY_CLIENT
     if (dht_via_proxy_)
-        dht_via_proxy_->setLoggers(std::forward<LogMethod>(error), std::forward<LogMethod>(warn), std::forward<LogMethod>(debug));
+        dht_via_proxy_->setLoggers(error, warn, debug);
 #endif
 }
 
@@ -342,7 +342,9 @@ std::vector<SockAddr>
 DhtRunner::getPublicAddress(sa_family_t af)
 {
     std::lock_guard<std::mutex> lck(dht_mtx);
-    return activeDht()->getPublicAddress(af);
+    if (auto dht = activeDht())
+        return dht->getPublicAddress(af);
+    return {};
 }
 std::vector<std::string>
 DhtRunner::getPublicAddressStr(sa_family_t af)
@@ -599,20 +601,24 @@ DhtRunner::listen(const std::string& key, GetCallback vcb, Value::Filter f, Wher
 void
 DhtRunner::cancelListen(InfoHash h, size_t token)
 {
-    std::lock_guard<std::mutex> lck(storage_mtx);
-    pending_ops.emplace([=](SecureDht& dht) {
+    {
+        std::lock_guard<std::mutex> lck(storage_mtx);
 #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);
+        pending_ops.emplace([=](SecureDht&) {
+            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);
+        pending_ops.emplace([=](SecureDht& dht) {
+            dht.cancelListen(h, token);
+        });
 #endif // OPENDHT_PROXY_CLIENT
-    });
+    }
     cv.notify_all();
 }
 
@@ -621,20 +627,21 @@ DhtRunner::cancelListen(InfoHash h, std::shared_future<size_t> ftoken)
 {
     {
         std::lock_guard<std::mutex> lck(storage_mtx);
-        pending_ops.emplace([=](SecureDht& dht) {
-            auto token = ftoken.get();
 #if OPENDHT_PROXY_CLIENT
-            auto it = listeners_.find(token);
+        pending_ops.emplace([=](SecureDht&) {
+            auto it = listeners_.find(ftoken.get());
             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
         });
+#else
+        pending_ops.emplace([=](SecureDht& dht) {
+            dht.cancelListen(h, ftoken.get());
+        });
+#endif // OPENDHT_PROXY_CLIENT
     }
     cv.notify_all();
 }
@@ -851,7 +858,7 @@ DhtRunner::connectivityChanged()
 {
     {
         std::lock_guard<std::mutex> lck(storage_mtx);
-        pending_ops.emplace([=](SecureDht& dht) {
+        pending_ops_prio.emplace([=](SecureDht& dht) {
             dht.connectivityChanged();
         });
     }
@@ -971,7 +978,7 @@ DhtRunner::pushNotificationReceived(const std::map<std::string, std::string>& da
 #if OPENDHT_PROXY_CLIENT && OPENDHT_PUSH_NOTIFICATIONS
     {
         std::lock_guard<std::mutex> lck(storage_mtx);
-        pending_ops_prio.emplace([=](SecureDht& dht) {
+        pending_ops_prio.emplace([=](SecureDht&) {
             if (dht_via_proxy_)
                 dht_via_proxy_->pushNotificationReceived(data);
         });

src/network_engine.cpp

@@ -262,7 +262,13 @@ NetworkEngine::isRunning(sa_family_t af) const
 
 void
 NetworkEngine::clear()
-{}
+{
+    for (auto& request : requests) {
+        request.second->cancel();
+        request.second->node->setExpired();
+    }
+    requests.clear();
+}
 
 void
 NetworkEngine::connectivityChanged(sa_family_t af)
@@ -292,24 +298,21 @@ NetworkEngine::requestStep(Sp<Request> sreq)
     auto err = send((char*)req.msg.data(), req.msg.size(),
             (node.getReplyTime() >= now - UDP_REPLY_TIME) ? 0 : MSG_CONFIRM,
             node.getAddr());
-    std::weak_ptr<Request> wreq = sreq;
     if (err == ENETUNREACH  ||
         err == EHOSTUNREACH ||
         err == EAFNOSUPPORT ||
-        err == EPIPE)
+        err == EPIPE        ||
+        err == EPERM)
     {
-        scheduler.add(now, [this,wreq]() mutable {
-            if (auto req = wreq.lock()) {
-                req->node->setExpired();
-                if (not req->node->id)
-                    requests.erase(req->tid);
-            }
-        });
+        node.setExpired();
+        if (not node.id)
+            requests.erase(req.tid);
     } else {
         if (err != EAGAIN) {
             ++req.attempt_count;
         }
         req.last_try = now;
+        std::weak_ptr<Request> wreq = sreq;
         scheduler.add(req.last_try + Node::MAX_RESPONSE_TIME, [this,wreq] {
             if (auto req = wreq.lock())
                 requestStep(req);

src/request.h

@@ -57,6 +57,15 @@ struct Request {
     bool pending() const { return state_ == State::PENDING; }
     bool over() const { return not pending(); }
     State getState() const { return state_; }
+    char getStateChar() const {
+        switch (state_) {
+            case State::PENDING:   return 'f';
+            case State::CANCELLED: return 'c';
+            case State::EXPIRED:   return 'e';
+            case State::COMPLETED: return 'a';
+            default:               return '?';
+        }
+    }
 
     Request(State state = State::PENDING) : state_(state) {}
     Request(MessageType type, Tid tid,

src/search.h

@@ -331,7 +331,7 @@ struct Dht::SearchNode {
         return ((gs != getStatus.cend() and gs->second and gs->second->pending())
                 or ack == acked.cend() or not ack->second.first or ack->second.first->pending())
                 ? time_point::max()
-                : ack->second.second - REANNOUNCE_MARGIN;
+                : (ack->second.first->completed() ? ack->second.second - REANNOUNCE_MARGIN : time_point::min());
     }
 
     /**

tests/Makefile.am

@@ -3,7 +3,7 @@ bin_PROGRAMS = opendht_unit_tests
 
 AM_CPPFLAGS = -I../include
 
-nobase_include_HEADERS = infohashtester.h dhtrunnertester.h
-opendht_unit_tests_SOURCES = tests_runner.cpp infohashtester.cpp dhtrunnertester.cpp
+nobase_include_HEADERS = infohashtester.h dhtrunnertester.h dhtproxytester.h
+opendht_unit_tests_SOURCES = tests_runner.cpp infohashtester.cpp dhtrunnertester.cpp dhtproxytester.cpp
 opendht_unit_tests_LDFLAGS = -lopendht -lcppunit -L@top_builddir@/src/.libs @GnuTLS_LIBS@
 endif

tests/dhtproxytester.cpp

@@ -0,0 +1,129 @@
+/*
+ *  Copyright (C) 2018 Savoir-faire Linux Inc.
+ *
+ *  Author: Sébastien Blin <sebastien.blin@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 "dhtproxytester.h"
+
+// std
+#include <iostream>
+#include <string>
+
+#include <chrono>
+#include <condition_variable>
+
+
+namespace test {
+CPPUNIT_TEST_SUITE_REGISTRATION(DhtProxyTester);
+
+void
+DhtProxyTester::setUp() {
+    nodePeer.run(42222, {}, true);
+    nodeProxy = std::make_shared<dht::DhtRunner>();
+    nodeClient = std::make_shared<dht::DhtRunner>();
+
+    nodeProxy->run(42232, {}, true);
+    nodeProxy->bootstrap(nodePeer.getBound());
+    server = std::unique_ptr<dht::DhtProxyServer>(new dht::DhtProxyServer(nodeProxy, 8080));
+
+    nodeClient->run(42242, {}, true);
+    nodeClient->bootstrap(nodePeer.getBound());
+    nodeClient->setProxyServer("127.0.0.1:8080");
+    nodeClient->enableProxy(true);
+}
+
+void
+DhtProxyTester::tearDown() {
+    nodePeer.join();
+    nodeClient->join();
+    server->stop();
+    server = nullptr;
+    nodeProxy->join();
+}
+
+void
+DhtProxyTester::testGetPut() {
+    bool done = false;
+    std::condition_variable cv;
+    std::mutex cv_m;
+
+    auto key = dht::InfoHash::get("GLaDOs");
+    dht::Value val {"Hey! It's been a long time. How have you been?"};
+    auto val_data = val.data;
+
+    nodePeer.put(key, std::move(val), [&](bool) {
+        done = true;
+        cv.notify_all();
+    });
+    std::unique_lock<std::mutex> lk(cv_m);
+    cv.wait_for(lk, std::chrono::seconds(10), [&]{ return done; });
+
+    auto vals = nodeClient->get(key).get();
+    CPPUNIT_ASSERT(not vals.empty());
+    CPPUNIT_ASSERT(vals.front()->data == val_data);
+}
+
+
+void
+DhtProxyTester::testListen() {
+    bool done = false;
+    std::condition_variable cv;
+    std::mutex cv_m;
+    std::unique_lock<std::mutex> lk(cv_m);
+    auto key = dht::InfoHash::get("GLaDOs");
+
+    // If a peer send a value, the listen operation from the client
+    // should retrieve this value
+    dht::Value firstVal {"Hey! It's been a long time. How have you been?"};
+    auto firstVal_data = firstVal.data;
+    nodePeer.put(key, std::move(firstVal), [&](bool) {
+        done = true;
+        cv.notify_all();
+    });
+    cv.wait_for(lk, std::chrono::seconds(10), [&]{ return done; });
+    done = false;
+
+    auto values = std::vector<dht::Blob>();
+    nodeClient->listen(key, [&](const std::vector<std::shared_ptr<dht::Value>>& v, bool) {
+        for (const auto& value : v)
+            values.emplace_back(value->data);
+        done = true;
+        cv.notify_all();
+        return true;
+    });
+
+    cv.wait_for(lk, std::chrono::seconds(10), [&]{ return done; });
+    done = false;
+    // Here values should contains 2 values
+    CPPUNIT_ASSERT_EQUAL(static_cast<int>(values.size()), 1);
+    CPPUNIT_ASSERT(values.front() == firstVal_data);
+
+    // And the listen should retrieve futures values
+    // All values
+    dht::Value secondVal {"You're a monster"};
+    auto secondVal_data = secondVal.data;
+    nodePeer.put(key, std::move(secondVal), [&](bool) {
+        done = true;
+        cv.notify_all();
+    });
+    cv.wait_for(lk, std::chrono::seconds(10), [&]{ return done; });
+    // Here values should contains 3 values
+    CPPUNIT_ASSERT_EQUAL(static_cast<int>(values.size()), 2);
+    CPPUNIT_ASSERT(values.back() == secondVal_data);
+}
+
+}  // namespace test

tests/dhtproxytester.h

@@ -0,0 +1,64 @@
+/*
+ *  Copyright (C) 2018 Savoir-faire Linux Inc.
+ *
+ *  Author: Sébastien Blin <sebastien.blin@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/>.
+ */
+
+#pragma once
+
+// cppunit
+#include <cppunit/TestFixture.h>
+#include <cppunit/extensions/HelperMacros.h>
+
+#include <opendht/dhtrunner.h>
+#include <opendht/dht_proxy_server.h>
+
+namespace test {
+
+class DhtProxyTester : public CppUnit::TestFixture {
+    CPPUNIT_TEST_SUITE(DhtProxyTester);
+    CPPUNIT_TEST(testGetPut);
+    CPPUNIT_TEST(testListen);
+    CPPUNIT_TEST_SUITE_END();
+
+ public:
+    /**
+     * Method automatically called before each test by CppUnit
+     * Init nodes
+     */
+    void setUp();
+    /**
+     * Method automatically called after each test CppUnit
+     */
+    void tearDown();
+    /**
+     * Test get and put methods
+     */
+    void testGetPut();
+    /**
+     * Test listen
+     */
+    void testListen();
+
+ private:
+    dht::DhtRunner nodePeer {};
+
+    std::shared_ptr<dht::DhtRunner> nodeClient;
+    std::shared_ptr<dht::DhtRunner> nodeProxy;
+    std::unique_ptr<dht::DhtProxyServer> server;
+};
+
+}  // namespace test

tools/dhtnode.cpp

@@ -199,6 +199,7 @@ void cmd_loop(std::shared_ptr<DhtRunner>& dht, dht_params& params
             continue;
         } else if (op == "cc") {
             dht->connectivityChanged();
+            continue;
         }
 #if OPENDHT_PROXY_SERVER
         else if (op == "pst") {
@@ -496,7 +497,7 @@ main(int argc, char **argv)
             crt = dht::crypto::generateIdentity("DHT Node", ca_tmp);
         }
 
-        dht::DhtRunner::Config config;
+        dht::DhtRunner::Config config {};
         config.dht_config.node_config.network = params.network;
         config.dht_config.node_config.maintain_storage = false;
         config.dht_config.id = crt;