update to version 1.7.2

proxy: set status to CONNECTED when receiving a push notifications

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

@@ -278,6 +278,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);
 
     /**

include/opendht/proxy.h

@@ -22,8 +22,8 @@
 namespace dht {
 namespace proxy {
 
-constexpr const std::chrono::minutes OP_TIMEOUT {1 * 10}; // one hour
-constexpr const std::chrono::minutes OP_MARGIN {5}; // 5 minutes
+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 char* const HTTP_PROTO {"http://"};
 using ListenToken = uint64_t;
 

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,13 @@
 
 namespace dht {
 
+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 +47,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)
@@ -322,19 +323,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]{
-            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()) {
                 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, p->second.value, {}, time_point::max(), true);
+                    scheduler.edit(p->second.refreshJob, scheduler.time() + proxy::OP_TIMEOUT - proxy::OP_MARGIN);
                 }
             }
         })});
@@ -656,14 +654,102 @@ 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 (std::runtime_error&) {
+                if (not state->cancel) {
+                    DHT_LOG.w("Listen closed by the proxy server");
+                    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);
+    auto settings = std::make_shared<restbed::Settings>();
+    restbed::Http::async(req, [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 (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();
+#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()) {
@@ -672,6 +758,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()) {
@@ -691,9 +778,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)
@@ -719,82 +809,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;
 }
 
@@ -884,75 +926,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);
+            });
+        }
     }
 }
 
@@ -961,9 +956,11 @@ 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());
+    {
+        // 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_);
@@ -976,7 +973,6 @@ DhtProxyClient::pushNotificationReceived(const std::map<std::string, std::string
                 // Refresh put
                 auto vid = std::stoull(vidIt->second);
                 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 {
@@ -1011,6 +1007,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());
@@ -1025,37 +1022,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

@@ -599,7 +599,6 @@ 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();
@@ -609,18 +608,16 @@ DhtProxyServer::put(const std::shared_ptr<restbed::Session>& session)
                                 auto r = sPuts->second.puts.emplace(vid, PermanentPut{});
                                 auto& pput = r.first->second;
                                 if (r.second) {
-                                    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()) {
-                                        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::to_string(vid) << " " << clientId << std::endl;
+                                            std::cout << "Permanent put refresh: " << infoHash << " " << vid << std::endl;
                                             Json::Value json;
                                             json["timeout"] = infoHash.toString();
                                             json["to"] = clientId;
@@ -630,7 +627,6 @@ DhtProxyServer::put(const std::shared_ptr<restbed::Session>& session)
                                     }
 #endif
                                 } else {
-                                    std::cout << "Refreshing put jobs" << std::endl;
                                     scheduler_.edit(pput.expireJob, timeout);
                                     if (pput.expireNotifyJob)
                                         scheduler_.edit(pput.expireNotifyJob, timeout - proxy::OP_MARGIN);

src/dhtrunner.cpp

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

@@ -292,24 +292,20 @@ 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)
     {
-        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") {