wip

proxy server, fix push listeners and avoid flooding nodes

CMakeLists.txt

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

README.md

@@ -6,12 +6,17 @@
 
 A lightweight C++11 Distributed Hash Table implementation.
 
- * Light and fast C++11 Kademlia DHT library
- * Distributed shared key->value data-store
- * Clean and powerful distributed map API with storage of arbitrary binary values of up to 56 KB
+OpenDHT provides an easy to use distributed in-memory data store.
+Every node in the network can read and write values to the store.
+Values are distributed over the network, with redundancy.
+
+ * Lightweight and scalable, designed for large networks and small devices
+ * High resilience to network disruption
  * Public key cryptography layer providing optional data signature and encryption (using GnuTLS)
  * IPv4 and IPv6 support
- * Python binding
+ * Clean and powerful C++11 map API
+ * Python 3 bindings
+ * REST API
 
 ## Documentation
 See the wiki: <https://github.com/savoirfairelinux/opendht/wiki>

configure.ac

@@ -1,7 +1,7 @@
 dnl define macros
 m4_define([opendht_major_version], 1)
 m4_define([opendht_minor_version], 7)
-m4_define([opendht_patch_version], 0)
+m4_define([opendht_patch_version], 1)
 m4_define([opendht_version],
 		  [opendht_major_version.opendht_minor_version.opendht_patch_version])
 

include/opendht/dht.h

@@ -414,7 +414,7 @@ private:
     void expireStorage(InfoHash h);
     void expireStore(decltype(store)::iterator);
 
-    void storageChanged(const InfoHash& id, Storage& st, ValueStorage&);
+    void storageChanged(const InfoHash& id, Storage& st, ValueStorage&, bool newValue);
     std::string printStorageLog(const decltype(store)::value_type&) const;
 
     /**
@@ -535,7 +535,9 @@ private:
      *
      * @param sr  The search to execute its operations.
      */
-    void searchStep(Sp<Search> sr);
+    void searchStep(Sp<Search>);
+    void searchSynchedNodeListen(const Sp<Search>&, SearchNode&);
+
     void dumpSearch(const Search& sr, std::ostream& out) const;
 
     bool neighbourhoodMaintenance(RoutingTable&);

include/opendht/dht_proxy_client.h

@@ -29,6 +29,7 @@
 #include "def.h"
 #include "dht_interface.h"
 #include "scheduler.h"
+#include "proxy.h"
 
 namespace restbed {
     class Request;
@@ -205,18 +206,18 @@ public:
     /**
      * Get data currently being put at the given hash.
      */
-    std::vector<Sp<Value>> getPut(const InfoHash&) { return {}; }
+    std::vector<Sp<Value>> getPut(const InfoHash&);
 
     /**
      * Get data currently being put at the given hash with the given id.
      */
-    Sp<Value> getPut(const InfoHash&, const Value::Id&) { return {}; }
+    Sp<Value> getPut(const InfoHash&, const Value::Id&);
 
     /**
      * Stop any put/announce operation at the given location,
      * for the value with the given id.
      */
-    bool cancelPut(const InfoHash&, const Value::Id&) { return false; }
+    bool cancelPut(const InfoHash&, const Value::Id&);
 
     void pingNode(const sockaddr*, socklen_t, DoneCallbackSimple&& /*cb*/={}) { }
 
@@ -249,8 +250,14 @@ public:
     void dumpTables() const {}
     std::vector<unsigned> getNodeMessageStats(bool) { return {}; }
     void setStorageLimit(size_t) {}
-    void connectivityChanged(sa_family_t) {}
-    void connectivityChanged() { }
+    void connectivityChanged(sa_family_t) {
+        restartListeners();
+    }
+    void connectivityChanged() {
+        getProxyInfos();
+        restartListeners();
+        loopSignal_();
+    }
 
 private:
     /**
@@ -263,7 +270,7 @@ private:
      * @return the JSON returned by the proxy
      */
     void getProxyInfos();
-    void onProxyInfos(const Json::Value& val);
+    void onProxyInfos(const Json::Value& val, sa_family_t family);
     SockAddr parsePublicAddress(const Json::Value& val);
 
     void opFailed();
@@ -271,6 +278,8 @@ private:
     size_t doListen(const InfoHash& key, ValueCallback, Value::Filter);
     bool doCancelListen(const InfoHash& key, size_t token);
 
+    void doPut(const InfoHash&, Sp<Value>, DoneCallback, time_point created, bool permanent);
+
     /**
      * Initialize statusIpvX_
      */
@@ -292,7 +301,8 @@ private:
     NodeStatus statusIpv6_ {NodeStatus::Disconnected};
     NodeStats stats4_ {};
     NodeStats stats6_ {};
-    SockAddr publicAddress_;
+    SockAddr publicAddressV4_;
+    SockAddr publicAddressV6_;
 
     InfoHash myid {};
 
@@ -304,10 +314,10 @@ private:
      */
     struct Listener;
     struct ProxySearch;
-    std::map<InfoHash, ProxySearch> listeners_;
 
-    size_t listener_token_ {0};
-    std::mutex lockListener_;
+    size_t listenerToken_ {0};
+    std::map<InfoHash, ProxySearch> searches_;
+    mutable std::mutex searchLock_;
 
     /**
      * Store current put and get requests.
@@ -351,12 +361,12 @@ private:
      * NOTE: empty by default to avoid to use services like FCM or APN.
      */
     std::string deviceKey_ {};
-    std::atomic_uint callbackId_ {0};
 
     const std::function<void()> loopSignal_;
 
 #if OPENDHT_PUSH_NOTIFICATIONS
-    void fillBodyToGetToken(std::shared_ptr<restbed::Request> request, unsigned callbackId);
+    void fillBodyToGetToken(std::shared_ptr<restbed::Request> request, unsigned token = 0);
+    void getPushRequest(Json::Value&) const;
 #endif // OPENDHT_PUSH_NOTIFICATIONS
 
     bool isDestroying_ {false};

include/opendht/dht_proxy_server.h

@@ -1,6 +1,7 @@
 /*
  *  Copyright (C) 2017-2018 Savoir-faire Linux Inc.
  *  Author: Sébastien Blin <sebastien.blin@savoirfairelinux.com>
+ *          Adrien Béraud <adrien.beraud@savoirfairelinux.com>
  *
  *  This program is free software; you can redistribute it and/or modify
  *  it under the terms of the GNU General Public License as published by
@@ -23,6 +24,9 @@
 #include "def.h"
 #include "sockaddr.h"
 #include "infohash.h"
+#include "scheduler.h"
+#include "value.h"
+#include "proxy.h"
 
 #include <thread>
 #include <memory>
@@ -59,6 +63,27 @@ public:
     DhtProxyServer& operator=(const DhtProxyServer& other) = delete;
     DhtProxyServer& operator=(DhtProxyServer&& other) = delete;
 
+    struct ServerStats {
+        /** Current number of listen operations */
+        size_t listenCount;
+        /** Current number of permanent put operations */
+        size_t putCount;
+        /** Current number of push tokens with at least one listen operation */
+        size_t pushListenersCount;
+        /** Average requests per second */
+        double requestRate;
+
+        std::string toString() const {
+            std::ostringstream ss;
+            ss << "Listens: " << listenCount << " Puts: " << putCount << " PushListeners: " << pushListenersCount << std::endl;
+            ss << "Requests: " << requestRate << " per second." << std::endl;
+            return ss.str();
+        }
+    };
+    ServerStats getStats() const;
+
+    std::shared_ptr<DhtRunner> getNode() const { return dht_; }
+
     /**
      * Stop the DhtProxyServer
      */
@@ -96,7 +121,7 @@ private:
      * On error: HTTP 503, body: {"err":"xxxx"}
      * @param session
      */
-    void listen(const std::shared_ptr<restbed::Session>& session) const;
+    void listen(const std::shared_ptr<restbed::Session>& session);
 
     /**
      * Put a value on the DHT
@@ -107,7 +132,9 @@ private:
      * HTTP 400, body: {"err":"xxxx"} if bad json or HTTP 502 if put fails
      * @param session
      */
-    void put(const std::shared_ptr<restbed::Session>& session) const;
+    void put(const std::shared_ptr<restbed::Session>& session);
+
+    void cancelPut(const InfoHash& key, Value::Id vid);
 
 #if OPENDHT_PROXY_SERVER_IDENTITY
     /**
@@ -154,43 +181,63 @@ private:
      */
     void handleOptionsMethod(const std::shared_ptr<restbed::Session>& session) const;
 
+    /**
+     * Remove finished listeners
+     * @param testSession if we remove the listener only if the session is closed
+     */
+    void removeClosedListeners(bool testSession = true);
+
 #if OPENDHT_PUSH_NOTIFICATIONS
     /**
      * Subscribe to push notifications for an iOS or Android device.
      * Method: SUBSCRIBE "/{InfoHash: .*}"
-     * Body: {"key": "device_key", (optional)"callback_id":y,
-     * (optional)"isAndroid":false (default true)}"
+     * Body: {"key": "device_key", (optional)"isAndroid":false (default true)}"
      * Return: {"token": x}" where x if a token to save
      * @note: the listen will timeout after six hours (and send a push notification).
      * so you need to refresh the operation each six hours.
-     * @note: callback_id is used to add the possibility to have multiple listen
-     * on same hash for same device and must be > 0
      * @param session
      */
-    void subscribe(const std::shared_ptr<restbed::Session>& session) const;
+    void subscribe(const std::shared_ptr<restbed::Session>& session);
     /**
      * Unsubscribe to push notifications for an iOS or Android device.
      * Method: UNSUBSCRIBE "/{InfoHash: .*}"
-     * Body: {"key": "device_key", "token": x, (optional)"callback_id":y"
-     * where x if the token to cancel
+     * Body: {"key": "device_key", "token": x} where x if the token to cancel
      * Return: nothing
-     * @note: callback id is used to add the possibility to have multiple listen
-     * on same hash for same device
      * @param session
      */
-    void unsubscribe(const std::shared_ptr<restbed::Session>& session) const;
+    void unsubscribe(const std::shared_ptr<restbed::Session>& session);
     /**
      * Send a push notification via a gorush push gateway
      * @param key of the device
      * @param json, the content to send
      */
     void sendPushNotification(const std::string& key, const Json::Value& json, bool isAndroid) const;
+
+    /**
+     * Remove a push listener between a client and a hash
+     * @param pushToken
+     * @param key
+     * @param clientId
+     */
+    void cancelPushListen(const std::string& pushToken, const InfoHash& key, const std::string& clientId);
+
+
 #endif //OPENDHT_PUSH_NOTIFICATIONS
 
+    using clock = std::chrono::steady_clock;
+    using time_point = clock::time_point;
+
     std::thread server_thread {};
     std::unique_ptr<restbed::Service> service_;
     std::shared_ptr<DhtRunner> dht_;
 
+    std::mutex schedulerLock_;
+    std::condition_variable schedulerCv_;
+    Scheduler scheduler_;
+    std::thread schedulerThread_;
+
+    Sp<Scheduler::Job> printStatsJob_;
+
     // Handle client quit for listen.
     // NOTE: can be simplified when we will supports restbed 5.0
     std::thread listenThread_;
@@ -199,37 +246,26 @@ private:
         InfoHash hash;
         std::future<size_t> token;
     };
-    mutable std::vector<SessionToHashToken> currentListeners_;
-    mutable std::mutex lockListener_;
+    std::vector<SessionToHashToken> currentListeners_;
+    std::mutex lockListener_;
     std::atomic_bool stopListeners {false};
 
-#if OPENDHT_PUSH_NOTIFICATIONS
-    struct PushListener {
-        std::string pushToken;
-        InfoHash hash;
-        unsigned token;
-        std::future<size_t> internalToken;
-        std::chrono::steady_clock::time_point deadline;
-        bool started {false};
-        unsigned callbackId {0};
-        std::string clientId {};
-        bool isAndroid {true};
-    };
-    mutable std::mutex lockPushListeners_;
-    mutable std::vector<PushListener> pushListeners_;
-    mutable unsigned tokenPushNotif_ {0};
-#endif //OPENDHT_PUSH_NOTIFICATIONS
+    struct PermanentPut;
+    struct SearchPuts;
+    std::map<InfoHash, SearchPuts> puts_;
+
+    mutable std::atomic<size_t> requestNum_ {0};
+    mutable std::atomic<time_point> lastStatsReset_ {time_point::min()};
+
     const std::string pushServer_;
 
