wip

CMakeLists.txt

@@ -1,7 +1,7 @@
 cmake_minimum_required (VERSION 3.1)
 project (opendht)
 set (opendht_VERSION_MAJOR 1)
-set (opendht_VERSION_MINOR 7.2)
+set (opendht_VERSION_MINOR 7.1)
 set (opendht_VERSION ${opendht_VERSION_MAJOR}.${opendht_VERSION_MINOR})
 set (PACKAGE_VERSION ${opendht_VERSION})
 set (VERSION "${opendht_VERSION}")
@@ -304,22 +304,13 @@ install (FILES ${CMAKE_CURRENT_BINARY_DIR}/opendhtConfigVersion.cmake DESTINATIO
 IF(OPENDHT_TESTS)
   FIND_PACKAGE(Cppunit REQUIRED)
   # unit testing
-    list (APPEND test_FILES
+    add_executable(opendht_unit_tests
+      tests/tests_runner.cpp
       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], 2)
+m4_define([opendht_patch_version], 1)
 m4_define([opendht_version],
 		  [opendht_major_version.opendht_minor_version.opendht_patch_version])
 

include/opendht/default_types.h

@@ -114,8 +114,6 @@ 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;
@@ -125,11 +123,10 @@ 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, datatype)
+    MSGPACK_DEFINE_MAP(id, msg, date, status)
 };
 
 class OPENDHT_PUBLIC TrustRequest : public EncryptedValue<TrustRequest>

include/opendht/dht_proxy_client.h

@@ -278,10 +278,6 @@ 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::seconds OP_TIMEOUT {1 * 60 * 60}; // onw hour
-constexpr const std::chrono::seconds OP_MARGIN {5 * 60}; // 5 minutes
+constexpr const std::chrono::minutes OP_TIMEOUT {1 * 10}; // one hour
+constexpr const std::chrono::minutes OP_MARGIN {5}; // 5 minutes
 constexpr const char* const HTTP_PROTO {"http://"};
 using ListenToken = uint64_t;
 

src/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;
-        auto new_req = network_engine.sendListen(n.node, sr->id, *query, n.token, prev_req,
+        r->second.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,11 +545,6 @@ 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;
-        }
     }
 }
 
@@ -1540,7 +1535,10 @@ Dht::dumpSearch(const Search& sr, std::ostream& out) const
                     if (ack == n.acked.end() or not ack->second.first) {
                         out << ' ';
                     } else {
-                        out << ack->second.first->getStateChar();
+                        if (ack->second.first->completed())
+                            out << 'a';
+                        else if (ack->second.first->pending())
+                            out << 'f';
                     }
                 }
                 out << "] ";

src/dht_proxy_client.cpp

@@ -33,13 +33,12 @@
 
 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;
@@ -47,7 +46,7 @@ struct DhtProxyClient::Listener
     Sp<restbed::Request> req;
     std::thread thread;
     unsigned callbackId;
-    Sp<ListenState> state;
+    Sp<State> state;
     Sp<proxy::ListenToken> pushNotifToken;
     Sp<Scheduler::Job> refreshJob;
     Listener(ValueCache&& c, Sp<Scheduler::Job>&& j, const Sp<restbed::Request>& r, Value::Filter&& f)
@@ -323,16 +322,19 @@ 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 - proxy::OP_MARGIN;
+        auto nextRefresh = scheduler.time() + proxy::OP_TIMEOUT;
         search->second.puts.erase(id);
         search->second.puts.emplace(id, PermanentPut {val, scheduler.add(nextRefresh, [this, key, id]{
+            DHT_LOG.w(key, "[search %s]: PermanentPut refresh callback vid %llu", key.to_c_str(), id);
             std::lock_guard<std::mutex> lock(searchLock_);
             auto s = searches_.find(key);
             if (s != searches_.end()) {
                 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 - proxy::OP_MARGIN);
+                    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);
                 }
             }
         })});
@@ -654,102 +656,14 @@ 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()) {
@@ -758,7 +672,6 @@ 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()) {
@@ -778,12 +691,9 @@ 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<ListenState>();
+    auto state = std::make_shared<Listener::State>();
     l->second.state = state;
     l->second.cb = [this,key,token,state](const std::vector<Sp<Value>>& values, bool expired) {
         if (state->cancel)
@@ -809,34 +719,82 @@ 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.req = req;
-
-    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);
-                }
+    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.
             }