-    /**
-     * Remove finished listeners
-     * @param testSession if we remove the listener only if the session is closed
-     */
-    void removeClosedListeners(bool testSession = true);
-    /**
-     * Launch or remove push listeners if needed
-     */
-    void handlePushListeners();
+#if OPENDHT_PUSH_NOTIFICATIONS
+    struct Listener;
+    struct PushListener;
+    std::mutex lockPushListeners_;
+    std::map<std::string, PushListener> pushListeners_;
+    proxy::ListenToken tokenPushNotif_ {0};
+#endif //OPENDHT_PUSH_NOTIFICATIONS
 };
 
 }

include/opendht/dhtrunner.h

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

include/opendht/proxy.h

@@ -0,0 +1,31 @@
+/*
+ *  Copyright (C) 2017-2018 Savoir-faire Linux Inc.
+ *  Author: Adrien Béraud <adrien.beraud@savoirfairelinux.com>
+ *
+ *  This program is free software; you can redistribute it and/or modify
+ *  it under the terms of the GNU General Public License as published by
+ *  the Free Software Foundation; either version 3 of the License, or
+ *  (at your option) any later version.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+#pragma once
+
+#include <chrono>
+
+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 char* const HTTP_PROTO {"http://"};
+using ListenToken = uint64_t;
+
+}
+}

include/opendht/scheduler.h

@@ -36,11 +36,10 @@ namespace dht {
  */
 class Scheduler {
 public:
-    Scheduler(const Logger& l) : DHT_LOG(l) {}
-
     struct Job {
         Job(std::function<void()>&& f) : do_(std::move(f)) {}
         std::function<void()> do_;
+        void cancel() { do_ = {}; }
     };
 
     /**
@@ -58,17 +57,19 @@ public:
         return job;
     }
 
+    void add(const Sp<Scheduler::Job>& job, time_point t) {
+        if (t != time_point::max())
+            timers.emplace(std::move(t), job);
+    }
+
     /**
      * Reschedules a job.
      *
-     * @param time  The time at which the job shall be rescheduled.
      * @param job  The job to edit.
-     *
-     * @return pointer to the newly scheduled job.
+     * @param t  The time at which the job shall be rescheduled.
      */
     void edit(Sp<Scheduler::Job>& job, time_point t) {
         if (not job) {
-            DHT_LOG.ERR("editing an empty job");
             return;
         }
         // std::function move doesn't garantee to leave the object empty.
@@ -118,7 +119,6 @@ public:
 private:
     time_point now {clock::now()};
     std::multimap<time_point, Sp<Job>> timers {}; /* the jobs ordered by time */
-    const Logger& DHT_LOG;
 };
 
 }

include/opendht/securedht.h

@@ -346,8 +346,6 @@ private:
     std::map<InfoHash, Sp<crypto::Certificate>> nodesCertificates_ {};
     std::map<InfoHash, Sp<const crypto::PublicKey>> nodesPubKeys_ {};
 
-    std::uniform_int_distribution<Value::Id> rand_id {};
-
     std::atomic_bool forward_all_ {false};
 };
 

include/opendht/utils.h

@@ -54,6 +54,12 @@ void erase_if(std::map<Key, Item>& map, const Condition& condition)
     }
 }
 
+/**
+ * Split "[host]:port" or "host:port" to pair<"host", "port">.
+ */
+OPENDHT_PUBLIC std::pair<std::string, std::string>
+splitPort(const std::string& s);
+
 class OPENDHT_PUBLIC DhtException : public std::runtime_error {
 public:
     DhtException(const std::string &str = "") :

include/opendht/value.h

@@ -996,6 +996,10 @@ unpackVector(const std::vector<std::shared_ptr<Value>>& vals) {
     return ret;
 }
 
+#if OPENDHT_PROXY_SERVER || OPENDHT_PROXY_CLIENT
+uint64_t unpackId(const Json::Value& json, const std::string& key);
+#endif
+
 }
 
 MSGPACK_ADD_ENUM(dht::Value::Field)

src/Makefile.am

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

src/crypto.cpp