-        });
-        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);
-        });
-    }
+#if OPENDHT_PUSH_NOTIFICATIONS
+            else
+                fillBodyToGetToken(req);
+#endif
+
+            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;
+                }
+            }, settings).get();
+            auto& s = *state;
+            if (not s.ok and not s.cancel) {
+                opFailed();
+            }
+        }
+    );
     return token;
 }
 
@@ -926,28 +884,75 @@ 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, req, cb, filter, state]() {
-                sendListen(req, cb, filter, state);
-            });
+    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();
+            }
+        );
+    }
     }
 }
 
@@ -956,11 +961,9 @@ 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;
+    DHT_LOG.d("Got push notification");
+    for (const auto& n : notification) {
+        DHT_LOG.d("   %s -> %s", n.first.c_str(), n.second.c_str());
     }
     try {
         std::lock_guard<std::mutex> lock(searchLock_);
@@ -973,6 +976,7 @@ 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 {
@@ -1007,7 +1011,6 @@ 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());
@@ -1022,9 +1025,37 @@ 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]() {
-        sendSubscribe(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();
     });
 #endif
 }

src/dht_proxy_server.cpp

@@ -599,6 +599,7 @@ 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();
@@ -608,16 +609,18 @@ 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 << " " << vid << std::endl;
+                                            std::cout << "Permanent put refresh: " << infoHash << " " << std::to_string(vid) << " " << clientId << std::endl;
                                             Json::Value json;
                                             json["timeout"] = infoHash.toString();
                                             json["to"] = clientId;
@@ -627,6 +630,7 @@ 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,9 +342,7 @@ std::vector<SockAddr>
 DhtRunner::getPublicAddress(sa_family_t af)
 {
     std::lock_guard<std::mutex> lck(dht_mtx);
-    if (auto dht = activeDht())
-        return dht->getPublicAddress(af);
-    return {};
+    return activeDht()->getPublicAddress(af);
 }
 std::vector<std::string>
 DhtRunner::getPublicAddressStr(sa_family_t af)
@@ -601,24 +599,20 @@ 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);
+    std::lock_guard<std::mutex> lck(storage_mtx);
+    pending_ops.emplace([=](SecureDht& dht) {
 #if OPENDHT_PROXY_CLIENT
-        pending_ops.emplace([=](SecureDht&) {
-            auto it = listeners_.find(token);
-            if (it == listeners_.end()) return;
-            if (it->second.tokenClassicDht)
-                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);
-        });
+        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
-        pending_ops.emplace([=](SecureDht& dht) {
-            dht.cancelListen(h, token);
-        });
+        dht.cancelListen(h, token);
 #endif // OPENDHT_PROXY_CLIENT
-    }
+    });
     cv.notify_all();
 }
 
@@ -627,21 +621,20 @@ 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
-        pending_ops.emplace([=](SecureDht&) {
-            auto it = listeners_.find(ftoken.get());
+            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
-        pending_ops.emplace([=](SecureDht& dht) {
-            dht.cancelListen(h, ftoken.get());
-        });
+            dht.cancelListen(h, token);
 #endif // OPENDHT_PROXY_CLIENT
+        });
     }
     cv.notify_all();
 }
@@ -858,7 +851,7 @@ DhtRunner::connectivityChanged()
 {
     {
         std::lock_guard<std::mutex> lck(storage_mtx);
-        pending_ops_prio.emplace([=](SecureDht& dht) {
+        pending_ops.emplace([=](SecureDht& dht) {
             dht.connectivityChanged();
         });
     }
@@ -978,7 +971,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&) {
+        pending_ops_prio.emplace([=](SecureDht& dht) {
             if (dht_via_proxy_)
                 dht_via_proxy_->pushNotificationReceived(data);
         });

src/network_engine.cpp

@@ -292,20 +292,24 @@ 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)
     {
-        node.setExpired();
-        if (not node.id)
-            requests.erase(req.tid);
+        scheduler.add(now, [this,wreq]() mutable {
+            if (auto req = wreq.lock()) {
+                req->node->setExpired();
+                if (not req->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,15 +57,6 @@ 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.first->completed() ? ack->second.second - REANNOUNCE_MARGIN : time_point::min());
+                : ack->second.second - REANNOUNCE_MARGIN;
     }
 
     /**

tests/Makefile.am

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

tests/dhtproxytester.cpp

@@ -1,129 +0,0 @@
-/*
- *  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

@@ -1,64 +0,0 @@
-/*
- *  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,7 +199,6 @@ 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") {