@@ -411,6 +411,8 @@ PublicKey::operator=(PublicKey&& o) noexcept
 void
 PublicKey::pack(Blob& b) const
 {
+    if (not pk)
+        throw CryptoException(std::string("Could not export public key: null key"));
     std::vector<uint8_t> tmp(2048);
     size_t sz = tmp.size();
     int err = gnutls_pubkey_export(pk, GNUTLS_X509_FMT_DER, tmp.data(), &sz);
@@ -437,6 +439,8 @@ PublicKey::unpack(const uint8_t* data, size_t data_size)
 std::string
 PublicKey::toString() const
 {
+    if (not pk)
+        throw CryptoException(std::string("Could not print public key: null key"));
     std::string ret;
     size_t sz = ret.size();
     int err = gnutls_pubkey_export(pk, GNUTLS_X509_FMT_PEM, (void*)ret.data(), &sz);

src/dht.cpp

@@ -180,7 +180,7 @@ Dht::expireBuckets(RoutingTable& list)
 {
     for (auto& b : list) {
         bool changed = false;
-        b.nodes.remove_if([this,&changed](const Sp<Node>& n) {
+        b.nodes.remove_if([&changed](const Sp<Node>& n) {
             if (n->isExpired()) {
                 changed = true;
                 return true;
@@ -387,7 +387,7 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr) {
         if (not n.isSynced(now))
             continue;
         if (std::find_if(sr->announce.cbegin(), sr->announce.cend(),
-            [this,&now,&n](const Announce& a) {
+            [&now,&n](const Announce& a) {
                 return n.getAnnounceTime(a.value->id) <= now;
             }) == sr->announce.cend())
             continue;
@@ -482,6 +482,72 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr) {
     }
 }
 
+void
+Dht::searchSynchedNodeListen(const Sp<Search>& sr, SearchNode& n)
+{
+    std::weak_ptr<Search> ws = sr;
+    for (const auto& l : sr->listeners) {
+        const auto& query = l.second.query;
+        auto list_token = l.first;
+        if (n.getListenTime(query) > scheduler.time())
+            continue;
+        DHT_LOG.w(sr->id, n.node->id, "[search %s] [node %s] sending 'listen'",
+                sr->id.toString().c_str(), n.node->toString().c_str());
+
+        auto r = n.listenStatus.find(query);
+        if (r == n.listenStatus.end()) {
+            r = n.listenStatus.emplace(query, SearchNode::CachedListenStatus{
+                [ws,list_token](const std::vector<Sp<Value>>& values, bool expired){
+                    if (auto sr = ws.lock()) {
+                        auto l = sr->listeners.find(list_token);
+                        if (l != sr->listeners.end()) {
+                            l->second.get_cb(l->second.filter.filter(values), expired);
+                        }
+                    }
+                }
+            }).first;
+            auto node = n.node;
+            r->second.cacheExpirationJob = scheduler.add(time_point::max(), [this,ws,query,node]{
+                if (auto sr = ws.lock()) {
+                    if (auto sn = sr->getNode(node)) {
+                        sn->expireValues(query, scheduler);
+                    }
+                }
+            });
+        }
+        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,
+            [this,ws,query](const net::Request& req, net::RequestAnswer&& answer) mutable
+            { /* on done */
+                if (auto sr = ws.lock()) {
+                    scheduler.edit(sr->nextSearchStep, scheduler.time());
+                    if (auto sn = sr->getNode(req.node))
+                        scheduler.add(sn->getListenTime(query), std::bind(&Dht::searchStep, this, sr));
+                    onListenDone(req.node, answer, sr);
+                }
+            },
+            [this,ws,query](const net::Request& req, bool over) mutable
+            { /* on request expired */
+                if (auto sr = ws.lock()) {
+                    scheduler.edit(sr->nextSearchStep, scheduler.time());
+                    if (over)
+                        if (auto sn = sr->getNode(req.node))
+                            sn->listenStatus.erase(query);
+                }
+            },
+            [this,ws,query](const Sp<Node>& node, net::RequestAnswer&& answer) mutable
+            { /* on new values */
+                if (auto sr = ws.lock()) {
+                    scheduler.edit(sr->nextSearchStep, scheduler.time());
+                    if (auto sn = sr->getNode(node)) {
+                        sn->onValues(query, std::move(answer), types, scheduler);
+                    }
+                }
+            }
+        );
+    }
+}
+
 /* When a search is in progress, we periodically call search_step to send
    further requests. */
 void
@@ -538,59 +604,7 @@ Dht::searchStep(Sp<Search> sr)
             for (auto& n : sr->nodes) {
                 if (not n.isSynced(now))
                     continue;
-                std::weak_ptr<Search> ws = sr;
-                for (const auto& l : sr->listeners) {
-                    const auto& query = l.second.query;
-                    auto list_token = l.first;
-                    if (n.getListenTime(query) > now)
-                        continue;
-                    DHT_LOG.w(sr->id, n.node->id, "[search %s] [node %s] sending 'listen'",
-                            sr->id.toString().c_str(), n.node->toString().c_str());
-
-                    auto r = n.listenStatus.find(query);
-                    if (r == n.listenStatus.end()) {
-                        r = n.listenStatus.emplace(query, SearchNode::CachedListenStatus{
-                            [this,ws,list_token](const std::vector<Sp<Value>>& values, bool expired){
-                                if (auto sr = ws.lock()) {
-                                    auto l = sr->listeners.find(list_token);
-                                    if (l != sr->listeners.end()) {
-                                        l->second.get_cb(l->second.filter.filter(values), expired);
-                                    }
-                                }
-                            }
-                        }).first;
-                    }
-                    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,
-                        [this,ws,query](const net::Request& req, net::RequestAnswer&& answer) mutable
-                        { /* on done */
-                            if (auto sr = ws.lock()) {
-                                scheduler.edit(sr->nextSearchStep, scheduler.time());
-                                if (auto sn = sr->getNode(req.node))
-                                    scheduler.add(sn->getListenTime(query), std::bind(&Dht::searchStep, this, sr));
-                                onListenDone(req.node, answer, sr);
-                            }
-                        },
-                        [this,ws,query](const net::Request& req, bool over) mutable
-                        { /* on expired */
-                            if (auto sr = ws.lock()) {
-                                scheduler.edit(sr->nextSearchStep, scheduler.time());
-                                if (over)
-                                    if (auto sn = sr->getNode(req.node))
-                                        sn->listenStatus.erase(query);
-                            }
-                        },
-                        [this,ws,query](const Sp<Node>& node, net::RequestAnswer&& answer) mutable
-                        { /* on new values */
-                            if (auto sr = ws.lock()) {
-                                scheduler.edit(sr->nextSearchStep, scheduler.time());
-                                if (auto sn = sr->getNode(node))
-                                    sn->onValues(query, std::move(answer), types, scheduler.time());
-                                    //onGetValuesDone(node, answer, sr, query);
-                            }
-                        }
-                    );
-                }
+                searchSynchedNodeListen(sr, n);
                 if (not n.candidate and ++i == LISTEN_NODES)
                     break;
             }
@@ -860,10 +874,10 @@ Dht::cancelListen(const InfoHash& id, size_t token)
     if (st != store.end() && tokenlocal)
         st->second.local_listeners.erase(tokenlocal);
 
-    auto searches_cancel_listen = [&id](std::map<InfoHash, Sp<Search>>& srs, size_t token) {
+    auto searches_cancel_listen = [this,&id](std::map<InfoHash, Sp<Search>>& srs, size_t token) {
         auto srp = srs.find(id);
         if (srp != srs.end() and token)
-            srp->second->cancelListen(token);
+            srp->second->cancelListen(token, scheduler);
     };
     searches_cancel_listen(searches4, std::get<1>(it->second));
     searches_cancel_listen(searches6, std::get<2>(it->second));
@@ -1125,25 +1139,27 @@ Dht::cancelPut(const InfoHash& id, const Value::Id& vid)
 // Storage
 
 void
-Dht::storageChanged(const InfoHash& id, Storage& st, ValueStorage& v)
+Dht::storageChanged(const InfoHash& id, Storage& st, ValueStorage& v, bool newValue)
 {
-    if (not st.local_listeners.empty()) {
-        DHT_LOG.d(id, "[store %s] %lu local listeners", id.toString().c_str(), st.local_listeners.size());
-        std::vector<std::pair<ValueCallback, std::vector<Sp<Value>>>> cbs;
-        for (const auto& l : st.local_listeners) {
-            std::vector<Sp<Value>> vals;
-            if (not l.second.filter or l.second.filter(*v.data))
-                vals.push_back(v.data);
-            if (not vals.empty()) {
-                DHT_LOG.d(id, "[store %s] sending update local listener with token %lu",
-                        id.toString().c_str(),
-                        l.first);
-                cbs.emplace_back(l.second.get_cb, std::move(vals));
+    if (newValue) {
+        if (not st.local_listeners.empty()) {
+            DHT_LOG.d(id, "[store %s] %lu local listeners", id.toString().c_str(), st.local_listeners.size());
+            std::vector<std::pair<ValueCallback, std::vector<Sp<Value>>>> cbs;
+            for (const auto& l : st.local_listeners) {
+                std::vector<Sp<Value>> vals;
+                if (not l.second.filter or l.second.filter(*v.data))
+                    vals.push_back(v.data);
+                if (not vals.empty()) {
+                    DHT_LOG.d(id, "[store %s] sending update local listener with token %lu",
+                            id.toString().c_str(),
+                            l.first);
+                    cbs.emplace_back(l.second.get_cb, std::move(vals));
+                }
             }
+            // listeners are copied: they may be deleted by the callback
+            for (auto& cb : cbs)
+                cb.first(cb.second, false);
         }
-        // listeners are copied: they may be deleted by the callback
-        for (auto& cb : cbs)
-            cb.first(cb.second, false);
     }
 
     if (not st.listeners.empty()) {
@@ -1197,7 +1213,7 @@ Dht::storageStore(const InfoHash& id, const Sp<Value>& value, time_point created
         if (total_store_size > max_store_size) {
             expireStore();
         }
-        storageChanged(id, st->second, *vs);
+        storageChanged(id, st->second, *vs, store.second.values_diff > 0);
     }
 
     return std::get<0>(store);
@@ -1258,6 +1274,9 @@ Dht::expireStore(decltype(store)::iterator i)
                 }
             }
         }
+        for (const auto& local_listeners : st.local_listeners) {
+            local_listeners.second.get_cb(stats.second, true);
+        }
     }
 }
 
@@ -1665,13 +1684,12 @@ Dht::~Dht()
         s.second->clear();
 }
 
-Dht::Dht() : store(), scheduler(DHT_LOG), network_engine(DHT_LOG, scheduler) {}
+Dht::Dht() : store(), network_engine(DHT_LOG, scheduler) {}
 
 Dht::Dht(const int& s, const int& s6, Config config)
     : myid(config.node_id ? config.node_id : InfoHash::getRandom()),
     is_bootstrap(config.is_bootstrap),
     maintain_storage(config.maintain_storage), store(), store_quota(),
-    scheduler(DHT_LOG),
     network_engine(myid, config.network, s, s6, DHT_LOG, scheduler,
             std::bind(&Dht::onError, this, _1, _2),
             std::bind(&Dht::onNewNode, this, _1, _2),

src/dht_proxy_client.cpp

@@ -21,42 +21,58 @@
 
 #include "dht_proxy_client.h"
 
-#include <chrono>
-#include <json/json.h>
-#include <restbed>
-#include <vector>
-
 #include "dhtrunner.h"
 #include "op_cache.h"
+#include "utils.h"
 
-constexpr const char* const HTTP_PROTO {"http://"};
+#include <restbed>
+#include <json/json.h>
+
+#include <chrono>
+#include <vector>
 
 namespace dht {
 
 struct DhtProxyClient::Listener
 {
+    struct State {
+        std::atomic_bool ok {true};
+        std::atomic_bool cancel {false};
+    };
     ValueCache cache;
+    Sp<Scheduler::Job> cacheExpirationJob {};
     ValueCallback cb;
     Value::Filter filter;
     Sp<restbed::Request> req;
     std::thread thread;
     unsigned callbackId;
-    Sp<bool> isCanceledViaClose;
-    Sp<unsigned> pushNotifToken; // NOTE: unused if not using push notifications
-    Listener(ValueCache&& c, const Sp<restbed::Request>& r, Value::Filter&& f, unsigned cid)
-        : cache(std::move(c)), filter(std::move(f)), req(r), callbackId(cid), isCanceledViaClose(std::make_shared<bool>(false))
+    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)
+        : cache(std::move(c)),
+          cacheExpirationJob(std::move(j)),
+          filter(std::move(f)),
+          req(r)
     {}
 };
 
+struct PermanentPut {
+    Sp<Value> value;
+    Sp<Scheduler::Job> refreshJob;
+};
+
 struct DhtProxyClient::ProxySearch {
     SearchCache ops {};
+    Sp<Scheduler::Job> opExpirationJob;
     std::map<size_t, Listener> listeners {};
+    std::map<Value::Id, PermanentPut> puts {};
 };
 
-DhtProxyClient::DhtProxyClient() : scheduler(DHT_LOG) {}
+DhtProxyClient::DhtProxyClient() {}
 
 DhtProxyClient::DhtProxyClient(std::function<void()> signal, const std::string& serverHost, const std::string& pushClientId)
-: serverHost_(serverHost), pushClientId_(pushClientId), scheduler(DHT_LOG), loopSignal_(signal)
+: serverHost_(serverHost), pushClientId_(pushClientId), loopSignal_(signal)
 {
     if (!serverHost_.empty())
         startProxy();
@@ -73,8 +89,9 @@ void
 DhtProxyClient::startProxy()
 {
     if (serverHost_.empty()) return;
-    DHT_LOG.WARN("Staring proxy client to %s", serverHost_.c_str());
+    DHT_LOG.w("Staring proxy client to %s", serverHost_.c_str());
     nextProxyConfirmation = scheduler.add(scheduler.time(), std::bind(&DhtProxyClient::confirmProxy, this));
+    loopSignal_();
 }
 
 DhtProxyClient::~DhtProxyClient()
@@ -82,20 +99,24 @@ DhtProxyClient::~DhtProxyClient()
     isDestroying_ = true;
     cancelAllOperations();
     cancelAllListeners();
+    if (statusThread_.joinable())
+        statusThread_.join();
 }
 
 std::vector<Sp<Value>>
 DhtProxyClient::getLocal(const InfoHash& k, Value::Filter filter) const {
-    auto s = listeners_.find(k);
-    if (s == listeners_.end())
+    std::lock_guard<std::mutex> lock(searchLock_);
+    auto s = searches_.find(k);
+    if (s == searches_.end())
         return {};
     return s->second.ops.get(filter);
 }
 
 Sp<Value>
 DhtProxyClient::getLocalById(const InfoHash& k, Value::Id id) const {
-    auto s = listeners_.find(k);
-    if (s == listeners_.end())
+    std::lock_guard<std::mutex> lock(searchLock_);
+    auto s = searches_.find(k);
+    if (s == searches_.end())
         return {};
     return s->second.ops.get(id);
 }
@@ -120,15 +141,22 @@ DhtProxyClient::cancelAllOperations()
 void
 DhtProxyClient::cancelAllListeners()
 {
-    std::lock_guard<std::mutex> lock(lockListener_);
-    for (auto& s: listeners_) {
-        for (auto& l : s.second.listeners)
-            if (l.second.thread.joinable()) {
+    std::lock_guard<std::mutex> lock(searchLock_);
+    DHT_LOG.w("Cancelling all listeners for %zu searches", searches_.size());
+    for (auto& s: searches_) {
+        s.second.ops.cancelAll([&](size_t token){
+            auto l = s.second.listeners.find(token);
+            if (l == s.second.listeners.end())
+                return;
+            if (l->second.thread.joinable()) {
                 // Close connection to stop listener?
-                if (l.second.req)
-                    restbed::Http::close(l.second.req);
-                l.second.thread.join();
+                l->second.state->cancel = true;
+                if (l->second.req)
+                    restbed::Http::close(l->second.req);
+                l->second.thread.join();
             }
+            s.second.listeners.erase(token);
+        });
     }
 }
 
@@ -205,7 +233,8 @@ DhtProxyClient::periodic(const uint8_t*, size_t, const SockAddr&)
 void
 DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Value::Filter&& f, Where&& w)
 {
-    restbed::Uri uri(HTTP_PROTO + serverHost_ + "/" + key.toString());
+    DHT_LOG.d(key, "[search %s]: get", key.to_c_str());
+    restbed::Uri uri(proxy::HTTP_PROTO + serverHost_ + "/" + key.toString());
     auto req = std::make_shared<restbed::Request>(uri);
     Value::Filter filter = w.empty() ? f : f.chain(w.getFilter());
 
@@ -239,7 +268,7 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Va
                         auto reader = std::unique_ptr<Json::CharReader>(rbuilder.newCharReader());
                         if (reader->parse(char_data, char_data + body.size(), &json, &err)) {
                             auto value = std::make_shared<Value>(json);
-                            if ((not filter or filter(*value)) && cb) {
+                            if ((not filter or filter(*value)) and cb) {
                                 std::lock_guard<std::mutex> lock(lockCallbacks);
                                 callbacks_.emplace_back([cb, value, finished]() {
                                     if (not *finished and not cb({value}))
@@ -276,14 +305,65 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Va
 }
 
 void
-DhtProxyClient::put(const InfoHash& key, Sp<Value> val, DoneCallback cb, time_point, bool permanent)
+DhtProxyClient::put(const InfoHash& key, Sp<Value> val, DoneCallback cb, time_point created, bool permanent)
 {
-    restbed::Uri uri(HTTP_PROTO + serverHost_ + "/" + key.toString());
+    DHT_LOG.d(key, "[search %s]: put", key.to_c_str());
+    scheduler.syncTime();
+    if (not val) {
+        if (cb) cb(false, {});
+        return;
+    }
+    if (val->id == Value::INVALID_ID) {
+        crypto::random_device rdev;
+        std::uniform_int_distribution<Value::Id> rand_id {};
+        val->id = rand_id(rdev);
+    }
+    if (permanent) {
+        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;
+        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, val, std::move(cb), created, permanent);
+}
+
+void
+DhtProxyClient::doPut(const InfoHash& key, Sp<Value> val, DoneCallback cb, time_point created, bool permanent)
+{
+    DHT_LOG.d(key, "[search %s] performing put of %s", key.to_c_str(), val->toString().c_str());
+    restbed::Uri uri(proxy::HTTP_PROTO + serverHost_ + "/" + key.toString());
     auto req = std::make_shared<restbed::Request>(uri);
     req->set_method("POST");
+
     auto json = val->toJson();
-    if (permanent)
-        json["permanent"] = true;
+    if (permanent) {
+        if (deviceKey_.empty()) {
+            json["permanent"] = true;
+        } else {
+#if OPENDHT_PUSH_NOTIFICATIONS
+            Json::Value refresh;
+            getPushRequest(refresh);
+            json["permanent"] = refresh;
+#else
+            json["permanent"] = true;
+#endif
+        }
+    }
     Json::StreamWriterBuilder wbuilder;
     wbuilder["commentStyle"] = "None";
     wbuilder["indentation"] = "";
@@ -298,7 +378,7 @@ DhtProxyClient::put(const InfoHash& key, Sp<Value> val, DoneCallback cb, time_po
     o.thread = std::thread([=](){
         auto ok = std::make_shared<std::atomic_bool>(true);
         restbed::Http::async(req,
-            [this, ok](const std::shared_ptr<restbed::Request>& /*req*/,
+            [ok](const std::shared_ptr<restbed::Request>& /*req*/,
                         const std::shared_ptr<restbed::Response>& reply) {
             auto code = reply->get_status_code();
 
@@ -342,6 +422,49 @@ DhtProxyClient::put(const InfoHash& key, Sp<Value> val, DoneCallback cb, time_po
     }
 }
 
+/**
+ * Get data currently being put at the given hash.
+ */
+std::vector<Sp<Value>>
+DhtProxyClient::getPut(const InfoHash& key) {
+    std::vector<Sp<Value>> ret;
+    auto search = searches_.find(key);
+    if (search != searches_.end()) {
+        ret.reserve(search->second.puts.size());
+        for (const auto& put : search->second.puts)
+            ret.emplace_back(put.second.value);
+    }
+    return ret;
+}
+
+/**
+ * Get data currently being put at the given hash with the given id.
+ */
+Sp<Value>
+DhtProxyClient::getPut(const InfoHash& key, const Value::Id& id) {
+    auto search = searches_.find(key);
+    if (search == searches_.end())
+        return {};
+    auto val = search->second.puts.find(id);
+    if (val == search->second.puts.end())
+        return {};
+    return val->second.value;
+}
+
+/**
+ * Stop any put/announce operation at the given location,
+ * for the value with the given id.
+ */
+bool
+DhtProxyClient::cancelPut(const InfoHash& key, const Value::Id& id)
+{
+    auto search = searches_.find(key);
+    if (search == searches_.end())
+        return false;
+    DHT_LOG.d(key, "[search %s] cancel put", key.to_c_str());
+    return search->second.puts.erase(id) > 0;
+}
+
 NodeStats
 DhtProxyClient::getNodesStats(sa_family_t af) const
 {
@@ -351,7 +474,7 @@ DhtProxyClient::getNodesStats(sa_family_t af) const
 void
 DhtProxyClient::getProxyInfos()
 {
-    DHT_LOG.DEBUG("Requesting proxy server node information");
+    DHT_LOG.d("Requesting proxy server node information");
 
     if (ongoingStatusUpdate_.test_and_set())
         return;
@@ -364,70 +487,104 @@ DhtProxyClient::getProxyInfos()
             statusIpv6_ = NodeStatus::Connecting;
     }
 
-    restbed::Uri uri(HTTP_PROTO + serverHost_ + "/");
-    auto req = std::make_shared<restbed::Request>(uri);
+    // A node can have a Ipv4 and a Ipv6. So, we need to retrieve all public ips
+    auto serverHost = serverHost_;
 
     // Try to contact the proxy and set the status to connected when done.
     // will change the connectivity status
-    statusThread_ = std::thread([this, req]{
-        restbed::Http::async(req,
-            [this](const std::shared_ptr<restbed::Request>&,
-                           const std::shared_ptr<restbed::Response>& reply) {
-            auto code = reply->get_status_code();
-            Json::Value proxyInfos;
-            if (code == 200) {
-                restbed::Http::fetch("\n", reply);
-                std::string body;
-                reply->get_body(body);
-
-                std::string err;
-                Json::CharReaderBuilder rbuilder;
-                auto reader = std::unique_ptr<Json::CharReader>(rbuilder.newCharReader());
-                try {
-                    reader->parse(body.data(), body.data() + body.size(), &proxyInfos, &err);
-                } catch (...) {
+    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>();
+        try {
+            auto hostAndService = splitPort(serverHost);
+            auto resolved_proxies = SockAddr::resolve(hostAndService.first, hostAndService.second);
+            std::vector<std::future<Sp<restbed::Response>>> reqs;
+            reqs.reserve(resolved_proxies.size());
+            for (const auto& resolved_proxy: resolved_proxies) {
+                auto server = resolved_proxy.toString();
+                if (resolved_proxy.getFamily() == AF_INET6) {
+                    // HACK restbed seems to not correctly handle directly http://[ipv6]
+                    // See https://github.com/Corvusoft/restbed/issues/290.
+                    server = serverHost;
                 }
+                restbed::Uri uri(proxy::HTTP_PROTO + server + "/");
+                auto req = std::make_shared<restbed::Request>(uri);
+                reqs.emplace_back(restbed::Http::async(req,
+                    [this, resolved_proxy, successCount](
+                                const std::shared_ptr<restbed::Request>&,
+                                const std::shared_ptr<restbed::Response>& reply)
+                {
+                    auto code = reply->get_status_code();
+                    Json::Value proxyInfos;
+                    if (code == 200) {
+                        restbed::Http::fetch("\n", reply);
+                        std::string body;
+                        reply->get_body(body);
+
+                        std::string err;
+                        Json::CharReaderBuilder rbuilder;
+                        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());
+                        }
+                        auto family = resolved_proxy.getFamily();
+                        if      (family == AF_INET)  successCount->ipv4++;
+                        else if (family == AF_INET6) successCount->ipv6++;
+                        onProxyInfos(proxyInfos, family);
+                    }
+                }));
             }
-            onProxyInfos(proxyInfos);
-            ongoingStatusUpdate_.clear();
-        });
+            for (auto& r : reqs)
+                r.get();
+            reqs.clear();
+        } 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();
     });
-    statusThread_.detach();
 }
 
 void
-DhtProxyClient::onProxyInfos(const Json::Value& proxyInfos)
+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")) {
+        status = NodeStatus::Disconnected;
+    } else {
+        try {
+            myid = InfoHash(proxyInfos["node_id"].asString());
+            stats4_ = NodeStats(proxyInfos["ipv4"]);
+            stats6_ = NodeStats(proxyInfos["ipv6"]);
+            if (stats4_.good_nodes + stats6_.good_nodes)
+                status = NodeStatus::Connected;
+            else if (stats4_.dubious_nodes + stats6_.dubious_nodes)
+                status = NodeStatus::Connecting;
+            else
+                status = NodeStatus::Disconnected;
 
-    try {
-        myid = InfoHash(proxyInfos["node_id"].asString());
-
-        stats4_ = NodeStats(proxyInfos["ipv4"]);
-        if (stats4_.good_nodes)
-            statusIpv4_ = NodeStatus::Connected;
-        else if (stats4_.dubious_nodes)
-            statusIpv4_ = NodeStatus::Connecting;
-        else
-            statusIpv4_ = NodeStatus::Disconnected;
-
-        stats6_ = NodeStats(proxyInfos["ipv6"]);
-        if (stats6_.good_nodes)
-            statusIpv6_ = NodeStatus::Connected;
-        else if (stats6_.dubious_nodes)
-            statusIpv6_ = NodeStatus::Connecting;
-        else
-            statusIpv6_ = NodeStatus::Disconnected;
-
-        publicAddress_ = parsePublicAddress(proxyInfos["public_ip"]);
-    } catch (...) {}
+            auto publicIp = parsePublicAddress(proxyInfos["public_ip"]);
+            auto publicFamily = publicIp.getFamily();
+            if (publicFamily == AF_INET)
+                publicAddressV4_ = publicIp;
+            else if (publicFamily == AF_INET6)
+                publicAddressV6_ = publicIp;
+        } catch (const std::exception& e) {
+            DHT_LOG.w("Error processing proxy infos: %s", e.what());
+        }
+    }
 
     auto newStatus = std::max(statusIpv4_, statusIpv6_);
-
-    if (newStatus == NodeStatus::Connecting || newStatus == NodeStatus::Connected) {
-        if (oldStatus == NodeStatus::Disconnected) {
+    if (newStatus == NodeStatus::Connected) {
+        if (oldStatus == NodeStatus::Disconnected || oldStatus == NodeStatus::Connecting) {
             restartListeners();
         }
         scheduler.edit(nextProxyConfirmation, scheduler.time() + std::chrono::minutes(15));
@@ -442,10 +599,8 @@ SockAddr
 DhtProxyClient::parsePublicAddress(const Json::Value& val)
 {
     auto public_ip = val.asString();
-    auto endIp = public_ip.find_last_of(':');
-    std::string service = public_ip.substr(endIp + 1);
-    std::string address = public_ip.substr(0, endIp - 1);
-    auto sa = SockAddr::resolve(address, service);
+    auto hostAndService = splitPort(public_ip);
+    auto sa = SockAddr::resolve(hostAndService.first);
     if (sa.empty()) return {};
     return sa.front().getMappedIPv4();
 }
@@ -454,15 +609,18 @@ std::vector<SockAddr>
 DhtProxyClient::getPublicAddress(sa_family_t family)
 {
     std::lock_guard<std::mutex> l(lockCurrentProxyInfos_);
-    if (not publicAddress_) return {};
-    return publicAddress_.getFamily() == family ? std::vector<SockAddr>{publicAddress_} : std::vector<SockAddr>{};
+    std::vector<SockAddr> result;
+    if (publicAddressV6_ && family != AF_INET) result.emplace_back(publicAddressV6_);
+    if (publicAddressV4_ && family != AF_INET6) result.emplace_back(publicAddressV4_);
+    return result;
 }
 
 size_t
 DhtProxyClient::listen(const InfoHash& key, ValueCallback cb, Value::Filter filter, Where where) {
-    auto it = listeners_.find(key);
-    if (it == listeners_.end()) {
-        it = listeners_.emplace(key, ProxySearch{}).first;
+    DHT_LOG.d(key, "[search %s]: listen", key.to_c_str());
+    auto it = searches_.find(key);
+    if (it == searches_.end()) {
+        it = searches_.emplace(key, ProxySearch{}).first;
     }
     auto query = std::make_shared<Query>(Select{}, where);
     auto token = it->second.ops.listen(cb, query, filter, [&](Sp<Query> q, ValueCallback vcb){
@@ -473,36 +631,82 @@ DhtProxyClient::listen(const InfoHash& key, ValueCallback cb, Value::Filter filt
 
 bool
 DhtProxyClient::cancelListen(const InfoHash& key, size_t gtoken) {
-    auto it = listeners_.find(key);
-    if (it == listeners_.end())
+    scheduler.syncTime();
+    DHT_LOG.d(key, "[search %s]: cancelListen %zu", key.to_c_str(), gtoken);
+    auto it = searches_.find(key);
+    if (it == searches_.end())
         return false;
-    return it->second.ops.cancelListen(gtoken, [&](size_t ltoken){
-        doCancelListen(key, ltoken);
-    });
+    auto& ops = it->second.ops;
+    bool canceled = ops.cancelListen(gtoken, scheduler.time());
+    if (not it->second.opExpirationJob) {
+        it->second.opExpirationJob = scheduler.add(time_point::max(), [this,key](){
+            auto it = searches_.find(key);
+            if (it != searches_.end()) {
+                auto next = it->second.ops.expire(scheduler.time(), [this,key](size_t ltoken){
+                    doCancelListen(key, ltoken);
+                });
+                if (next != time_point::max()) {
+                    scheduler.edit(it->second.opExpirationJob, next);
+                }
+            }
+        });
+    }
+    scheduler.edit(it->second.opExpirationJob, ops.getExpiration());
+    loopSignal_();
+    return canceled;
 }
 
 size_t
 DhtProxyClient::doListen(const InfoHash& key, ValueCallback cb, Value::Filter filter/*, Where where*/)
 {
-    restbed::Uri uri(HTTP_PROTO + serverHost_ + "/" + key.toString());
+    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(lockListener_);
-    auto search = listeners_.find(key);
-    if (search == listeners_.end()) {
-        std::cerr << "doListen: search not found" << std::endl;
+    std::lock_guard<std::mutex> lock(searchLock_);
+    auto search = searches_.find(key);
+    if (search == searches_.end()) {
+        DHT_LOG.e(key, "[search %s] listen: search not found", key.to_c_str());
         return 0;
     }
+    DHT_LOG.d(key, "[search %s] sending %s", key.to_c_str(), deviceKey_.empty() ? "listen" : "subscribe");
 
-    auto token = ++listener_token_;
-    auto callbackId = ++callbackId_;
-    auto l = search->second.listeners.emplace(token, Listener{
-        ValueCache(cb), req, std::move(filter), callbackId
-    }).first;
+    auto token = ++listenerToken_;
+    auto l = search->second.listeners.find(token);
+    if (l == search->second.listeners.end()) {
+        auto f = filter;
+        l = search->second.listeners.emplace(token, Listener {
+            ValueCache(cb), scheduler.add(time_point::max(), [this, key, token]{
+                std::lock_guard<std::mutex> lock(searchLock_);
+                auto s = searches_.find(key);
+                if (s == searches_.end()) {
+                    return;
+                }
+                auto l = s->second.listeners.find(token);
+                if (l == s->second.listeners.end()) {
+                    return;
+                }
+                auto next = l->second.cache.expireValues(scheduler.time());
+                scheduler.edit(l->second.cacheExpirationJob, next);
+            }), req, std::move(f)
+        }).first;
+    }
 
-    ValueCache& cache = l->second.cache;
-    l->second.cb = [this,&cache](const std::vector<Sp<Value>>& values, bool expired) {
+    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)
+            return false;
+        std::lock_guard<std::mutex> lock(searchLock_);
+        auto s = searches_.find(key);
+        if (s == searches_.end()) {
+            return false;
+        }
+        auto l = s->second.listeners.find(token);
+        if (l == s->second.listeners.end()) {
+            return false;
+        }
         const std::vector<Sp<Value>> new_values_empty;
         std::vector<Value::Id> expired_ids;
         if (expired) {
@@ -510,13 +714,15 @@ DhtProxyClient::doListen(const InfoHash& key, ValueCallback cb, Value::Filter fi
             for (const auto& v : values)
                 expired_ids.emplace_back(v->id);
         }
-        cache.onValues(expired ? new_values_empty : values, std::vector<Value::Id>{}, expired_ids, types, scheduler.time());
+        auto next = l->second.cache.onValues(expired ? new_values_empty : values, std::vector<Value::Id>{}, expired_ids, types, scheduler.time());
+        scheduler.edit(l->second.cacheExpirationJob, next);
+        loopSignal_();
         return true;
     };
-    std::weak_ptr<bool> isCanceledViaClose(l->second.isCanceledViaClose);
-    auto pushNotifToken = std::make_shared<unsigned>(0);
+    auto pushNotifToken = std::make_shared<proxy::ListenToken>(0);
+    auto vcb = l->second.cb;
     l->second.pushNotifToken = pushNotifToken;
-    l->second.thread = std::thread([=]()
+    l->second.thread = std::thread([this,req,filter,vcb,pushNotifToken,state]()
         {
             auto settings = std::make_shared<restbed::Settings>();
             if (deviceKey_.empty()) {
@@ -525,17 +731,12 @@ DhtProxyClient::doListen(const InfoHash& key, ValueCallback cb, Value::Filter fi
             }
 #if OPENDHT_PUSH_NOTIFICATIONS
             else
-                fillBodyToGetToken(req, callbackId);
+                fillBodyToGetToken(req);
 #endif
 
-            struct State {
-                std::atomic_bool ok {true};
-                std::atomic_bool cancel {false};
-            };
-            auto state = std::make_shared<State>();
             restbed::Http::async(req,
-                [this, filter, cb, pushNotifToken, state](const std::shared_ptr<restbed::Request>& req,
-                                                               const std::shared_ptr<restbed::Response>& reply) {
+                [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 {
@@ -552,8 +753,7 @@ DhtProxyClient::doListen(const InfoHash& key, ValueCallback cb, Value::Filter fi
 
                             auto* char_data = reinterpret_cast<const char*>(&body[0]);
                             if (reader->parse(char_data, char_data + body.size(), &json, &err)) {
-                                if (!json.isMember("token")) return;
-                                *pushNotifToken = json["token"].asLargestUInt();
+                                *pushNotifToken = unpackId(json, "token");
                             } else {
                                 state->ok = false;
                             }
@@ -569,10 +769,10 @@ DhtProxyClient::doListen(const InfoHash& key, ValueCallback cb, Value::Filter fi
                                 auto* char_data = reinterpret_cast<const char*>(&body[0]);
                                 if (reader->parse(char_data, char_data + body.size(), &json, &err)) {
                                     auto value = std::make_shared<Value>(json);
-                                    if (not filter or filter(*value))  {
+                                    if ((not filter or filter(*value)) and vcb)  {
                                         std::lock_guard<std::mutex> lock(lockCallbacks);
-                                        callbacks_.emplace_back([cb, value, state]() {
-                                            if (not state->cancel and not cb({value}, false))
+                                        callbacks_.emplace_back([vcb, value, state]() {
+                                            if (not state->cancel and not vcb({value}, false))
                                                 state->cancel = true;
                                         });
                                         loopSignal_();
@@ -583,14 +783,14 @@ DhtProxyClient::doListen(const InfoHash& key, ValueCallback cb, Value::Filter fi
                             }
                         }
                     } catch (std::runtime_error&) {
-                        state->ok = false;
+                        // NOTE: Http::close() can occurs here. Ignore this.
                     }
                 } else {
                     state->ok = false;
                 }
             }, settings).get();
-            auto isCanceledNormally = isCanceledViaClose.lock();
-            if (not state->ok and isCanceledNormally and not *isCanceledNormally) {
+            auto& s = *state;
+            if (not s.ok and not s.cancel) {
                 opFailed();
             }
         }
@@ -601,32 +801,34 @@ DhtProxyClient::doListen(const InfoHash& key, ValueCallback cb, Value::Filter fi
 bool
 DhtProxyClient::doCancelListen(const InfoHash& key, size_t ltoken)
 {
-    std::lock_guard<std::mutex> lock(lockListener_);
+    std::lock_guard<std::mutex> lock(searchLock_);
 
-    auto search = listeners_.find(key);
-    if (search == listeners_.end())
+    auto search = searches_.find(key);
+    if (search == searches_.end())
         return false;
 
     auto it = search->second.listeners.find(ltoken);
     if (it == search->second.listeners.end())
         return false;
 
+    DHT_LOG.d(key, "[search %s] cancel listen", key.to_c_str());
+
     auto& listener = it->second;
-    if (!deviceKey_.empty()) {
+    listener.state->cancel = true;
+    if (not deviceKey_.empty()) {
         // First, be sure to have a token
         if (listener.thread.joinable()) {
             listener.thread.join();
         }
         // UNSUBSCRIBE
-        restbed::Uri uri(HTTP_PROTO + serverHost_ + "/" + key.toString());
+        restbed::Uri uri(proxy::HTTP_PROTO + serverHost_ + "/" + key.toString());
         auto req = std::make_shared<restbed::Request>(uri);
         req->set_method("UNSUBSCRIBE");
         // fill request body
         Json::Value body;
         body["key"] = deviceKey_;
         body["client_id"] = pushClientId_;
-        body["token"] = std::to_string(ltoken);
-        body["callback_id"] = listener.callbackId;
+        body["token"] = std::to_string(*listener.pushNotifToken);
         Json::StreamWriterBuilder wbuilder;
         wbuilder["commentStyle"] = "None";
         wbuilder["indentation"] = "";
@@ -639,16 +841,16 @@ DhtProxyClient::doCancelListen(const InfoHash& key, size_t ltoken)
     } else {
         // Just stop the request
         if (listener.thread.joinable()) {
-            // Close connection to stop listener?
-            *(listener.isCanceledViaClose) = true;
+            // Close connection to stop listener
             if (listener.req)
                 restbed::Http::close(listener.req);
             listener.thread.join();
         }
     }
     search->second.listeners.erase(it);
+    DHT_LOG.d(key, "[search %s] cancelListen: %zu listener remaining", key.to_c_str(), search->second.listeners.size());
     if (search->second.listeners.empty()) {
-        listeners_.erase(search);
+        searches_.erase(search);
     }
 
     return true;
@@ -657,7 +859,7 @@ DhtProxyClient::doCancelListen(const InfoHash& key, size_t ltoken)
 void
 DhtProxyClient::opFailed()
 {
-    DHT_LOG.ERR("Proxy request failed");
+    DHT_LOG.e("Proxy request failed");
     {
         std::lock_guard<std::mutex> l(lockCurrentProxyInfos_);
         statusIpv4_ = NodeStatus::Disconnected;
@@ -676,20 +878,31 @@ DhtProxyClient::getConnectivityStatus()
 void
 DhtProxyClient::restartListeners()
 {
-    std::lock_guard<std::mutex> lock(lockListener_);
-    for (auto& search: listeners_) {
+    if (not deviceKey_.empty()) {
+        return;
+    }
+    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;
-        if (listener.thread.joinable())
+        auto state = listener.state;
+        if (listener.thread.joinable()) {
+            state->cancel = true;
+            if (listener.req)
+                restbed::Http::close(listener.req);
             listener.thread.join();
+        }
         // Redo listen
+        state->cancel = false;
+        state->ok = true;
         auto filter = listener.filter;
         auto cb = listener.cb;
-        restbed::Uri uri(HTTP_PROTO + serverHost_ + "/" + search.first.toString());
+        restbed::Uri uri(proxy::HTTP_PROTO + serverHost_ + "/" + search.first.toString());
         auto req = std::make_shared<restbed::Request>(uri);
         req->set_method("LISTEN");
         listener.req = req;
-        listener.thread = std::thread([this, filter, cb, req]()
+        listener.thread = std::thread([this, filter, cb, req, state]()
             {
                 auto settings = std::make_shared<restbed::Settings>();
                 std::chrono::milliseconds timeout(std::numeric_limits<int>::max());
@@ -697,14 +910,16 @@ DhtProxyClient::restartListeners()
 
                 auto ok = std::make_shared<std::atomic_bool>(true);
                 restbed::Http::async(req,
-                    [this, filter, cb, ok](const std::shared_ptr<restbed::Request>& req,
-                                                const std::shared_ptr<restbed::Response>& reply) {
+                    [this, filter, cb, state](const std::shared_ptr<restbed::Request>& req,
+                                           const std::shared_ptr<restbed::Response>& reply) {
                     auto code = reply->get_status_code();
 
                     if (code == 200) {
                         try {
-                            while (restbed::Http::is_open(req)) {
+                            while (restbed::Http::is_open(req) and not state->cancel) {
                                 restbed::Http::fetch("\n", reply);
+                                if (state->cancel)
+                                    break;
                                 std::string body;
                                 reply->get_body(body);
                                 reply->set_body(""); // Reset the body for the next fetch
@@ -712,24 +927,16 @@ DhtProxyClient::restartListeners()
                                 Json::Value json;
                                 std::string err;
                                 Json::CharReaderBuilder rbuilder;
-                                auto* char_data = reinterpret_cast<const char*>(&body[0]);
                                 auto reader = std::unique_ptr<Json::CharReader>(rbuilder.newCharReader());
-                                if (reader->parse(char_data, char_data + body.size(), &json, &err)) {
+                                if (reader->parse(body.data(), body.data() + body.size(), &json, &err)) {
                                     auto value = std::make_shared<Value>(json);
-                                    if ((not filter or filter(*value)) && cb) {
-                                        auto okCb = std::make_shared<std::promise<bool>>();
-                                        auto futureCb = okCb->get_future();
-                                        {
-                                            std::lock_guard<std::mutex> lock(lockCallbacks);
-                                            callbacks_.emplace_back([cb, value, okCb](){
-                                                okCb->set_value(cb({value}, false));
-                                            });
-                                            loopSignal_();
-                                        }
-                                        futureCb.wait();
-                                        if (!futureCb.get()) {
-                                            return;
-                                        }
+                                    if ((not filter or filter(*value)) and cb)  {
+                                        std::lock_guard<std::mutex> lock(lockCallbacks);
+                                        callbacks_.emplace_back([cb, value, state]() {
+                                            if (not state->cancel and not cb({value}, false))
+                                                state->cancel = true;
+                                        });
+                                        loopSignal_();
                                     }
                                 }
                             }
@@ -737,10 +944,12 @@ DhtProxyClient::restartListeners()
                             // NOTE: Http::close() can occurs here. Ignore this.
                         }
                     } else {
-                        *ok = false;
+                        state->ok = false;
                     }
                 }, settings).get();
-                if (!ok) opFailed();
+                auto& s = *state;
+                if (not s.ok and not s.cancel)
+                    opFailed();
             }
         );
     }
@@ -751,29 +960,48 @@ void
 DhtProxyClient::pushNotificationReceived(const std::map<std::string, std::string>& notification)
 {
 #if OPENDHT_PUSH_NOTIFICATIONS
+    scheduler.syncTime();
+    DHT_LOG.d("Got push notification");
+    for (const auto& n : notification) {
+        DHT_LOG.d("   %s -> %s", n.first.c_str(), n.second.c_str());
+    }
     try {
-        auto token = std::stoul(notification.at("token"));
-        for (auto& search: listeners_) {
-            for (auto& list : search.second.listeners) {
-                auto& listener = list.second;
-                if (*listener.pushNotifToken!= token)
-                    continue;
-                if (notification.find("timeout") == notification.cend()) {
-                    // Wake up daemon and get values
-                    auto cb = listener.cb;
-                    auto filter = listener.filter;
-                    get(search.first, [cb](const std::vector<Sp<Value>>& vals){
+        std::lock_guard<std::mutex> lock(searchLock_);
+        auto timeout = notification.find("timeout");
+        if (timeout != notification.cend()) {
+            InfoHash key(timeout->second);
+            auto& search = searches_.at(key);
+            auto vidIt = notification.find("vid");
+            if (vidIt != notification.end()) {
+                // 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 {
+                // Refresh listen
+                for (auto& list : search.listeners)
+                    resubscribe(key, list.second);
+            }
+        } else {
+            auto token = std::stoull(notification.at("token"));
+            for (auto& search: searches_) {
+                for (auto& list : search.second.listeners) {
+                    if (*list.second.pushNotifToken != token or list.second.state->cancel)
+                        continue;
+                    DHT_LOG.d(search.first, "[search %s] handling push notification", search.first.to_c_str());
+                    auto cb = list.second.cb;
+                    auto filter = list.second.filter;
+                    get(search.first, [cb](const std::vector<Sp<Value>>& vals) {
                         cb(vals, false);
                         return true;
                     }, DoneCallbackSimple{}, std::move(filter));
-                } else {
-                    // A timeout has occured, we need to relaunch the listener
-                    resubscribe(search.first, listener);
                 }
             }
         }
     } catch (const std::exception& e) {
-        std::cerr << "pushNotificationReceived: error " << e.what() << std::endl;
+        DHT_LOG.e("Error handling push notification: %s", e.what());
     }
 #endif
 }
@@ -783,25 +1011,25 @@ DhtProxyClient::resubscribe(const InfoHash& key, Listener& listener)
 {
 #if OPENDHT_PUSH_NOTIFICATIONS
     if (deviceKey_.empty()) return;
+    DHT_LOG.d(key, "[search %s] resubscribe push listener", key.to_c_str());
     // Subscribe
-    restbed::Uri uri(HTTP_PROTO + serverHost_ + "/" + key.toString());
+    restbed::Uri uri(proxy::HTTP_PROTO + serverHost_ + "/" + key.toString());
     auto req = std::make_shared<restbed::Request>(uri);
     req->set_method("SUBSCRIBE");
 
-    auto pushNotifToken = std::make_shared<unsigned>(0);
-
+    auto pushNotifToken = std::make_shared<proxy::ListenToken>(0);
+    auto state = listener.state;
     if (listener.thread.joinable())
         listener.thread.join();
+    state->cancel = false;
+    state->ok = true;
     listener.req = req;
     listener.pushNotifToken = pushNotifToken;
-    auto callbackId = listener.callbackId;
-    listener.thread = std::thread([=]()
-    {
-        fillBodyToGetToken(req, callbackId);
+    listener.thread = std::thread([this, req, pushNotifToken, state]() {
+        fillBodyToGetToken(req);
         auto settings = std::make_shared<restbed::Settings>();
-        auto ok = std::make_shared<std::atomic_bool>(true);
         restbed::Http::async(req,
-            [this, pushNotifToken, ok](const std::shared_ptr<restbed::Request>&,
+            [pushNotifToken, state](const std::shared_ptr<restbed::Request>&,
                                        const std::shared_ptr<restbed::Response>& reply) {
             auto code = reply->get_status_code();
             if (code == 200) {
@@ -813,43 +1041,51 @@ DhtProxyClient::resubscribe(const InfoHash& key, Listener& listener)
                     std::string err;
                     Json::Value json;
                     Json::CharReaderBuilder rbuilder;
-                    auto* char_data = reinterpret_cast<const char*>(&body[0]);
                     auto reader = std::unique_ptr<Json::CharReader>(rbuilder.newCharReader());
-                    if (reader->parse(char_data, char_data + body.size(), &json, &err)) {
+                    if (reader->parse(body.data(), body.data() + body.size(), &json, &err)) {
                         if (!json.isMember("token")) return;
-                        *pushNotifToken = json["token"].asLargestUInt();
+                        *pushNotifToken = unpackId(json, "token");
                     }
                 } catch (std::runtime_error&) {
                     // NOTE: Http::close() can occurs here. Ignore this.
                 }
             } else {
-                *ok = false;
+                state->ok = false;
             }
         }, settings).get();
-        if (!ok) opFailed();
+        auto& s = *state;
+        if (not s.ok and not s.cancel)
+            opFailed();
     });
 #endif
 }
 
 #if OPENDHT_PUSH_NOTIFICATIONS
 void
-DhtProxyClient::fillBodyToGetToken(std::shared_ptr<restbed::Request> req, unsigned callbackId)
+DhtProxyClient::getPushRequest(Json::Value& body) const
 {
-    // Fill body with
-    // {
-    //   "key":"device_key",
-    //   "callback_id": xxx
-    // }
-    Json::Value body;
     body["key"] = deviceKey_;
     body["client_id"] = pushClientId_;
-    body["callback_id"] = callbackId;
 #ifdef __ANDROID__
     body["platform"] = "android";
 #endif
 #ifdef __APPLE__
     body["platform"] = "apple";
 #endif
+}
+
+void
+DhtProxyClient::fillBodyToGetToken(std::shared_ptr<restbed::Request> req, unsigned token)
+{
+    // Fill body with
+    // {
+    //   "key":"device_key",
+    //   "token": xxx
+    // }
+    Json::Value body;
+    getPushRequest(body);
+    if (token > 0)
+        body["token"] = token;
     Json::StreamWriterBuilder wbuilder;
     wbuilder["commentStyle"] = "None";
     wbuilder["indentation"] = "";

src/dhtrunner.cpp

@@ -120,7 +120,8 @@ DhtRunner::run(const SockAddr& local4, const SockAddr& local6, DhtRunner::Config
             } catch (const dht::SocketException& e) {
                 startNetwork(local4, local6);
             }
-            cv.wait_until(lk, wakeup, [this]() {
+
+            auto hasJobToDo = [this]() {
                 if (not running)
                     return true;
                 {
@@ -137,7 +138,11 @@ DhtRunner::run(const SockAddr& local4, const SockAddr& local6, DhtRunner::Config
                         return true;
                 }
                 return false;
-            });
+            };
+            if (wakeup == time_point::max())
+                cv.wait(lk, hasJobToDo);
+            else
+                cv.wait_until(lk, wakeup, hasJobToDo);
         }
     });
 }
@@ -595,8 +600,8 @@ void
 DhtRunner::cancelListen(InfoHash h, size_t token)
 {
     std::lock_guard<std::mutex> lck(storage_mtx);
+    pending_ops.emplace([=](SecureDht& dht) {
 #if OPENDHT_PROXY_CLIENT
-    pending_ops.emplace([=](SecureDht&) {
         auto it = listeners_.find(token);
         if (it == listeners_.end()) return;
         if (it->second.tokenClassicDht)
@@ -604,23 +609,31 @@ DhtRunner::cancelListen(InfoHash h, size_t token)
         if (it->second.tokenProxyDht and dht_via_proxy_)
             dht_via_proxy_->cancelListen(h, it->second.tokenProxyDht);
         listeners_.erase(it);
-    });
 #else
-    pending_ops.emplace([=](SecureDht& dht) {
         dht.cancelListen(h, token);
-    });
 #endif // OPENDHT_PROXY_CLIENT
+    });
     cv.notify_all();
 }
 
 void
-DhtRunner::cancelListen(InfoHash h, std::shared_future<size_t> token)
+DhtRunner::cancelListen(InfoHash h, std::shared_future<size_t> ftoken)
 {
     {
         std::lock_guard<std::mutex> lck(storage_mtx);
         pending_ops.emplace([=](SecureDht& dht) {
-            auto tk = token.get();
-            dht.cancelListen(h, tk);
+            auto token = ftoken.get();
+#if OPENDHT_PROXY_CLIENT
+            auto it = listeners_.find(token);
+            if (it == listeners_.end()) return;
+            if (it->second.tokenClassicDht)
+                dht_->cancelListen(h, it->second.tokenClassicDht);
+            if (it->second.tokenProxyDht and dht_via_proxy_)
+                dht_via_proxy_->cancelListen(h, it->second.tokenProxyDht);
+            listeners_.erase(it);
+#else
+            dht.cancelListen(h, token);
+#endif // OPENDHT_PROXY_CLIENT
         });
     }
     cv.notify_all();
@@ -953,11 +966,17 @@ DhtRunner::setPushNotificationToken(const std::string& token) {
 }
 
 void
-DhtRunner::pushNotificationReceived(const std::map<std::string, std::string>& data) const
+DhtRunner::pushNotificationReceived(const std::map<std::string, std::string>& data)
 {
 #if OPENDHT_PROXY_CLIENT && OPENDHT_PUSH_NOTIFICATIONS
-    if (dht_via_proxy_)
-        dht_via_proxy_->pushNotificationReceived(data);
+    {
+        std::lock_guard<std::mutex> lck(storage_mtx);
+        pending_ops_prio.emplace([=](SecureDht& dht) {
+            if (dht_via_proxy_)
+                dht_via_proxy_->pushNotificationReceived(data);
+        });
+    }
+    cv.notify_all();
 #endif
 }
 

src/listener.h

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

src/network_engine.cpp

@@ -140,7 +140,7 @@ RequestAnswer::RequestAnswer(ParsedMessage&& msg)
 {}
 
 NetworkEngine::NetworkEngine(Logger& log, Scheduler& scheduler, const int& s, const int& s6)
-    : myid(zeroes), DHT_LOG(log), scheduler(scheduler), dht_socket(s), dht_socket6(s6)
+    : myid(zeroes), dht_socket(s), dht_socket6(s6), DHT_LOG(log), scheduler(scheduler)
 {}
 NetworkEngine::NetworkEngine(InfoHash& myid, NetId net, const int& s, const int& s6, Logger& log, Scheduler& scheduler,
         decltype(NetworkEngine::onError) onError,
@@ -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);
@@ -610,6 +614,7 @@ NetworkEngine::process(std::unique_ptr<ParsedMessage>&& msg, const SockAddr& fro
                 DHT_LOG.d(msg->info_hash, node->id, "[node %s] got 'listen' request for %s", node->toString().c_str(), msg->info_hash.toString().c_str());
                 ++in_stats.listen;
                 RequestAnswer answer = onListen(node, msg->info_hash, msg->token, msg->socket_id, std::move(msg->query));
+                auto nnodes = bufferNodes(from.getFamily(), msg->info_hash, msg->want, answer.nodes4, answer.nodes6);
                 sendListenConfirmation(from, msg->tid);
                 break;
             }

src/op_cache.cpp

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

src/op_cache.h

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

src/search.h

@@ -63,6 +63,7 @@ struct Dht::SearchNode {
 
     struct CachedListenStatus {
         ValueCache cache;
+        Sp<Scheduler::Job> cacheExpirationJob {};
         Sp<net::Request> req {};
         CachedListenStatus(ValueStateCallback&& cb) : cache(std::forward<ValueStateCallback>(cb)) {}
         CachedListenStatus(CachedListenStatus&&) = default;
@@ -213,13 +214,22 @@ struct Dht::SearchNode {
         getStatus.clear();
     }
 
-    void onValues(const Sp<Query>& q, net::RequestAnswer&& answer, const TypeStore& types, const time_point& now)
+    void onValues(const Sp<Query>& q, net::RequestAnswer&& answer, const TypeStore& types, Scheduler& scheduler)
     {
         auto l = listenStatus.find(q);
         if (l != listenStatus.end()) {
-            l->second.cache.onValues(answer.values,
+            auto next = l->second.cache.onValues(answer.values,
                                      answer.refreshed_values,
-                                     answer.expired_values, types, now);
+                                     answer.expired_values, types, scheduler.time());
+            scheduler.edit(l->second.cacheExpirationJob, next);
+        }
+    }
+
+    void expireValues(const Sp<Query>& q, Scheduler& scheduler) {
+        auto l = listenStatus.find(q);
+        if (l != listenStatus.end()) {
+            auto next = l->second.cache.expireValues(scheduler.time());
+            scheduler.edit(l->second.cacheExpirationJob, next);
         }
     }
 
@@ -373,8 +383,11 @@ struct Dht::Search {
 
     /* Cache */
     SearchCache cache;
+    Sp<Scheduler::Job> opExpirationJob;
 
     ~Search() {
+        if (opExpirationJob)
+            opExpirationJob->cancel();
         for (auto& get : callbacks) {
             get.second.done_cb(false, {});
             get.second.done_cb = {};
@@ -473,21 +486,27 @@ struct Dht::Search {
         });
     }
 
-    void cancelListen(size_t token) {
-        cache.cancelListen(token, [&](size_t t){
-            Sp<Query> query;
-            const auto& ll = listeners.find(t);
-            if (ll != listeners.cend()) {
-                query = ll->second.query;
-                listeners.erase(ll);
-            }
-            for (auto& sn : nodes) {
-                if (listeners.empty())
-                    sn.cancelListen();
-                else if (query)
-                    sn.cancelListen(query);
-            }
-        });
+    void cancelListen(size_t token, Scheduler& scheduler) {
+        cache.cancelListen(token, scheduler.time());
+        if (not opExpirationJob)
+            opExpirationJob = scheduler.add(time_point::max(), [this,&scheduler]{
+                auto nextExpire = cache.expire(scheduler.time(), [&](size_t t){
+                    Sp<Query> query;
+                    const auto& ll = listeners.find(t);
+                    if (ll != listeners.cend()) {
+                        query = ll->second.query;
+                        listeners.erase(ll);
+                    }
+                    for (auto& sn : nodes) {
+                        if (listeners.empty())
+                            sn.cancelListen();
+                        else if (query)
+                            sn.cancelListen(query);
+                    }
+                });
+                scheduler.edit(opExpirationJob, nextExpire);
+            });
+        scheduler.edit(opExpirationJob, cache.getExpiration());
     }
 
     /**

src/securedht.cpp

@@ -213,11 +213,13 @@ SecureDht::findPublicKey(const InfoHash& node, std::function<void(const Sp<const
     findCertificate(node, [=](const Sp<crypto::Certificate> crt) {
         if (crt && *crt) {
             auto pk = std::make_shared<crypto::PublicKey>(crt->getPublicKey());
-            nodesPubKeys_[pk->getId()] = pk;
-            if (cb) cb(pk);
-        } else {
-            if (cb) cb(nullptr);
+            if (*pk) {
+                nodesPubKeys_[pk->getId()] = pk;
+                if (cb) cb(pk);
+                return;
+            }
         }
+        if (cb) cb(nullptr);
     });
 }
 
@@ -236,7 +238,8 @@ SecureDht::checkValue(const Sp<Value>& v)
         try {
             Value decrypted_val (decrypt(*v));
             if (decrypted_val.recipient == getId()) {
-                nodesPubKeys_[decrypted_val.owner->getId()] = decrypted_val.owner;
+                if (decrypted_val.owner)
+                    nodesPubKeys_[decrypted_val.owner->getId()] = decrypted_val.owner;
                 return std::make_shared<Value>(std::move(decrypted_val));
             }
             // Ignore values belonging to other people
@@ -317,6 +320,7 @@ SecureDht::putSigned(const InfoHash& hash, Sp<Value> val, DoneCallback callback,
 {
     if (val->id == Value::INVALID_ID) {
         crypto::random_device rdev;
+        std::uniform_int_distribution<Value::Id> rand_id;
         val->id = rand_id(rdev);
     }
 

src/utils.cpp

@@ -29,6 +29,26 @@ namespace dht {
 
 static constexpr std::array<uint8_t, 12> MAPPED_IPV4_PREFIX {{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xff, 0xff}};
 
+std::pair<std::string, std::string>
+splitPort(const std::string& s) {
+    if (s.empty())
+        return {};
+    if (s[0] == '[') {
+        std::size_t closure = s.find_first_of(']');
+        std::size_t found = s.find_last_of(':');
+        if (closure == std::string::npos)
+            return {s, ""};
+        if (found == std::string::npos or found < closure)
+            return {s.substr(1,closure-1), ""};
+        return {s.substr(1,closure-1), s.substr(found+1)};
+    }
+    std::size_t found = s.find_last_of(':');
+    std::size_t first = s.find_first_of(':');
+    if (found == std::string::npos or found != first)
+        return {s, ""};
+    return {s.substr(0,found), s.substr(found+1)};
+}
+
 std::vector<SockAddr>
 SockAddr::resolve(const std::string& host, const std::string& service)
 {

src/value.cpp

@@ -175,45 +175,37 @@ Value::msgpack_unpack_body(const msgpack::object& o)
 #if OPENDHT_PROXY_SERVER || OPENDHT_PROXY_CLIENT
 Value::Value(Json::Value& json)
 {
-   try {
-       if (json.isMember("id")) {
-           if (json["id"].isString()) {
-               id = Value::Id(std::stoull(json["id"].asString()));
-           } else {
-               id = Value::Id(json["id"].asLargestUInt());
-           }
-       }
-   } catch (...) { }
-   if (json.isMember("cypher")) {
-       auto cypherStr = json["cypher"].asString();
-       cypherStr = base64_decode(cypherStr);
-       cypher = std::vector<unsigned char>(cypherStr.begin(), cypherStr.end());
-   }
-   if (json.isMember("sig")) {
-       auto sigStr = json["sig"].asString();
-       sigStr = base64_decode(sigStr);
-       signature = std::vector<unsigned char>(sigStr.begin(), sigStr.end());
-   }
-   if (json.isMember("seq"))
-       seq = json["seq"].asInt();
-   if (json.isMember("owner")) {
-       auto ownerStr = json["owner"].asString();
-       auto ownerBlob = std::vector<unsigned char>(ownerStr.begin(), ownerStr.end());
-       owner = std::make_shared<const crypto::PublicKey>(ownerBlob);
-   }
-   if (json.isMember("to")) {
-       auto toStr = json["to"].asString();
-       recipient = InfoHash(toStr);
-   }
-   if (json.isMember("type"))
-       type = json["type"].asInt();
-   if (json.isMember("data")){
-       auto dataStr = json["data"].asString();
-       dataStr = base64_decode(dataStr);
-       data = std::vector<unsigned char>(dataStr.begin(), dataStr.end());
-   }
-   if (json.isMember("utype"))
-       user_type = json["utype"].asString();
+    id = Value::Id(unpackId(json, "id"));
+    if (json.isMember("cypher")) {
+        auto cypherStr = json["cypher"].asString();
+        cypherStr = base64_decode(cypherStr);
+        cypher = std::vector<unsigned char>(cypherStr.begin(), cypherStr.end());
+    }
+    if (json.isMember("sig")) {
+        auto sigStr = json["sig"].asString();
+        sigStr = base64_decode(sigStr);
+        signature = std::vector<unsigned char>(sigStr.begin(), sigStr.end());
+    }
+    if (json.isMember("seq"))
+        seq = json["seq"].asInt();
+    if (json.isMember("owner")) {
+        auto ownerStr = json["owner"].asString();
+        auto ownerBlob = std::vector<unsigned char>(ownerStr.begin(), ownerStr.end());
+        owner = std::make_shared<const crypto::PublicKey>(ownerBlob);
+    }
+    if (json.isMember("to")) {
+        auto toStr = json["to"].asString();
+        recipient = InfoHash(toStr);
+    }
+    if (json.isMember("type"))
+        type = json["type"].asInt();
+    if (json.isMember("data")){
+        auto dataStr = json["data"].asString();
+        dataStr = base64_decode(dataStr);
+        data = std::vector<unsigned char>(dataStr.begin(), dataStr.end());
+    }
+    if (json.isMember("utype"))
+        user_type = json["utype"].asString();
 }
 
 Json::Value
@@ -240,6 +232,22 @@ Value::toJson() const
     }
     return val;
 }
+
+uint64_t
+unpackId(const Json::Value& json, const std::string& key) {
+    uint64_t ret = 0;
+    try {
+        if (json.isMember(key)) {
+            const auto& t = json[key];
+            if (t.isString()) {
+                ret = std::stoull(t.asString());
+            } else {
+                ret = t.asLargestUInt();
+            }
+        }
+    } catch (...) {}
+    return ret;
+}
 #endif //OPENDHT_PROXY_SERVER
 
 bool

src/value_cache.h

@@ -26,10 +26,10 @@ using CallbackQueue = std::list<std::function<void()>>;
 
 class ValueCache {
 public:
-    ValueCache(ValueStateCallback&& cb) {
-        callbacks.emplace_back(std::forward<ValueStateCallback>(cb));
+    ValueCache(ValueStateCallback&& cb) : callback(std::forward<ValueStateCallback>(cb)) {}
+    ValueCache(ValueCache&& o) : values(std::move(o.values)), callback(std::move(o.callback)) {
+        o.callback = {};
     }
-    ValueCache(ValueCache&&) = default;
 
     ~ValueCache() {
         auto q = clear();
@@ -44,24 +44,35 @@ public:
             expired_values.emplace_back(std::move(v.second.data));
         values.clear();
         CallbackQueue ret;
-        if (not expired_values.empty() and not callbacks.empty()) {
-            auto cbs = callbacks;
-            ret.emplace_back([expired_values, cbs]{
-                for (auto& cb : cbs)
-                    cb(expired_values, true);
+        if (not expired_values.empty() and callback) {
+            auto cb = callback;
+            ret.emplace_back([expired_values, cb]{
+                cb(expired_values, true);
             });
         }
         return ret;
     }
 
-    CallbackQueue expireValues(const time_point& now) {
+    time_point expireValues(const time_point& now) {
+        time_point ret = time_point::max();
+        auto cbs = expireValues(now, ret);
+        while (not cbs.empty()) {
+            cbs.front()();
+            cbs.pop_front();
+        }
+        return ret;
+    }
+
+    CallbackQueue expireValues(const time_point& now, time_point& next) {
         std::vector<Sp<Value>> expired_values;
         for (auto it = values.begin(); it != values.end();) {
             if (it->second.expiration < now) {
                 expired_values.emplace_back(std::move(it->second.data));
                 it = values.erase(it);
-            } else
+            } else {
+                next = std::min(next, it->second.expiration);
                 ++it;
+            }
         }
         while (values.size() >= MAX_VALUES) {
             // too many values, remove oldest values
@@ -78,40 +89,42 @@ public:
             }
         }
         CallbackQueue ret;
-        if (not expired_values.empty()) {
-            auto cbs = callbacks;
-            ret.emplace_back([cbs, expired_values]{
-                for (auto& cb : cbs)
-                    if (cb) cb(expired_values, true);
+        if (not expired_values.empty() and callback) {
+            auto cb = callback;
+            ret.emplace_back([cb, expired_values]{
+                if (cb) cb(expired_values, true);
             });
         }
         return ret;
     }
 
-    void onValues
+    time_point onValues
         (const std::vector<Sp<Value>>& values,
         const std::vector<Value::Id>& refreshed_values,
         const std::vector<Value::Id>& expired_values,
         const TypeStore& types, const time_point& now)
     {
         CallbackQueue cbs;
+        time_point ret = time_point::max();
         if (not values.empty())
             cbs.splice(cbs.end(), addValues(values, types, now));
         for (const auto& vid : refreshed_values)
             refreshValue(vid, types, now);
         for (const auto& vid : expired_values)
             cbs.splice(cbs.end(), expireValue(vid));
-        cbs.splice(cbs.end(), expireValues(now));
+        cbs.splice(cbs.end(), expireValues(now, ret));
         while (not cbs.empty()) {
             cbs.front()();
             cbs.pop_front();
         }
+        return ret;
     }
 
 private:
     // prevent copy
     ValueCache(const ValueCache&) = delete;
     ValueCache& operator=(const ValueCache&) = delete;
+    ValueCache& operator=(ValueCache&&) = delete;
 
     /* The maximum number of values we store in the cache. */
     static constexpr unsigned MAX_VALUES {1024};
@@ -127,27 +140,27 @@ private:
     };
 
     std::map<Value::Id, CacheValueStorage> values;
-    std::vector<ValueStateCallback> callbacks;
+    ValueStateCallback callback;
 
     CallbackQueue addValues(const std::vector<Sp<Value>>& new_values, const TypeStore& types, const time_point& now) {
         std::vector<Sp<Value>> nvals;
         for (const auto& value : new_values) {
             auto v = values.find(value->id);
             if (v == values.end()) {
-                nvals.emplace_back(value);
                 // new value
+                nvals.emplace_back(value);
                 values.emplace(value->id, CacheValueStorage(value, now, now + types.getType(value->type).expiration));
             } else {
+                // refreshed value
                 v->second.created = now;
                 v->second.expiration = now + types.getType(v->second.data->type).expiration;
             }
         }
-        auto cbs = callbacks;
+        auto cb = callback;
         CallbackQueue ret;
         if (not nvals.empty())
-            ret.emplace_back([cbs, nvals]{
-                for (auto& cb : cbs)
-                    if (cb) cb(nvals, false);
+            ret.emplace_back([cb, nvals]{
+                if (cb) cb(nvals, false);
             });
         return ret;
     }
@@ -157,11 +170,10 @@ private:
             return {};
         const std::vector<Sp<Value>> val {std::move(v->second.data)};
         values.erase(v);
-        auto cbs = callbacks;
+        auto cb = callback;
         CallbackQueue ret;
-        ret.emplace_back([cbs, val]{
-            for (auto& cb : cbs)
-                if (cb) cb(val, true);
+        ret.emplace_back([cb, val]{
+            if (cb) cb(val, true);
         });
         return ret;
     }
@@ -174,5 +186,4 @@ private:
     }
 };
 
-
 }

tests/Makefile.am

@@ -3,7 +3,7 @@ bin_PROGRAMS = opendht_unit_tests
 
 AM_CPPFLAGS = -I../include
 
-nobase_include_HEADERS = infohashtester.h
-opendht_unit_tests_SOURCES = tests_runner.cpp infohashtester.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/dhtrunnertester.cpp

@@ -0,0 +1,64 @@
+/*
+ *  Copyright (C) 2018 Savoir-faire Linux Inc.
+ *
+ *  Author: Adrien Béraud <adrien.beraud@savoirfairelinux.com>
+ *
+ *  This program is free software; you can redistribute it and/or modify
+ *  it under the terms of the GNU General Public License as published by
+ *  the Free Software Foundation; either version 3 of the License, or
+ *  (at your option) any later version.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+
+#include "dhtrunnertester.h"
+
+// std
+#include <iostream>
+#include <string>
+
+namespace test {
+CPPUNIT_TEST_SUITE_REGISTRATION(DhtRunnerTester);
+
+void
+DhtRunnerTester::setUp() {
+    node1.run(42222, {}, true);
+    node2.run(42232, {}, true);
+    node2.bootstrap(node1.getBound());
+}
+
+void
+DhtRunnerTester::tearDown() {
+    node1.join();
+    node2.join();
+}
+
+void
+DhtRunnerTester::testConstructors() {
+    CPPUNIT_ASSERT(node1.getBoundPort() == 42222);
+    CPPUNIT_ASSERT(node2.getBoundPort() == 42232);
+}
+
+void
+DhtRunnerTester::testGetPut() {
+    auto key = dht::InfoHash::get("123");
+    dht::Value val {"hey"};
+    auto val_data = val.data;
+    node1.put(key, std::move(val));
+    auto vals = node2.get(key).get();
+    CPPUNIT_ASSERT(not vals.empty());
+    CPPUNIT_ASSERT(vals.front()->data == val_data);
+}
+
+void
+DhtRunnerTester::testListen() {
+    // TODO
+}
+
+}  // namespace test

tests/dhtrunnertester.h

@@ -0,0 +1,62 @@
+/*
+ *  Copyright (C) 2018 Savoir-faire Linux Inc.
+ *
+ *  Author: Adrien Béraud <adrien.beraud@savoirfairelinux.com>
+ *
+ *  This program is free software; you can redistribute it and/or modify
+ *  it under the terms of the GNU General Public License as published by
+ *  the Free Software Foundation; either version 3 of the License, or
+ *  (at your option) any later version.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+
+#pragma once
+
+// cppunit
+#include <cppunit/TestFixture.h>
+#include <cppunit/extensions/HelperMacros.h>
+
+#include <opendht/dhtrunner.h>
+
+namespace test {
+
+class DhtRunnerTester : public CppUnit::TestFixture {
+    CPPUNIT_TEST_SUITE(DhtRunnerTester);
+    CPPUNIT_TEST(testConstructors);
+    CPPUNIT_TEST(testGetPut);
+    CPPUNIT_TEST(testListen);
+    CPPUNIT_TEST_SUITE_END();
+
+    dht::DhtRunner node1 {};
+    dht::DhtRunner node2 {};
+ public:
+    /**
+     * Method automatically called before each test by CppUnit
+     */
+    void setUp();
+    /**
+     * Method automatically called after each test CppUnit
+     */
+    void tearDown();
+    /**
+     * Test the differents behaviors of constructors
+     */
+    void testConstructors();
+    /**
+     * Test get and put methods
+     */
+    void testGetPut();
+    /**
+     * Test listen method
+     */
+    void testListen();
+};
+
+}  // namespace test

tests/infohashtester.cpp

@@ -14,8 +14,7 @@
  *  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, write to the Free Software
- *  Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301 USA.
+ *  along with this program. If not, see <https://www.gnu.org/licenses/>.
  */
 
 #include "infohashtester.h"

tests/infohashtester.h

@@ -14,8 +14,7 @@
  *  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, write to the Free Software
- *  Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301 USA.
+ *  along with this program. If not, see <https://www.gnu.org/licenses/>.
  */
 
 #pragma once

tools/dhtnode.cpp

@@ -73,7 +73,7 @@ void print_help() {
 #if OPENDHT_PUSH_NOTIFICATIONS
               << "  stt [server_address] <device_key> Start the proxy client." << std::endl
               << "  rs  [token]                       Resubscribe to opendht." << std::endl
-              << "  rp  [push_notification]           Inject a push notification in Opendht." << std::endl
+              << "  rp  [token]                       Inject a push notification in Opendht." << std::endl
 #else
               << "  stt [server_address]              Start the proxy client." << std::endl
 #endif // OPENDHT_PUSH_NOTIFICATIONS
@@ -87,8 +87,10 @@ void print_help() {
               << "  cl <key> <token>      Cancel listen for <token> and <key>." << std::endl
               << "  p <key> <str>         Put string value at <key>." << std::endl
               << "  pp <key> <str>        Put string value at <key> (persistent version)." << std::endl
+              << "  cpp <key> <id>        Cancel persistent put operation for <key> and value <id>." << std::endl
               << "  s <key> <str>         Put string value at <key>, signed with our generated private key." << std::endl
-              << "  e <key> <dest> <str>  Put string value at <key>, encrypted for <dest> with its public key (if found)." << std::endl;
+              << "  e <key> <dest> <str>  Put string value at <key>, encrypted for <dest> with its public key (if found)." << std::endl
+              << "  cc                    Trigger connectivity changed signal." << std::endl;
 
 #ifdef OPENDHT_INDEXATION
     std::cout << std::endl << "Indexation operations on the DHT:" << std::endl
@@ -139,6 +141,12 @@ void cmd_loop(std::shared_ptr<DhtRunner>& dht, dht_params& params
             std::cout << dht->getNodesStats(AF_INET).toString() << std::endl;
             std::cout << "IPv6 stats:" << std::endl;
             std::cout << dht->getNodesStats(AF_INET6).toString() << std::endl;
+#if OPENDHT_PROXY_SERVER
+            for (const auto& proxy : proxies) {
+                std::cout << "Stats for proxy on port " << proxy.first << std::endl;
+                std::cout << "  " << proxy.second->getStats().toString() << std::endl;
+            }
+#endif
             continue;
         } else if (op == "lr") {
             std::cout << "IPv4 routing table:" << std::endl;
@@ -172,11 +180,8 @@ void cmd_loop(std::shared_ptr<DhtRunner>& dht, dht_params& params
             iss >> idstr;
             try {
                 auto addr = splitPort(idstr);
-                if (not addr.first.empty() and addr.second.empty()){
-                    std::stringstream ss;
-                    ss << DHT_DEFAULT_PORT;
-                    addr.second = ss.str();
-                }
+                if (not addr.first.empty() and addr.second.empty())
+                    addr.second = std::to_string(DHT_DEFAULT_PORT);
                 dht->bootstrap(addr.first.c_str(), addr.second.c_str());
             } catch (const std::exception& e) {
                 std::cerr << e.what() << std::endl;
@@ -192,6 +197,8 @@ void cmd_loop(std::shared_ptr<DhtRunner>& dht, dht_params& params
                 log::disableLogging(*dht);
             dht->setLogFilter(filter);
             continue;
+        } else if (op == "cc") {
+            dht->connectivityChanged();
         }
 #if OPENDHT_PROXY_SERVER
         else if (op == "pst") {
@@ -230,7 +237,7 @@ void cmd_loop(std::shared_ptr<DhtRunner>& dht, dht_params& params
 #if OPENDHT_PUSH_NOTIFICATIONS
         else if (op == "rp") {
             iss >> value;
-            dht->pushNotificationReceived({{"token", value}});
+            dht->pushNotificationReceived({{"to", "dhtnode"}, {"token", value}});
             continue;
         }
 #endif // OPENDHT_PUSH_NOTIFICATIONS
@@ -491,6 +498,7 @@ main(int argc, char **argv)
 
         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;
         config.threaded = true;
         config.proxy_server = params.proxyclient;

tools/tools_common.h

@@ -45,29 +45,6 @@
 #include <sstream>
 #include <fstream>
 
-/**
- * Split "[host]:port" or "host:port" to pair<"host", "port">.
- */
-std::pair<std::string, std::string>
-splitPort(const std::string& s) {
-    if (s.empty())
-        return {};
-    if (s[0] == '[') {
-        std::size_t closure = s.find_first_of(']');
-        std::size_t found = s.find_last_of(':');
-        if (closure == std::string::npos)
-            return {s, ""};
-        if (found == std::string::npos or found < closure)
-            return {s.substr(1,closure-1), ""};
-        return {s.substr(1,closure-1), s.substr(found+1)};
-    }
-    std::size_t found = s.find_last_of(':');
-    std::size_t first = s.find_first_of(':');
-    if (found == std::string::npos or found != first)
-        return {s, ""};
-    return {s.substr(0,found), s.substr(found+1)};
-}
-
 /*
  * The mapString shall have the following format:
  *
@@ -181,7 +158,7 @@ parseArgs(int argc, char **argv) {
             params.network = strtoul(optarg, nullptr, 0);
             break;
         case 'b':
-            params.bootstrap = splitPort((optarg[0] == '=') ? optarg+1 : optarg);
+            params.bootstrap = dht::splitPort((optarg[0] == '=') ? optarg+1 : optarg);
             if (not params.bootstrap.first.empty() and params.bootstrap.second.empty()) {
                 params.bootstrap.second = std::to_string(DHT_DEFAULT_PORT);
             }