proxyclient: avoid to generate tokens with \0

The rest is not needed.

CMakeLists.txt

@@ -4,7 +4,7 @@ include(CMakeDependentOption)
 
 project (opendht)
 set (opendht_VERSION_MAJOR 2)
-set (opendht_VERSION_MINOR 0.0)
+set (opendht_VERSION_MINOR 1.0)
 set (opendht_VERSION ${opendht_VERSION_MAJOR}.${opendht_VERSION_MINOR})
 set (PACKAGE_VERSION ${opendht_VERSION})
 set (VERSION "${opendht_VERSION}")
@@ -37,7 +37,7 @@ option (OPENDHT_DOCUMENTATION "Create and install the HTML based API documentati
 # Dependencies
 list (APPEND CMAKE_MODULE_PATH "${PROJECT_SOURCE_DIR}/cmake")
 find_package (Threads)
-find_package (PkgConfig)
+find_package (PkgConfig REQUIRED)
 find_package (GnuTLS 3.3 REQUIRED)
 pkg_search_module (Nettle nettle)
 find_package (Msgpack 1.2 REQUIRED)
@@ -45,7 +45,6 @@ if (OPENDHT_TOOLS)
     find_package (Readline 6 REQUIRED)
 endif ()
 if (NOT OPENDHT_ARGON2)
-    find_package(PkgConfig)
     pkg_search_module(argon2 libargon2)
     if (argon2_FOUND)
         message("-- Found Argon2: " ${argon2_LIBRARY_DIRS} " (found version \"" ${argon2_VERSION} "\")")
@@ -300,9 +299,9 @@ if (OPENDHT_STATIC)
     endif ()
     target_link_libraries(opendht-static
         PRIVATE  ${argon2_LIBRARIES}
-        PUBLIC ${CMAKE_THREAD_LIBS_INIT} ${GNUTLS_LIBRARIES} ${Nettle_LIBRARIES}
-               ${Jsoncpp_LIBRARIES} ${FMT_LIBRARY} ${HTTP_PARSER_LIBRARY}
-               ${OPENSSL_LIBRARIES})
+        PUBLIC ${CMAKE_THREAD_LIBS_INIT} ${GNUTLS_LIBRARIES} ${Nettle_STATIC_LIBRARIES}
+               ${Jsoncpp_STATIC_LIBRARIES} ${FMT_LIBRARY} ${HTTP_PARSER_LIBRARY}
+               ${OPENSSL_STATIC_LIBRARIES})
     install (TARGETS opendht-static DESTINATION ${CMAKE_INSTALL_LIBDIR} EXPORT opendht)
 endif ()
 
@@ -380,7 +379,7 @@ install (FILES ${CMAKE_CURRENT_BINARY_DIR}/opendhtConfigVersion.cmake DESTINATIO
 
 # Unit tests
 if (OPENDHT_TESTS)
-    FIND_PACKAGE(Cppunit REQUIRED)
+    pkg_search_module(Cppunit REQUIRED cppunit)
     # unit testing
     list (APPEND test_FILES
         tests/infohashtester.h
@@ -412,7 +411,8 @@ if (OPENDHT_TESTS)
         tests/tests_runner.cpp
         ${test_FILES}
     )
-    target_include_directories(opendht_unit_tests SYSTEM PRIVATE ${CPPUNIT_INCLUDE_DIRS})
+    target_include_directories(opendht_unit_tests SYSTEM PRIVATE ${Cppunit_INCLUDE_DIR})
+    target_link_directories(opendht_unit_tests PRIVATE ${Cppunit_LIBRARY_DIRS})
     if (OPENDHT_SHARED)
         target_link_libraries(opendht_unit_tests opendht)
     else ()
@@ -420,7 +420,7 @@ if (OPENDHT_TESTS)
     endif ()
     target_link_libraries(opendht_unit_tests
        ${CMAKE_THREAD_LIBS_INIT}
-       ${CPPUNIT_LIBRARIES}
+       ${Cppunit_LIBRARIES}
        ${GNUTLS_LIBRARIES}
        ${Jsoncpp_LIBRARIES}
     )

cmake/FindCppunit.cmake

@@ -1,28 +0,0 @@
-INCLUDE(FindPkgConfig)
-PKG_CHECK_MODULES(PC_CPPUNIT "cppunit")
-
-FIND_PATH(CPPUNIT_INCLUDE_DIRS
-    NAMES cppunit/TestCase.h
-    HINTS ${PC_CPPUNIT_INCLUDE_DIR}
-    ${CMAKE_INSTALL_PREFIX}/include
-    PATHS
-    /usr/local/include
-    /usr/include
-)
-
-FIND_LIBRARY(CPPUNIT_LIBRARIES
-    NAMES cppunit
-    HINTS ${PC_CPPUNIT_LIBDIR}
-    ${CMAKE_INSTALL_PREFIX}/lib
-    ${CMAKE_INSTALL_PREFIX}/lib64
-    PATHS
-    ${CPPUNIT_INCLUDE_DIRS}/../lib
-    /usr/local/lib
-    /usr/lib
-)
-
-LIST(APPEND CPPUNIT_LIBRARIES ${CMAKE_DL_LIBS})
-
-INCLUDE(FindPackageHandleStandardArgs)
-FIND_PACKAGE_HANDLE_STANDARD_ARGS(CPPUNIT DEFAULT_MSG CPPUNIT_LIBRARIES CPPUNIT_INCLUDE_DIRS)
-MARK_AS_ADVANCED(CPPUNIT_LIBRARIES CPPUNIT_INCLUDE_DIRS)

configure.ac

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

include/opendht/callbacks.h

@@ -82,6 +82,8 @@ struct OPENDHT_PUBLIC NodeInfo {
     NodeStats ipv4 {};
     NodeStats ipv6 {};
     size_t ongoing_ops {0};
+    in_port_t bound4 {0};
+    in_port_t bound6 {0};
 
 #ifdef OPENDHT_JSONCPP
     /**

include/opendht/dht.h

@@ -445,7 +445,7 @@ private:
     void reportedAddr(const SockAddr&);
 
     // Storage
-    void storageAddListener(const InfoHash& id, const Sp<Node>& node, size_t tid, Query&& = {});
+    void storageAddListener(const InfoHash& id, const Sp<Node>& node, size_t tid, Query&& = {}, int version = 0);
     bool storageStore(const InfoHash& id, const Sp<Value>& value, time_point created, const SockAddr& sa = {}, bool permanent = false);
     bool storageErase(const InfoHash& id, Value::Id vid);
     bool storageRefresh(const InfoHash& id, Value::Id vid);
@@ -600,7 +600,8 @@ private:
             const InfoHash& hash,
             const Blob& token,
             size_t socket_id,
-            const Query& query);
+            const Query& query,
+            int version = 0);
     void onListenDone(const Sp<Node>& status,
             net::RequestAnswer& a,
             Sp<Search>& sr);

include/opendht/dht_proxy_server.h

@@ -36,6 +36,15 @@
 #include <memory>
 #include <mutex>
 
+namespace dht {
+enum class PushType {
+    None = 0,
+    Android,
+    iOS
+};
+}
+MSGPACK_ADD_ENUM(dht::PushType)
+
 namespace http {
 class Request;
 struct ListenerSession;
@@ -52,6 +61,13 @@ class DhtRunner;
 using RestRouter = restinio::router::express_router_t<>;
 using RequestStatus = restinio::request_handling_status_t;
 
+struct ProxyServerConfig {
+    in_port_t port {8000};
+    std::string pushServer {};
+    std::string persistStatePath {};
+    dht::crypto::Identity identity {};
+};
+
 /**
  * Describes the REST API
  */
@@ -66,10 +82,9 @@ public:
      * @note if the server fails to start (if port is already used or reserved),
      * it will fails silently
      */
-    DhtProxyServer(
-       dht::crypto::Identity identity,
-       std::shared_ptr<DhtRunner> dht, in_port_t port = 8000, const std::string& pushServer = "",
-       std::shared_ptr<dht::Logger> logger = {});
+    DhtProxyServer(const std::shared_ptr<DhtRunner>& dht,
+        const ProxyServerConfig& config = {},
+        const std::shared_ptr<dht::Logger>& logger = {});
 
     virtual ~DhtProxyServer();
 
@@ -282,17 +297,17 @@ private:
      * @param key of the device
      * @param json, the content to send
      */
-    void sendPushNotification(const std::string& key, Json::Value&& json, bool isAndroid, bool highPriority);
+    void sendPushNotification(const std::string& key, Json::Value&& json, PushType type, bool highPriority);
 
     /**
      * Send push notification with an expire timeout.
      * @param ec
      * @param pushToken
      * @param json
-     * @param isAndroid
+     * @param type
      */
     void handleNotifyPushListenExpire(const asio::error_code &ec, const std::string pushToken,
-                                      std::function<Json::Value()> json, const bool isAndroid);
+                                      std::function<Json::Value()> json, PushType type);
 
     /**
      * Remove a push listener between a client and a hash
@@ -309,6 +324,12 @@ private:
     void handlePrintStats(const asio::error_code &ec);
     void updateStats();
 
+    template <typename Os>
+    void saveState(Os& stream);
+
+    template <typename Is>
+    void loadState(Is& is, size_t size);
+
     using clock = std::chrono::steady_clock;
     using time_point = clock::time_point;
 
@@ -316,6 +337,9 @@ private:
     std::shared_ptr<DhtRunner> dht_;
     Json::StreamWriterBuilder jsonBuilder_;
     Json::CharReaderBuilder jsonReaderBuilder_;
+    std::mt19937_64 rd {crypto::getSeededRandomEngine<std::mt19937_64>()};
+
+    std::string persistPath_;
 
     // http server
     std::thread serverThread_;
@@ -344,6 +368,7 @@ private:
     struct PushSessionContext {
         std::mutex lock;
         std::string sessionId;
+        PushSessionContext(const std::string& id) : sessionId(id) {}
     };
     struct PermanentPut {
         time_point expiration;
@@ -353,9 +378,32 @@ private:
         std::unique_ptr<asio::steady_timer> expireTimer;
         std::unique_ptr<asio::steady_timer> expireNotifyTimer;
         Sp<Value> value;
+        PushType type;
+
+        template <typename Packer>
+        void msgpack_pack(Packer& p) const
+        {
+            p.pack_map(2 + (sessionCtx ? 1 : 0) + (clientId.empty() ? 0 : 1) + (type == PushType::None ? 0 : 2));
+            p.pack("value"); p.pack(value);
+            p.pack("exp"); p.pack(to_time_t(expiration));
+            if (not clientId.empty()) {
+                p.pack("cid"); p.pack(clientId);
+            }
+            if (sessionCtx) {
+                std::lock_guard<std::mutex> l(sessionCtx->lock);
+                p.pack("sid"); p.pack(sessionCtx->sessionId);
+            }
+            if (type != PushType::None) {
+                p.pack("t"); p.pack(type);
+                p.pack("token"); p.pack(pushToken);
+            }
+        }
+
+        void msgpack_unpack(const msgpack::object& o);
     };
     struct SearchPuts {
         std::map<dht::Value::Id, PermanentPut> puts;
+        MSGPACK_DEFINE_ARRAY(puts)
     };
     std::mutex lockSearchPuts_;
     std::map<InfoHash, SearchPuts> puts_;
@@ -367,14 +415,32 @@ private:
 
 #ifdef OPENDHT_PUSH_NOTIFICATIONS
     struct Listener {
+        time_point expiration;
         std::string clientId;
         std::shared_ptr<PushSessionContext> sessionCtx;
         std::future<size_t> internalToken;
         std::unique_ptr<asio::steady_timer> expireTimer;
         std::unique_ptr<asio::steady_timer> expireNotifyTimer;
+        PushType type;
+
+        template <typename Packer>
+        void msgpack_pack(Packer& p) const
+        {
+            p.pack_map(sessionCtx ? 4 : 3);
+            p.pack("cid"); p.pack(clientId);
+            p.pack("exp"); p.pack(to_time_t(expiration));
+            if (sessionCtx) {
+                std::lock_guard<std::mutex> l(sessionCtx->lock);
+                p.pack("sid"); p.pack(sessionCtx->sessionId);
+            }
+            p.pack("t"); p.pack(type);
+        }
+
+        void msgpack_unpack(const msgpack::object& o);
     };
     struct PushListener {
         std::map<InfoHash, std::vector<Listener>> listeners;
+        MSGPACK_DEFINE_ARRAY(listeners)
     };
     std::mutex lockPushListeners_;
     std::map<std::string, PushListener> pushListeners_;

include/opendht/network_engine.h

@@ -180,7 +180,8 @@ private:
             const InfoHash&,
             const Blob&,
             Tid,
-            const Query&)> onListen {};
+            const Query&,
+            int)> onListen {};
     /**
      * Called on announce request.
      *
@@ -237,9 +238,10 @@ public:
     void clear();
 
     /**
-     * Sends values (with closest nodes) to a listenner.
+     * Sends values (with closest nodes) to a listener.
      *
-     * @param sa          The address of the listenner.
+     * @deprecated
+     * @param sa          The address of the listener.
      * @param sslen       The length of the sockaddr structure.
      * @param socket_id  The tid to use to write to the request socket.
      * @param hash        The hash key of the value.
@@ -248,13 +250,14 @@ public:
      * @param nodes       The ipv4 closest nodes.
      * @param nodes6      The ipv6 closest nodes.
      * @param values      The values to send.
+     * @param version     If version = 1, a request will be used to answer to the listener
      */
     void tellListener(Sp<Node> n, Tid socket_id, const InfoHash& hash, want_t want, const Blob& ntoken,
             std::vector<Sp<Node>>&& nodes, std::vector<Sp<Node>>&& nodes6,
-            std::vector<Sp<Value>>&& values, const Query& q);
+            std::vector<Sp<Value>>&& values, const Query& q, int version);
 
-    void tellListenerRefreshed(Sp<Node> n, Tid socket_id, const InfoHash& hash, const Blob& ntoken, const std::vector<Value::Id>& values);
-    void tellListenerExpired(Sp<Node> n, Tid socket_id, const InfoHash& hash, const Blob& ntoken, const std::vector<Value::Id>& values);
+    void tellListenerRefreshed(Sp<Node> n, Tid socket_id, const InfoHash& hash, const Blob& ntoken, const std::vector<Value::Id>& values, int version);
+    void tellListenerExpired(Sp<Node> n, Tid socket_id, const InfoHash& hash, const Blob& ntoken, const std::vector<Value::Id>& values, int version);
 
     bool isRunning(sa_family_t af) const;
     inline want_t want () const { return dht_socket->hasIPv4() and dht_socket->hasIPv6() ? (WANT4 | WANT6) : -1; }
@@ -401,6 +404,24 @@ public:
                                  const Blob& token,
                                  RequestCb&& on_done,
                                  RequestExpiredCb&& on_expired);
+    /**
+     * Send a "update" request to a given node. Used for Listen operations
+     *
+     * @param n           The node.
+     * @param hash        The target hash.
+     * @param values      The values.
+     * @param created     Time id.
+     * @param token       A security token.
+     * @param sid         The socket id.
+     *
+     * @return the request with information concerning its success.
+     */
+    Sp<Request> sendUpdateValues(Sp<Node> n,
+                                 const InfoHash& infohash,
+                                 const std::vector<Sp<Value>>& values,
+                                 time_point created,
+                                 const Blob& token,
+                                 const size_t& sid);
 
     /**
      * Parses a message and calls appropriate callbacks.
@@ -490,19 +511,20 @@ private:
     void sendRequest(const Sp<Request>& request);
 
     struct MessageStats {
-        unsigned ping    {0};
-        unsigned find    {0};
-        unsigned get     {0};
-        unsigned put     {0};
-        unsigned listen  {0};
-        unsigned refresh {0};
+        unsigned ping         {0};
+        unsigned find         {0};
+        unsigned get          {0};
+        unsigned put          {0};
+        unsigned listen       {0};
+        unsigned refresh      {0};
+        unsigned updateValue  {0};
     };
 
 
     // basic wrapper for socket sendto function
     int send(const SockAddr& addr, const char *buf, size_t len, bool confirmed = false);
 
-    void sendValueParts(const TransId& tid, const std::vector<Blob>& svals, const SockAddr& addr);
+    void sendValueParts(Tid tid, const std::vector<Blob>& svals, const SockAddr& addr);
     std::vector<Blob> packValueHeader(msgpack::sbuffer&, const std::vector<Sp<Value>>&);
     void maintainRxBuffer(Tid tid);
 

include/opendht/utils.h

@@ -173,7 +173,13 @@ unpackMsg(Blob b) {
 
 msgpack::unpacked unpackMsg(Blob b);
 
-msgpack::object* findMapValue(const msgpack::object& map, const char* key);
-msgpack::object* findMapValue(const msgpack::object& map, const std::string& key);
+msgpack::object* findMapValue(const msgpack::object& map, const char* key, size_t length);
+
+inline msgpack::object* findMapValue(const msgpack::object& map, const char* key) {
+    return findMapValue(map, key, strlen(key));
+}
+inline msgpack::object* findMapValue(const msgpack::object& map, const std::string& key) {
+    return findMapValue(map, key.c_str(), key.size());
+}
 
 } // namespace dht

src/dht.cpp

@@ -1201,7 +1201,7 @@ Dht::storageChanged(const InfoHash& id, Storage& st, ValueStorage& v, bool newVa
                 vals.push_back(v.data);
                 Blob ntoken = makeToken(node_listeners.first->getAddr(), false);
                 network_engine.tellListener(node_listeners.first, l.first, id, 0, ntoken, {}, {},
-                        std::move(vals), l.second.query);
+                        std::move(vals), l.second.query, l.second.version);
             }
         }
     }
@@ -1259,7 +1259,7 @@ Dht::storageErase(const InfoHash& id, Value::Id vid)
 }
 
 void
-Dht::storageAddListener(const InfoHash& id, const Sp<Node>& node, size_t socket_id, Query&& query)
+Dht::storageAddListener(const InfoHash& id, const Sp<Node>& node, size_t socket_id, Query&& query, int version)
 {
     const auto& now = scheduler.time();
     auto st = store.find(id);
@@ -1275,9 +1275,9 @@ Dht::storageAddListener(const InfoHash& id, const Sp<Node>& node, size_t socket_
         if (not vals.empty()) {
             network_engine.tellListener(node, socket_id, id, WANT4 | WANT6, makeToken(node->getAddr(), false),
                     dht4.buckets.findClosestNodes(id, now, TARGET_NODES), dht6.buckets.findClosestNodes(id, now, TARGET_NODES),
-                    std::move(vals), query);
+                    std::move(vals), query, version);
         }
-        node_listeners.emplace(socket_id, Listener {now, std::forward<Query>(query)});
+        node_listeners.emplace(socket_id, Listener {now, std::forward<Query>(query), version});
     }
     else
         l->second.refresh(now, std::forward<Query>(query));
@@ -1312,7 +1312,7 @@ Dht::expireStore(decltype(store)::iterator i)
                             id.toString().c_str(),
                             node_listeners.first->toString().c_str());
                     Blob ntoken = makeToken(node_listeners.first->getAddr(), false);
-                    network_engine.tellListenerExpired(node_listeners.first, l.first, id, ntoken, ids);
+                    network_engine.tellListenerExpired(node_listeners.first, l.first, id, ntoken, ids, l.second.version);
                 }
             }
         }
@@ -1755,7 +1755,7 @@ Dht::Dht(std::unique_ptr<net::DatagramSocket>&& sock, const Config& config, cons
             std::bind(&Dht::onPing, this, _1),
             std::bind(&Dht::onFindNode, this, _1, _2, _3),
             std::bind(&Dht::onGetValues, this, _1, _2, _3, _4),
-            std::bind(&Dht::onListen, this, _1, _2, _3, _4, _5),
+            std::bind(&Dht::onListen, this, _1, _2, _3, _4, _5, _6),
             std::bind(&Dht::onAnnounce, this, _1, _2, _3, _4, _5),
             std::bind(&Dht::onRefresh, this, _1, _2, _3, _4)),
     persistPath(config.persist_path),
@@ -1982,14 +1982,19 @@ Dht::onDisconnected()
         return;
     if (logger_)
         logger_->d(myid, "Bootstraping");
-    for (const auto& boootstrap : bootstrap_nodes)
+    for (const auto& boootstrap : bootstrap_nodes) {
         try {
-            for (const auto& ip : network_engine.getSocket()->resolve(boootstrap.first, boootstrap.second))
+            auto ips = network_engine.getSocket()->resolve(boootstrap.first, boootstrap.second);
+            for (auto& ip : ips) {
+                if (ip.getPort() == 0)
+                    ip.setPort(net::DHT_DEFAULT_PORT);
                 pingNode(ip);
+            }
         } catch (const std::exception& e) {
             if (logger_)
                 logger_->e(myid, "Can't resolve %s:%s: %s", boootstrap.first.c_str(), boootstrap.second.c_str(), e.what());
         }
+    }
     if (bootstrapJob)
         bootstrapJob->cancel();
     bootstrapJob = scheduler.add(scheduler.time() + bootstrap_period, std::bind(&Dht::onDisconnected, this));
@@ -2320,7 +2325,7 @@ void Dht::onGetValuesDone(const Sp<Node>& node,
 }
 
 net::RequestAnswer
-Dht::onListen(Sp<Node> node, const InfoHash& hash, const Blob& token, size_t socket_id, const Query& query)
+Dht::onListen(Sp<Node> node, const InfoHash& hash, const Blob& token, size_t socket_id, const Query& query, int version)
 {
     if (not hash) {
         if (logger_)
@@ -2336,7 +2341,7 @@ Dht::onListen(Sp<Node> node, const InfoHash& hash, const Blob& token, size_t soc
         throw net::DhtProtocolException {net::DhtProtocolException::UNAUTHORIZED, net::DhtProtocolException::LISTEN_WRONG_TOKEN};
     }
     Query q = query;
-    storageAddListener(hash, node, socket_id, std::move(q));
+    storageAddListener(hash, node, socket_id, std::move(q), version);
     return {};
 }
 
@@ -2475,7 +2480,7 @@ Dht::storageRefresh(const InfoHash& id, Value::Id vid)
                             id.toString().c_str(),
                             node_listeners.first->toString().c_str());
                     Blob ntoken = makeToken(node_listeners.first->getAddr(), false);
-                    network_engine.tellListenerRefreshed(node_listeners.first, l.first, id, ntoken, ids);
+                    network_engine.tellListenerRefreshed(node_listeners.first, l.first, id, ntoken, ids, l.second.version);
                 }
             }
         }

src/dht_proxy_client.cpp

@@ -98,7 +98,7 @@ getRandomSessionId(size_t length = 8) {
     static constexpr const char chars[] = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789!#$%&()*+,./:;<=>?@[]^_`{|}~";
     std::string str(length, 0);
     crypto::random_device rdev;
-    std::uniform_int_distribution<> dist(0, (sizeof(chars)/sizeof(char)) - 1);
+    std::uniform_int_distribution<> dist(0, (sizeof(chars)/sizeof(char)) - 2);
     std::generate_n( str.begin(), length, [&]{ return chars[dist(rdev)]; } );
     return str;
 }

src/dht_proxy_server.cpp

@@ -34,6 +34,7 @@
 #include <functional>
 #include <limits>
 #include <iostream>
+#include <fstream>
 
 using namespace std::placeholders;
 using namespace std::chrono_literals;
@@ -173,26 +174,73 @@ struct DhtProxyServer::RestRouterTraits : public restinio::default_traits_t
     using connection_state_listener_t = ConnectionListener;
 };
 
-DhtProxyServer::DhtProxyServer(
-    crypto::Identity identity,
-    std::shared_ptr<DhtRunner> dht, in_port_t port, const std::string& pushServer,
-    std::shared_ptr<Logger> logger
+void
+DhtProxyServer::PermanentPut::msgpack_unpack(const msgpack::object& o)
+{
+    if (auto cid = findMapValue(o, "cid")) {
+        clientId = cid->as<std::string>();
+    }
+    if (auto exp = findMapValue(o, "exp")) {
+        expiration = from_time_t(exp->as<time_t>());
+    }
+    if (auto token = findMapValue(o, "token")) {
+        pushToken = token->as<std::string>();
+    }
+    if (auto sid = findMapValue(o, "sid")) {
+        if (not sessionCtx)
+            sessionCtx = std::make_shared<PushSessionContext>(sid->as<std::string>());
+        else
+            sessionCtx->sessionId = sid->as<std::string>();
+    }
+    if (auto t = findMapValue(o, "t")) {
+        type = t->as<PushType>();
+    }
+    if (auto val = findMapValue(o, "value")) {
+        value = std::make_shared<dht::Value>(*val);
+    }
+}
+
+#ifdef OPENDHT_PUSH_NOTIFICATIONS
+void
+DhtProxyServer::Listener::msgpack_unpack(const msgpack::object& o)
+{
+    if (auto cid = findMapValue(o, "cid")) {
+        clientId = cid->as<std::string>();
+    }
+    if (auto exp = findMapValue(o, "exp")) {
+        expiration = from_time_t(exp->as<time_t>());
+    }
+    if (auto sid = findMapValue(o, "sid")) {
+        if (not sessionCtx)
+            sessionCtx = std::make_shared<PushSessionContext>(sid->as<std::string>());
+        else
+            sessionCtx->sessionId = sid->as<std::string>();
+    }
+    if (auto t = findMapValue(o, "t")) {
+        type = t->as<PushType>();
+    }
+}
+#endif
+
+DhtProxyServer::DhtProxyServer(const std::shared_ptr<DhtRunner>& dht,
+        const ProxyServerConfig& config,
+        const std::shared_ptr<dht::Logger>& logger
 )
     :   ioContext_(std::make_shared<asio::io_context>()),
-        dht_(dht), logger_(logger),
+        dht_(dht), persistPath_(config.persistStatePath), logger_(logger),
         printStatsTimer_(std::make_unique<asio::steady_timer>(*ioContext_, 3s)),
         connListener_(std::make_shared<ConnectionListener>(std::bind(&DhtProxyServer::onConnectionClosed, this, std::placeholders::_1))),
-        pushServer_(pushServer)
+        pushServer_(config.pushServer)
 {
     if (not dht_)
         throw std::invalid_argument("A DHT instance must be provided");
 
     if (logger_)
-        logger_->d("[proxy:server] [init] running on %i", port);
-    if (not pushServer.empty()){
+        logger_->d("[proxy:server] [init] running on %i", config.port);
+    if (not pushServer_.empty()){
 #ifdef OPENDHT_PUSH_NOTIFICATIONS
         if (logger_)
-            logger_->d("[proxy:server] [init] using push server %s", pushServer.c_str());
+            logger_->d("[proxy:server] [init] using push server %s", pushServer_.c_str());
 #else
         if (logger_)
             logger_->e("[proxy:server] [init] opendht built without push notification support");
@@ -202,9 +250,9 @@ DhtProxyServer::DhtProxyServer(
     jsonBuilder_["commentStyle"] = "None";
     jsonBuilder_["indentation"] = "";
 
-    if (!pushServer.empty()){
+    if (!pushServer_.empty()){
         // no host delim, assume port only
-        if (pushServer.find(":") == std::string::npos)
+        if (pushServer_.find(":") == std::string::npos)
             pushServer_ =  "localhost:" + pushServer_;
         // define http request destination for push notifications
         pushHostPort_ = splitPort(pushServer_);
@@ -212,7 +260,7 @@ DhtProxyServer::DhtProxyServer(
             logger_->d("Using push server for notifications: %s:%s", pushHostPort_.first.c_str(),
                                                                      pushHostPort_.second.c_str());
     }
-    if (identity.first and identity.second) {
+    if (config.identity.first and config.identity.second) {
         asio::error_code ec;
         // define tls context
         asio::ssl::context tls_context { asio::ssl::context::sslv23 };
@@ -226,13 +274,13 @@ DhtProxyServer::DhtProxyServer(
         SSL_CTX_set_options(tls_context.native_handle(), SSL_OP_NO_RENEGOTIATION); // CVE-2009-3555
 #endif
         // node private key
-        auto key = identity.first->serialize();
+        auto key = config.identity.first->serialize();
         tls_context.use_private_key(asio::const_buffer{key.data(), key.size()},
                                     asio::ssl::context::file_format::pem, ec);
         if (ec)
             throw std::runtime_error("Error setting node's private key: " + ec.message());
         // certificate chain
-        auto certchain = identity.second->toString(true/*chain*/);
+        auto certchain = config.identity.second->toString(true/*chain*/);
         tls_context.use_certificate_chain(asio::const_buffer{certchain.data(), certchain.size()}, ec);
         if (ec)
             throw std::runtime_error("Error setting certificate chain: " + ec.message());
@@ -241,7 +289,7 @@ DhtProxyServer::DhtProxyServer(
         // build http server
         auto settings = restinio::run_on_this_thread_settings_t<RestRouterTraitsTls>();
         addServerSettings(settings);
-        settings.port(port);
+        settings.port(config.port);
         settings.tls_context(std::move(tls_context));
         httpsServer_ = std::make_unique<restinio::http_server_t<RestRouterTraitsTls>>(
             ioContext_,
@@ -258,7 +306,7 @@ DhtProxyServer::DhtProxyServer(
     else {
         auto settings = restinio::run_on_this_thread_settings_t<RestRouterTraits>();
         addServerSettings(settings);
-        settings.port(port);
+        settings.port(config.port);
         httpServer_ = std::make_unique<restinio::http_server_t<RestRouterTraits>>(
             ioContext_,
             std::forward<restinio::run_on_this_thread_settings_t<RestRouterTraits>>(std::move(settings))
@@ -274,8 +322,150 @@ DhtProxyServer::DhtProxyServer(
     dht->forwardAllMessages(true);
     updateStats();
     printStatsTimer_->async_wait(std::bind(&DhtProxyServer::handlePrintStats, this, std::placeholders::_1));
+
+    if (not persistPath_.empty()) {
+        try {
+            std::ifstream stateFile(persistPath_, std::ios::binary | std::ios::ate);
+            if (stateFile) {
+                std::streamsize size = stateFile.tellg();
+                stateFile.seekg(0, std::ios::beg);
+                if (logger_)
+                    logger_->d("Loading proxy state from %.*s (%td bytes)", (int)persistPath_.size(), persistPath_.c_str(), size);
+                loadState(stateFile, size);
+            }
+        } catch (const std::exception& e) {
+            if (logger_)
+                logger_->e("Error loading state from file: %s", e.what());
+        }
+    }
 }
 
+template <typename Os>
+void
+DhtProxyServer::saveState(Os& stream) {
+    msgpack::packer<Os> pk(&stream);
+    pk.pack_map(2);
+    {
+        std::lock_guard<std::mutex> lock(lockSearchPuts_);
+        pk.pack("puts");
+        pk.pack(puts_);
+    }
+    {
+        std::lock_guard<std::mutex> lock(lockListener_);
+        pk.pack("pushListeners");
+        pk.pack(pushListeners_);
+    }
+}
+
+template <typename Is>
+void
+DhtProxyServer::loadState(Is& is, size_t size) {
+    msgpack::unpacker pac;
+    pac.reserve_buffer(size);
+    if (is.read(pac.buffer(), size)) {
+        pac.buffer_consumed(size);
+
+        msgpack::object_handle oh;
+        while (pac.next(oh)) {
+            if (oh.get().type != msgpack::type::MAP)
+                continue;
+            if (auto puts = findMapValue(oh.get(), "puts")) {
+                std::lock_guard<std::mutex> lock(lockSearchPuts_);
+                puts_ = puts->as<decltype(puts_)>();
+                if (logger_)
+                    logger_->d("Loading %zu persistent puts", puts_.size());
+                for (auto& put : puts_) {
+                    for (auto& pput : put.second.puts) {
+                        pput.second.expireTimer = std::make_unique<asio::steady_timer>(io_context(), pput.second.expiration);
+                        pput.second.expireTimer->async_wait(std::bind(&DhtProxyServer::handleCancelPermamentPut, this,
+                                                std::placeholders::_1, put.first, pput.first));
+                        auto jsonProvider = [infoHash=put.first.toString(), clientId=pput.second.clientId, vid = pput.first, sessionCtx = pput.second.sessionCtx](){
+                            Json::Value json;
+                            json["timeout"] = infoHash;
+                            json["to"] = clientId;
+                            json["vid"] = std::to_string(vid);
+                            std::lock_guard<std::mutex> l(sessionCtx->lock);
+                            json["s"] = sessionCtx->sessionId;
+                            return json;
+                        };
+                        pput.second.expireNotifyTimer = std::make_unique<asio::steady_timer>(io_context(), pput.second.expiration - proxy::OP_MARGIN);
+                        pput.second.expireNotifyTimer->async_wait(std::bind(
+                            &DhtProxyServer::handleNotifyPushListenExpire, this,
+                            std::placeholders::_1, pput.second.pushToken, std::move(jsonProvider), pput.second.type));
+                        dht_->put(put.first, pput.second.value, DoneCallbackSimple{}, time_point::max(), true);
+                    }
+                }
+            } else {
+                if (logger_)
+                    logger_->d("No persistent puts in state");
+            }
+#ifdef OPENDHT_PUSH_NOTIFICATIONS
+            if (auto listeners = findMapValue(oh.get(), "pushListeners")) {
+                std::lock_guard<std::mutex> lock(lockListener_);
+                pushListeners_ = listeners->as<decltype(pushListeners_)>();
+                if (logger_)
+                    logger_->d("Loading %zu push listeners", pushListeners_.size());
+                for (auto& pushListener : pushListeners_) {
+                    for (auto& listeners : pushListener.second.listeners) {
+                        for (auto& listener : listeners.second) {
+                            listener.internalToken = dht_->listen(listeners.first,
+                                [this, infoHash=listeners.first, pushToken=pushListener.first, type=listener.type, clientId=listener.clientId, sessionCtx = listener.sessionCtx]
+                                (const std::vector<std::shared_ptr<Value>>& values, bool expired) {
+                                    // Build message content
+                                    Json::Value json;
+                                    json["key"] = infoHash.toString();
+                                    json["to"] = clientId;
+                                    json["t"] = Json::Value::Int64(std::chrono::duration_cast<std::chrono::milliseconds>(system_clock::now().time_since_epoch()).count());
+                                    {
+                                        std::lock_guard<std::mutex> l(sessionCtx->lock);
+                                        json["s"] = sessionCtx->sessionId;
+                                    }
+                                    if (expired and values.size() < 2){
+                                        std::stringstream ss;
+                                        for(size_t i = 0; i < values.size(); ++i){
+                                            if(i != 0) ss << ",";
+                                            ss << values[i]->id;
+                                        }
+                                        json["exp"] = ss.str();
+                                    }
+                                    auto maxPrio = 1000u;
+                                    for (const auto& v : values)
+                                        maxPrio = std::min(maxPrio, v->priority);
+                                    sendPushNotification(pushToken, std::move(json), type, !expired and maxPrio == 0);
+                                    return true;
+                                }
+                            );
+                            // expire notify
+                            listener.expireNotifyTimer = std::make_unique<asio::steady_timer>(io_context(), listener.expiration - proxy::OP_MARGIN);
+                            auto jsonProvider = [infoHash = listeners.first.toString(), clientId = listener.clientId, sessionCtx = listener.sessionCtx](){
+                                Json::Value json;
+                                json["timeout"] = infoHash;
+                                json["to"] = clientId;
+                                std::lock_guard<std::mutex> l(sessionCtx->lock);
+                                json["s"] = sessionCtx->sessionId;
+                                return json;
+                            };
+                            listener.expireNotifyTimer->async_wait(std::bind(&DhtProxyServer::handleNotifyPushListenExpire, this,
+                                                                std::placeholders::_1, pushListener.first, std::move(jsonProvider), listener.type));
+                            // cancel push listen
+                            listener.expireTimer = std::make_unique<asio::steady_timer>(io_context(), listener.expiration);
+                            listener.expireTimer->async_wait(std::bind(&DhtProxyServer::handleCancelPushListen, this,
+                                                            std::placeholders::_1, pushListener.first, listeners.first, listener.clientId));
+                        }
+                    }
+                }
+            } else {
+                if (logger_)
+                    logger_->d("No push listeners in state");
+            }
+#endif
+        }
+        if (logger_)
+            logger_->d("loading ended");
+    }
+}
+
+
 asio::io_context&
 DhtProxyServer::io_context() const
 {
@@ -284,6 +474,12 @@ DhtProxyServer::io_context() const
 
 DhtProxyServer::~DhtProxyServer()
 {
+    if (not persistPath_.empty()) {
+        if (logger_)
+            logger_->d("Saving proxy state to %.*s", (int)persistPath_.size(), persistPath_.c_str());
+        std::ofstream stateFile(persistPath_, std::ios::binary);
+        saveState(stateFile);
+    }
     if (dht_) {
         std::lock_guard<std::mutex> lock(lockListener_);
         for (auto& l : listeners_) {
@@ -585,12 +781,11 @@ DhtProxyServer::subscribe(restinio::request_handle_t request,
                           restinio::router::route_params_t params)
 {
     requestNum_++;
-
-    InfoHash infoHash(params["hash"].to_string());
-    if (!infoHash)
-        infoHash = InfoHash::get(params["hash"].to_string());
-
     try {
+        InfoHash infoHash(params["hash"].to_string());
+        if (!infoHash)
+            infoHash = InfoHash::get(params["hash"].to_string());
+
         std::string err;
         Json::Value r;
         auto* char_data = reinterpret_cast<const char*>(request->body().data());
@@ -607,117 +802,115 @@ DhtProxyServer::subscribe(restinio::request_handle_t request,
             response.set_body(RESP_MSG_NO_TOKEN);
             return response.done();
         }
-        auto isAndroid = root["platform"].asString() == "android";
+        auto type = root["platform"].asString() == "android" ? PushType::Android : PushType::iOS;
         auto clientId = root["client_id"].asString();
         auto sessionId = root["session_id"].asString();
 
         if (logger_)
             logger_->d("[proxy:server] [subscribe %s] [client %s] [session %s]", infoHash.toString().c_str(), clientId.c_str(), sessionId.c_str());
-        // ================ Search for existing listener ===================
-        // start the timer
-        auto timeout = std::chrono::steady_clock::now() + proxy::OP_TIMEOUT;
-        std::lock_guard<std::mutex> lock(lockPushListeners_);
 
         // Insert new or return existing push listeners of a token
+        std::lock_guard<std::mutex> lock(lockPushListeners_);
         auto& pushListener = pushListeners_[pushToken];
         auto& pushListeners = pushListener.listeners[infoHash];
 
-        for (auto &listener: pushListeners) {
-            // Found -> Resubscribe
-            if (listener.clientId == clientId) {
-                if (logger_)
-                    logger_->d("[proxy:server] [subscribe] found [client %s]", listener.clientId.c_str());
-                // Reset timers
-                listener.expireTimer->expires_at(timeout);
-                listener.expireNotifyTimer->expires_at(timeout - proxy::OP_MARGIN);
-                {
-                    std::lock_guard<std::mutex> l(listener.sessionCtx->lock);
-                    listener.sessionCtx->sessionId = sessionId;
-                }
-                // Send response header
-                auto response = std::make_shared<ResponseByPartsBuilder>(
-                    initHttpResponse(request->create_response<ResponseByParts>()));
-                response->flush();
-                if (!root["refresh"].asBool()) {
-                    // No Refresh
-                    dht_->get(infoHash, [this, response](const Sp<Value>& value){
-                        auto output = Json::writeString(jsonBuilder_, value->toJson()) + "\n";
-                        response->append_chunk(output);
-                        response->flush();
-                        return true;
-                    },
-                    [response] (bool){
-                        response->done();
-                    });
-                } else {
-                    // Refresh
-                    response->append_chunk("{}\n");
-                    response->done();
-                }
-                return restinio::request_handling_status_t::accepted;
-            }
+        auto listIt = std::find_if(pushListeners.begin(), pushListeners.end(), [&](const Listener& l) {
+            return l.clientId == clientId;
+        });
+        bool newListener = listIt == pushListeners.end();
+        if (newListener) {
+            pushListeners.emplace_back(Listener{});
+            listIt = std::prev(pushListeners.end());
+            listIt->clientId = clientId;
+            listIt->sessionCtx = std::make_shared<PushSessionContext>(sessionId);
+        } else {
+            std::lock_guard<std::mutex> l(listIt->sessionCtx->lock);
+            listIt->sessionCtx->sessionId = sessionId;
         }
-        // =========== No existing listener for an infoHash ============
-        // Add new listener to list of listeners
-        pushListeners.emplace_back(Listener{});
-        auto &listener = pushListeners.back();
-        listener.clientId = clientId;
-        listener.sessionCtx = std::make_shared<PushSessionContext>();
-        listener.sessionCtx->sessionId = sessionId;
+        auto& listener = *listIt;
 
-        // Add listen on dht
-        listener.internalToken = dht_->listen(infoHash,
-            [this, infoHash, pushToken, isAndroid, clientId, sessionCtx = listener.sessionCtx]
-            (const std::vector<std::shared_ptr<Value>>& values, bool expired){
-                // Build message content
-                Json::Value json;
-                json["key"] = infoHash.toString();
-                json["to"] = clientId;
-                json["t"] = Json::Value::Int64(std::chrono::duration_cast<std::chrono::milliseconds>(system_clock::now().time_since_epoch()).count());
-                {
-                    std::lock_guard<std::mutex> l(sessionCtx->lock);
-                    json["s"] = sessionCtx->sessionId;
-                }
-                if (expired and values.size() < 2){
-                    std::stringstream ss;
-                    for(size_t i = 0; i < values.size(); ++i){
-                        if(i != 0) ss << ",";
-                        ss << values[i]->id;
-                    }
-                    json["exp"] = ss.str();
-                }
-                auto maxPrio = 1000u;
-                for (const auto& v : values)
-                    maxPrio = std::min(maxPrio, v->priority);
-                sendPushNotification(pushToken, std::move(json), isAndroid, !expired and maxPrio == 0);
-                return true;
-            }
-        );
-        // expire notify
-        if (!listener.expireNotifyTimer)
-            listener.expireNotifyTimer = std::make_unique<asio::steady_timer>(io_context(), timeout - proxy::OP_MARGIN);
-        else
+        // Expiration
+        auto timeout = std::chrono::steady_clock::now() + proxy::OP_TIMEOUT;
+        listener.expiration = timeout;
+        listener.type = type;
+        if (listener.expireNotifyTimer)
             listener.expireNotifyTimer->expires_at(timeout - proxy::OP_MARGIN);
-        auto jsonProvider = [infoHash, clientId, sessionCtx = listener.sessionCtx](){
+        else
+            listener.expireNotifyTimer = std::make_unique<asio::steady_timer>(io_context(), timeout - proxy::OP_MARGIN);
+        auto jsonProvider = [h=infoHash.toString(), clientId, sessionCtx = listener.sessionCtx](){
             Json::Value json;
-            json["timeout"] = infoHash.toString();
+            json["timeout"] = h;
             json["to"] = clientId;
             std::lock_guard<std::mutex> l(sessionCtx->lock);
             json["s"] = sessionCtx->sessionId;
             return json;
         };
         listener.expireNotifyTimer->async_wait(std::bind(&DhtProxyServer::handleNotifyPushListenExpire, this,
-                                               std::placeholders::_1, pushToken, std::move(jsonProvider), isAndroid));
-        // cancel push listen
+                                               std::placeholders::_1, pushToken, std::move(jsonProvider), listener.type));
         if (!listener.expireTimer)
             listener.expireTimer = std::make_unique<asio::steady_timer>(io_context(), timeout);
         else
             listener.expireTimer->expires_at(timeout);
         listener.expireTimer->async_wait(std::bind(&DhtProxyServer::handleCancelPushListen, this,
-                                         std::placeholders::_1, pushToken, infoHash, clientId));
-        auto response = initHttpResponse(request->create_response());
-        response.set_body("{}\n");
-        return response.done();
+                                        std::placeholders::_1, pushToken, infoHash, clientId));
+
+        // Send response
+        if (not newListener) {
+            if (logger_)
+                logger_->d("[proxy:server] [subscribe] found [client %s]", listener.clientId.c_str());
+            // Send response header
+            auto response = std::make_shared<ResponseByPartsBuilder>(initHttpResponse(request->create_response<ResponseByParts>()));
+            response->flush();
+            if (!root["refresh"].asBool()) {
+                // No Refresh
+                dht_->get(infoHash, [this, response](const Sp<Value>& value){
+                    auto output = Json::writeString(jsonBuilder_, value->toJson()) + "\n";
+                    response->append_chunk(output);
+                    response->flush();
+                    return true;
+                },
+                [response] (bool){
+                    response->done();
+                });
+            } else {
+                // Refresh
+                response->append_chunk("{}\n");
+                return response->done();
+            }
+        } else {
+            // =========== No existing listener for an infoHash ============
+            // Add listen on dht
+            listener.internalToken = dht_->listen(infoHash,
+                [this, infoHash, pushToken, type, clientId, sessionCtx = listener.sessionCtx]
+                (const std::vector<std::shared_ptr<Value>>& values, bool expired){
+                    // Build message content
+                    Json::Value json;
+                    json["key"] = infoHash.toString();
+                    json["to"] = clientId;
+                    json["t"] = Json::Value::Int64(std::chrono::duration_cast<std::chrono::milliseconds>(system_clock::now().time_since_epoch()).count());
+                    {
+                        std::lock_guard<std::mutex> l(sessionCtx->lock);
+                        json["s"] = sessionCtx->sessionId;
+                    }
+                    if (expired and values.size() < 2){
+                        std::stringstream ss;
+                        for(size_t i = 0; i < values.size(); ++i){
+                            if(i != 0) ss << ",";
+                            ss << values[i]->id;
+                        }
+                        json["exp"] = ss.str();
+                    }
+                    auto maxPrio = 1000u;
+                    for (const auto& v : values)
+                        maxPrio = std::min(maxPrio, v->priority);
+                    sendPushNotification(pushToken, std::move(json), type, !expired and maxPrio == 0);
+                    return true;
+                }
+            );
+            auto response = initHttpResponse(request->create_response());
+            response.set_body("{}\n");
+            return response.done();
+        }
     }
     catch (...) {
         return serverError(*request);
@@ -766,7 +959,7 @@ DhtProxyServer::unsubscribe(restinio::request_handle_t request,
 
 void
 DhtProxyServer::handleNotifyPushListenExpire(const asio::error_code &ec, const std::string pushToken,
-                                             std::function<Json::Value()> jsonProvider, const bool isAndroid)
+                                             std::function<Json::Value()> jsonProvider, PushType type)
 {
     if (ec == asio::error::operation_aborted)
         return;
@@ -776,7 +969,7 @@ DhtProxyServer::handleNotifyPushListenExpire(const asio::error_code &ec, const s
     }
     if (logger_)
         logger_->d("[proxy:server] [subscribe] sending put refresh to %s token", pushToken.c_str());
-    sendPushNotification(pushToken, jsonProvider(), isAndroid, false);
+    sendPushNotification(pushToken, jsonProvider(), type, false);
 }
 
 void
@@ -818,7 +1011,7 @@ DhtProxyServer::handleCancelPushListen(const asio::error_code &ec, const std::st
 }
 
 void
-DhtProxyServer::sendPushNotification(const std::string& token, Json::Value&& json, bool isAndroid, bool highPriority)
+DhtProxyServer::sendPushNotification(const std::string& token, Json::Value&& json, PushType type, bool highPriority)
 {
     if (pushServer_.empty())
         return;
@@ -840,7 +1033,7 @@ DhtProxyServer::sendPushNotification(const std::string& token, Json::Value&& jso
         Json::Value tokens(Json::arrayValue);
         tokens[0] = token;
         notification["tokens"] = std::move(tokens);
-        notification["platform"] = isAndroid ? 2 : 1;
+        notification["platform"] = type == PushType::Android ? 2 : 1;
         notification["data"] = std::move(json);
         notification["priority"] = highPriority ? "high" : "normal";
         notification["time_to_live"] = 600;
@@ -889,6 +1082,7 @@ DhtProxyServer::handleCancelPermamentPut(const asio::error_code &ec, const InfoH
     }
     if (logger_)
         logger_->d("[proxy:server] [put %s] cancel permament put %i", key.toString().c_str(), vid);
+    std::lock_guard<std::mutex> lock(lockSearchPuts_);
     auto sPuts = puts_.find(key);
     if (sPuts == puts_.end())
         return;
@@ -898,6 +1092,8 @@ DhtProxyServer::handleCancelPermamentPut(const asio::error_code &ec, const InfoH
         return;
     if (dht_)
         dht_->cancelPut(key, vid);
+    if (put->second.expireTimer)
+        put->second.expireTimer->cancel();
     if (put->second.expireNotifyTimer)
         put->second.expireNotifyTimer->cancel();
     sPutsMap.erase(put);
@@ -941,7 +1137,7 @@ DhtProxyServer::put(restinio::request_handle_t request,
                     platform = pVal["platform"].asString();
                     sessionId = pVal["session_id"].asString();
                 }
-                std::unique_lock<std::mutex> lock(lockSearchPuts_);
+                std::lock_guard<std::mutex> lock(lockSearchPuts_);
                 auto timeout = std::chrono::steady_clock::now() + proxy::OP_TIMEOUT;
                 auto& sPuts = puts_[infoHash];
                 if (value->id == Value::INVALID_ID) {
@@ -955,32 +1151,36 @@ DhtProxyServer::put(restinio::request_handle_t request,
                                                         std::placeholders::_1, infoHash, pp.second.value->id));
                             if (not sessionId.empty()) {
                                 if (not pp.second.sessionCtx)
-                                    pp.second.sessionCtx = std::make_shared<PushSessionContext>();
-                                std::lock_guard<std::mutex> l(pp.second.sessionCtx->lock);
-                                pp.second.sessionCtx->sessionId = sessionId;
+                                    pp.second.sessionCtx = std::make_shared<PushSessionContext>(sessionId);
+                                else {
+                                    std::lock_guard<std::mutex> l(pp.second.sessionCtx->lock);
+                                    pp.second.sessionCtx->sessionId = sessionId;
+                                }
                             }
                             auto response = initHttpResponse(request->create_response());
                             response.append_body(Json::writeString(jsonBuilder_, value->toJson()) + "\n");
                             return response.done();
                         }
                     }
+                    value->id = std::uniform_int_distribution<Value::Id>{1}(rd);
                 }
 
                 auto vid = value->id;
                 auto& pput = sPuts.puts[vid];
                 pput.value = value;
+                pput.expiration = timeout;
                 if (not pput.expireTimer) {
                     auto &ctx = io_context();
                     // cancel permanent put
                     pput.expireTimer = std::make_unique<asio::steady_timer>(ctx, timeout);
 #ifdef OPENDHT_PUSH_NOTIFICATIONS
                     if (not pushToken.empty()){
+                        bool isAndroid = platform == "android";
                         pput.pushToken = pushToken;
                         pput.clientId = clientId;
-                        pput.sessionCtx = std::make_shared<PushSessionContext>();
-                        pput.sessionCtx->sessionId = sessionId;
+                        pput.type = isAndroid ? PushType::Android : PushType::iOS;
+                        pput.sessionCtx = std::make_shared<PushSessionContext>(sessionId);
                         // notify push listen expire
-                        bool isAndroid = platform == "android";
                         auto jsonProvider = [infoHash, clientId, vid, sessionCtx = pput.sessionCtx](){
                             Json::Value json;
                             json["timeout"] = infoHash.toString();
@@ -997,15 +1197,17 @@ DhtProxyServer::put(restinio::request_handle_t request,
                             pput.expireNotifyTimer->expires_at(timeout - proxy::OP_MARGIN);
                         pput.expireNotifyTimer->async_wait(std::bind(
                             &DhtProxyServer::handleNotifyPushListenExpire, this,
-                            std::placeholders::_1, pushToken, std::move(jsonProvider), isAndroid));
+                            std::placeholders::_1, pushToken, std::move(jsonProvider), pput.type));
                     }
 #endif
                 } else {
                     if (not sessionId.empty()) {
                         if (not pput.sessionCtx)
-                            pput.sessionCtx = std::make_shared<PushSessionContext>();
-                        std::lock_guard<std::mutex> l(pput.sessionCtx->lock);
-                        pput.sessionCtx->sessionId = sessionId;
+                            pput.sessionCtx = std::make_shared<PushSessionContext>(sessionId);
+                        else {
+                            std::lock_guard<std::mutex> l(pput.sessionCtx->lock);
+                            pput.sessionCtx->sessionId = sessionId;
+                        }
                     }
                     pput.expireTimer->expires_at(timeout);
                     if (pput.expireNotifyTimer)

src/dhtrunner.cpp

@@ -475,6 +475,10 @@ DhtRunner::getNodeInfo() const {
         info.node_id = dht->getNodeId();
         info.ipv4 = dht->getNodesStats(AF_INET);
         info.ipv6 = dht->getNodesStats(AF_INET6);
+        if (auto sock = dht->getSocket()) {
+            info.bound4 = sock->getBoundRef(AF_INET).getPort();
+            info.bound6 = sock->getBoundRef(AF_INET6).getPort();
+        }
     }
     info.ongoing_ops = ongoing_ops;
     return info;
@@ -492,6 +496,10 @@ DhtRunner::getNodeInfo(std::function<void(std::shared_ptr<NodeInfo>)> cb)
         info.node_id = dht.getNodeId();
         info.ipv4 = dht.getNodesStats(AF_INET);
         info.ipv6 = dht.getNodesStats(AF_INET6);
+        if (auto sock = dht.getSocket()) {
+            info.bound4 = sock->getBoundRef(AF_INET).getPort();
+            info.bound6 = sock->getBoundRef(AF_INET6).getPort();
+        }
         info.ongoing_ops = ongoing_ops;
         cb(std::move(sinfo));
         opEnded();

src/listener.h

@@ -30,8 +30,9 @@ namespace dht {
 struct Listener {
     time_point time;
     Query query;
+    int version;
 
-    Listener(time_point t, Query&& q) : time(t), query(std::move(q)) {}
+    Listener(time_point t, Query&& q, int version = 0) : time(t), query(std::move(q)), version(version) {}
 
     void refresh(time_point t, Query&& q) {
         time = t;

src/net.h

@@ -31,7 +31,8 @@ enum class MessageType {
     Refresh,
     Listen,
     ValueData,
-    ValueUpdate
+    ValueUpdate,
+    UpdateValue
 };
 
 } /* namespace net */

src/network_engine.cpp

@@ -48,23 +48,6 @@ static constexpr uint8_t v4prefix[16] = {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xFF, 0xF
 constexpr unsigned SEND_NODES {8};
 
 
-/* Transaction-ids are 4-bytes long, with the first two bytes identifying
- * the kind of request, and the remaining two a sequence number in
- * host order.
- */
-struct TransId final : public std::array<uint8_t, 4> {
-    TransId() { std::fill(begin(), end(), 0); }
-    TransId(const std::array<char, 4>& o) { std::copy(o.begin(), o.end(), begin()); }
-    TransId(uint32_t id) {
-        *reinterpret_cast<uint32_t*>(data()) = htonl(id);
-    }
-
-    uint32_t toInt() const {
-        return ntohl(*reinterpret_cast<const uint32_t*>(&(*this)[0]));
-    }
-};
-
-
 struct NetworkEngine::PartialMessage {
     SockAddr from;
     time_point start;
@@ -140,11 +123,15 @@ void
 NetworkEngine::tellListener(Sp<Node> node, Tid socket_id, const InfoHash& hash, want_t want,
         const Blob& ntoken, std::vector<Sp<Node>>&& nodes,
         std::vector<Sp<Node>>&& nodes6, std::vector<Sp<Value>>&& values,
-        const Query& query)
+        const Query& query, int version)
 {
     auto nnodes = bufferNodes(node->getFamily(), hash, want, nodes, nodes6);
     try {
-        sendNodesValues(node->getAddr(), socket_id, nnodes.first, nnodes.second, values, query, ntoken);
+        if (version >= 1) {
+            sendUpdateValues(node, hash, values, scheduler.time(), ntoken, socket_id);
+        } else {
+            sendNodesValues(node->getAddr(), socket_id, nnodes.first, nnodes.second, values, query, ntoken);
+        }
     } catch (const std::overflow_error& e) {
         if (logger_)
             logger_->e("Can't send value: buffer not large enough !");
@@ -152,11 +139,12 @@ NetworkEngine::tellListener(Sp<Node> node, Tid socket_id, const InfoHash& hash,
 }
 
 void
-NetworkEngine::tellListenerRefreshed(Sp<Node> n, Tid socket_id, const InfoHash&, const Blob& token, const std::vector<Value::Id>& values)
+NetworkEngine::tellListenerRefreshed(Sp<Node> n, Tid socket_id, const InfoHash&, const Blob& token, const std::vector<Value::Id>& values, int version)
 {
     msgpack::sbuffer buffer;
     msgpack::packer<msgpack::sbuffer> pk(&buffer);
-    pk.pack_map(4+(config.network?1:0));
+
+    pk.pack_map(4 + (version >= 1 ? 2 : 0) + (config.network?1:0));
 
     pk.pack(KEY_U);
     pk.pack_map(1 + (not values.empty()?1:0) + (not token.empty()?1:0));
@@ -178,16 +166,34 @@ NetworkEngine::tellListenerRefreshed(Sp<Node> n, Tid socket_id, const InfoHash&,
         pk.pack(KEY_NETID); pk.pack(config.network);
     }
 
+    if (version >= 1) {
+        Tid tid (n->getNewTid());
+        Tid sid (socket_id);
+
+        pk.pack(KEY_REQ_SID);   pk.pack(sid);
+        pk.pack(KEY_TID); pk.pack(tid);
+
+        auto req = std::make_shared<Request>(MessageType::UpdateValue, tid, n,
+            Blob(buffer.data(), buffer.data() + buffer.size()),
+            [=](const Request&, ParsedMessage&&) { /* on done */ },
+            [=](const Request&, bool) { /* on expired */ }
+        );
+        sendRequest(req);
+        ++out_stats.updateValue;
+        return;
+    }
+
     // send response
     send(n->getAddr(), buffer.data(), buffer.size());
 }
 
 void
-NetworkEngine::tellListenerExpired(Sp<Node> n, Tid socket_id, const InfoHash&, const Blob& token, const std::vector<Value::Id>& values)
+NetworkEngine::tellListenerExpired(Sp<Node> n, Tid socket_id, const InfoHash&, const Blob& token, const std::vector<Value::Id>& values, int version)
 {
     msgpack::sbuffer buffer;
     msgpack::packer<msgpack::sbuffer> pk(&buffer);
-    pk.pack_map(4+(config.network?1:0));
+
+    pk.pack_map(4 + (version >= 1 ? 2 : 0) + (config.network?1:0));
 
     pk.pack(KEY_U);
     pk.pack_map(1 + (not values.empty()?1:0) + (not token.empty()?1:0));
@@ -209,6 +215,23 @@ NetworkEngine::tellListenerExpired(Sp<Node> n, Tid socket_id, const InfoHash&, c
         pk.pack(KEY_NETID); pk.pack(config.network);
     }
 
+    if (version >= 1) {
+        Tid tid (n->getNewTid());
+        Tid sid (socket_id);
+
+        pk.pack(KEY_REQ_SID);   pk.pack(sid);
+        pk.pack(KEY_TID); pk.pack(tid);
+
+        auto req = std::make_shared<Request>(MessageType::UpdateValue, tid, n,
+            Blob(buffer.data(), buffer.data() + buffer.size()),
+            [=](const Request&, ParsedMessage&&) { /* on done */ },
+            [=](const Request&, bool) { /* on expired */ }
+        );
+        sendRequest(req);
+        ++out_stats.updateValue;
+        return;
+    }
+
     // send response
     send(n->getAddr(), buffer.data(), buffer.size());
 }
@@ -617,18 +640,28 @@ NetworkEngine::process(std::unique_ptr<ParsedMessage>&& msg, const SockAddr& fro
                 if (logIncoming_ and logger_)
                     logger_->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));
+                RequestAnswer answer = onListen(node, msg->info_hash, msg->token, msg->socket_id, std::move(msg->query), msg->version);
                 auto nnodes = bufferNodes(from.getFamily(), msg->info_hash, msg->want, answer.nodes4, answer.nodes6);
                 sendListenConfirmation(from, msg->tid);
                 break;
             }
+            case MessageType::UpdateValue: {
+                if (logIncoming_ and logger_)
+                    logger_->d(msg->info_hash, node->id, "[node %s] got 'update' request for %s", node->toString().c_str(), msg->info_hash.toString().c_str());
+                ++in_stats.updateValue;
+                if (auto rsocket = node->getSocket(msg->socket_id))
+                    rsocket->on_receive(node, std::move(*msg));
+                else if (logger_)
+                    logger_->e(msg->info_hash, node->id, "[node %s] 'update' request without socket for %s", node->toString().c_str(), msg->info_hash.toString().c_str());
+                break;
+            }
             default:
                 break;
             }
         } catch (const std::overflow_error& e) {
             if (logger_)
                 logger_->e("Can't send value: buffer not large enough !");
-        } catch (DhtProtocolException& e) {
+        } catch (const DhtProtocolException& e) {
             sendError(from, msg->tid, e.getCode(), e.getMsg().c_str(), true);
         }
     }
@@ -654,7 +687,7 @@ NetworkEngine::send(const SockAddr& addr, const char *buf, size_t len, bool conf
 
 Sp<Request>
 NetworkEngine::sendPing(Sp<Node> node, RequestCb&& on_done, RequestExpiredCb&& on_expired) {
-    TransId tid (node->getNewTid());
+    Tid tid (node->getNewTid());
     msgpack::sbuffer buffer;
     msgpack::packer<msgpack::sbuffer> pk(&buffer);
     pk.pack_map(5+(config.network?1:0));
@@ -663,15 +696,14 @@ NetworkEngine::sendPing(Sp<Node> node, RequestCb&& on_done, RequestExpiredCb&& o
      pk.pack(KEY_REQ_ID); pk.pack(myid);
 
     pk.pack(KEY_Q); pk.pack(QUERY_PING);
-    pk.pack(KEY_TID); pk.pack_bin(tid.size());
-                              pk.pack_bin_body((const char*)tid.data(), tid.size());
+    pk.pack(KEY_TID); pk.pack(tid);
     pk.pack(KEY_Y); pk.pack(KEY_Q);
     pk.pack(KEY_UA); pk.pack(my_v);
     if (config.network) {
         pk.pack(KEY_NETID); pk.pack(config.network);
     }
 
-    auto req = std::make_shared<Request>(MessageType::Ping, tid.toInt(), node,
+    auto req = std::make_shared<Request>(MessageType::Ping, tid, node,
         Blob(buffer.data(), buffer.data() + buffer.size()),
         [=](const Request& req_status, ParsedMessage&&) {
             if (logger_)
@@ -701,9 +733,7 @@ NetworkEngine::sendPong(const SockAddr& addr, Tid tid) {
       pk.pack(KEY_REQ_ID); pk.pack(myid);
       insertAddr(pk, addr);
 
-    TransId t (tid);
-    pk.pack(KEY_TID); pk.pack_bin(t.size());
-                               pk.pack_bin_body((const char*)t.data(), t.size());
+    pk.pack(KEY_TID); pk.pack(tid);
     pk.pack(KEY_Y); pk.pack(KEY_R);
     pk.pack(KEY_UA); pk.pack(my_v);
     if (config.network) {
@@ -716,7 +746,7 @@ NetworkEngine::sendPong(const SockAddr& addr, Tid tid) {
 Sp<Request>
 NetworkEngine::sendFindNode(Sp<Node> n, const InfoHash& target, want_t want,
         RequestCb&& on_done, RequestExpiredCb&& on_expired) {
-    TransId tid (n->getNewTid());
+    Tid tid (n->getNewTid());
     msgpack::sbuffer buffer;
     msgpack::packer<msgpack::sbuffer> pk(&buffer);
     pk.pack_map(5+(config.network?1:0));
@@ -732,15 +762,14 @@ NetworkEngine::sendFindNode(Sp<Node> n, const InfoHash& target, want_t want,
     }
 
     pk.pack(KEY_Q); pk.pack(QUERY_FIND);
-    pk.pack(KEY_TID); pk.pack_bin(tid.size());
-                               pk.pack_bin_body((const char*)tid.data(), tid.size());
+    pk.pack(KEY_TID); pk.pack(tid);
     pk.pack(KEY_Y); pk.pack(KEY_Q);
     pk.pack(KEY_UA); pk.pack(my_v);
     if (config.network) {
         pk.pack(KEY_NETID); pk.pack(config.network);
     }
 
-    auto req = std::make_shared<Request>(MessageType::FindNode, tid.toInt(), n,
+    auto req = std::make_shared<Request>(MessageType::FindNode, tid, n,
         Blob(buffer.data(), buffer.data() + buffer.size()),
         [=](const Request& req_status, ParsedMessage&& msg) { /* on done */
             if (on_done) {
@@ -762,7 +791,7 @@ NetworkEngine::sendFindNode(Sp<Node> n, const InfoHash& target, want_t want,
 Sp<Request>
 NetworkEngine::sendGetValues(Sp<Node> n, const InfoHash& info_hash, const Query& query, want_t want,
         RequestCb&& on_done, RequestExpiredCb&& on_expired) {
-    TransId tid (n->getNewTid());
+    Tid tid (n->getNewTid());
     msgpack::sbuffer buffer;
     msgpack::packer<msgpack::sbuffer> pk(&buffer);
     pk.pack_map(5+(config.network?1:0));
@@ -786,15 +815,14 @@ NetworkEngine::sendGetValues(Sp<Node> n, const InfoHash& info_hash, const Query&
       }
 
     pk.pack(KEY_Q); pk.pack(QUERY_GET);
-    pk.pack(KEY_TID); pk.pack_bin(tid.size());
-                      pk.pack_bin_body((const char*)tid.data(), tid.size());
+    pk.pack(KEY_TID); pk.pack(tid);
     pk.pack(KEY_Y); pk.pack(KEY_Q);
     pk.pack(KEY_UA); pk.pack(my_v);
     if (config.network) {
         pk.pack(KEY_NETID); pk.pack(config.network);
     }
 
-    auto req = std::make_shared<Request>(MessageType::GetValues, tid.toInt(), n,
+    auto req = std::make_shared<Request>(MessageType::GetValues, tid, n,
         Blob(buffer.data(), buffer.data() + buffer.size()),
         [=](const Request& req_status, ParsedMessage&& msg) { /* on done */
             if (on_done) {
@@ -896,7 +924,7 @@ NetworkEngine::packValueHeader(msgpack::sbuffer& buffer, const std::vector<Sp<Va
 }
 
 void
-NetworkEngine::sendValueParts(const TransId& tid, const std::vector<Blob>& svals, const SockAddr& addr)
+NetworkEngine::sendValueParts(Tid tid, const std::vector<Blob>& svals, const SockAddr& addr)
 {
     msgpack::sbuffer buffer;
     unsigned i=0;
@@ -911,8 +939,7 @@ NetworkEngine::sendValueParts(const TransId& tid, const std::vector<Blob>& svals
                 pk.pack(KEY_NETID); pk.pack(config.network);
             }
             pk.pack(KEY_Y); pk.pack(KEY_V);
-            pk.pack(KEY_TID); pk.pack_bin(tid.size());
-                                       pk.pack_bin_body((const char*)tid.data(), tid.size());
+            pk.pack(KEY_TID); pk.pack(tid);
             pk.pack(KEY_V); pk.pack_map(1);
                 pk.pack(i); pk.pack_map(2);
                     pk.pack(std::string("o")); pk.pack(start);
@@ -967,9 +994,7 @@ NetworkEngine::sendNodesValues(const SockAddr& addr, Tid tid, const Blob& nodes,
         }
     }
 
-    TransId t (tid);
-    pk.pack(KEY_TID); pk.pack_bin(t.size());
-                      pk.pack_bin_body((const char*)t.data(), t.size());
+    pk.pack(KEY_TID); pk.pack(tid);
     pk.pack(KEY_Y); pk.pack(KEY_R);
     pk.pack(KEY_UA); pk.pack(my_v);
     if (config.network) {
@@ -1045,7 +1070,7 @@ NetworkEngine::sendListen(Sp<Node> n,
         SocketCb&& socket_cb)
 {
     Tid socket;
-    TransId tid (n->getNewTid());
+    Tid tid (n->getNewTid());
     if (previous and previous->node == n) {
         socket = previous->getSocket();
     } else {
@@ -1060,33 +1085,30 @@ NetworkEngine::sendListen(Sp<Node> n,
             logger_->e(hash, "[node %s] unable to get a valid socket for listen. Aborting listen", n->toString().c_str());
         return {};
     }
-    TransId sid(socket);
-
     msgpack::sbuffer buffer;
     msgpack::packer<msgpack::sbuffer> pk(&buffer);
     pk.pack_map(5+(config.network?1:0));
 
     auto has_query = not query.where.empty() or not query.select.empty();
-    pk.pack(KEY_A); pk.pack_map(4 + has_query);
+    pk.pack(KEY_A); pk.pack_map(5 + has_query);
       pk.pack(KEY_REQ_ID);    pk.pack(myid);
+      pk.pack(KEY_VERSION);   pk.pack(1);
       pk.pack(KEY_REQ_H);     pk.pack(hash);
       pk.pack(KEY_REQ_TOKEN); packToken(pk, token);
-      pk.pack(KEY_REQ_SID);   pk.pack_bin(sid.size());
-                              pk.pack_bin_body((const char*)sid.data(), sid.size());
+      pk.pack(KEY_REQ_SID);   pk.pack(socket);
       if (has_query) {
           pk.pack(KEY_REQ_QUERY); pk.pack(query);
       }
 
     pk.pack(KEY_Q); pk.pack(QUERY_LISTEN);
-    pk.pack(KEY_TID); pk.pack_bin(tid.size());
-                               pk.pack_bin_body((const char*)tid.data(), tid.size());
+    pk.pack(KEY_TID); pk.pack(tid);
     pk.pack(KEY_Y); pk.pack(KEY_Q);
     pk.pack(KEY_UA); pk.pack(my_v);
     if (config.network) {
         pk.pack(KEY_NETID); pk.pack(config.network);
     }
 
-    auto req = std::make_shared<Request>(MessageType::Listen, tid.toInt(), n,
+    auto req = std::make_shared<Request>(MessageType::Listen, tid, n,
         Blob(buffer.data(), buffer.data() + buffer.size()),
         [=](const Request& req_status, ParsedMessage&& msg) { /* on done */
             if (on_done)
@@ -1113,9 +1135,7 @@ NetworkEngine::sendListenConfirmation(const SockAddr& addr, Tid tid) {
       pk.pack(KEY_REQ_ID); pk.pack(myid);
       insertAddr(pk, addr);
 
-    TransId t (tid);
-    pk.pack(KEY_TID); pk.pack_bin(t.size());
-                               pk.pack_bin_body((const char*)t.data(), t.size());
+    pk.pack(KEY_TID); pk.pack(tid);
     pk.pack(KEY_Y); pk.pack(KEY_R);
     pk.pack(KEY_UA); pk.pack(my_v);
     if (config.network) {
@@ -1134,7 +1154,7 @@ NetworkEngine::sendAnnounceValue(Sp<Node> n,
         RequestCb&& on_done,
         RequestExpiredCb&& on_expired)
 {
-    TransId tid (n->getNewTid());
+    Tid tid (n->getNewTid());
     msgpack::sbuffer buffer;
     msgpack::packer<msgpack::sbuffer> pk(&buffer);
     pk.pack_map(5+(config.network?1:0));
@@ -1150,15 +1170,14 @@ NetworkEngine::sendAnnounceValue(Sp<Node> n,
       pk.pack(KEY_REQ_TOKEN);  pk.pack(token);
 
     pk.pack(KEY_Q);   pk.pack(QUERY_PUT);
-    pk.pack(KEY_TID); pk.pack_bin(tid.size());
-                      pk.pack_bin_body((const char*)tid.data(), tid.size());
+    pk.pack(KEY_TID); pk.pack(tid);
     pk.pack(KEY_Y);   pk.pack(KEY_Q);
     pk.pack(KEY_UA);  pk.pack(my_v);
     if (config.network) {
         pk.pack(KEY_NETID); pk.pack(config.network);
     }
 
-    auto req = std::make_shared<Request>(MessageType::AnnounceValue, tid.toInt(), n,
+    auto req = std::make_shared<Request>(MessageType::AnnounceValue, tid, n,
         Blob(buffer.data(), buffer.data() + buffer.size()),
         [=](const Request& req_status, ParsedMessage&& msg) { /* on done */
             if (msg.value_id == Value::INVALID_ID) {
@@ -1184,6 +1203,52 @@ NetworkEngine::sendAnnounceValue(Sp<Node> n,
     return req;
 }
 
+Sp<Request>
+NetworkEngine::sendUpdateValues(Sp<Node> n,
+                                const InfoHash& infohash,
+                                const std::vector<Sp<Value>>& values,
+                                time_point created,
+                                const Blob& token,
+                                const size_t& socket_id)
+{
+    Tid tid (n->getNewTid());
+    Tid sid (socket_id);
+
+    msgpack::sbuffer buffer;
+    msgpack::packer<msgpack::sbuffer> pk(&buffer);
+    pk.pack_map(5+(config.network?1:0));
+
+    pk.pack(KEY_A); pk.pack_map((created < scheduler.time() ? 7 : 6));
+      pk.pack(KEY_REQ_ID);     pk.pack(myid);
+      pk.pack(KEY_VERSION);    pk.pack(1);
+      pk.pack(KEY_REQ_H);      pk.pack(infohash);
+      pk.pack(KEY_REQ_SID);   pk.pack(sid);
+      auto v = packValueHeader(buffer, values);
+      if (created < scheduler.time()) {
+          pk.pack(KEY_REQ_CREATION);
+          pk.pack(to_time_t(created));
+      }
+      pk.pack(KEY_REQ_TOKEN);  pk.pack(token);
+
+    pk.pack(KEY_Q);   pk.pack(QUERY_UPDATE);
+    pk.pack(KEY_TID); pk.pack(tid);
+    pk.pack(KEY_Y);   pk.pack(KEY_Q);
+    pk.pack(KEY_UA);  pk.pack(my_v);
+    if (config.network) {
+        pk.pack(KEY_NETID); pk.pack(config.network);
+    }
+
+    auto req = std::make_shared<Request>(MessageType::UpdateValue, tid, n,
+        Blob(buffer.data(), buffer.data() + buffer.size()),
+        [=](const Request&, ParsedMessage&&) { /* on done */ },
+        [=](const Request&, bool) { /* on expired */ }
+    );
+    req->parts = std::move(v);
+    sendRequest(req);
+    ++out_stats.updateValue;
+    return req;
+}
+
 Sp<Request>
 NetworkEngine::sendRefreshValue(Sp<Node> n,
                 const InfoHash& infohash,
@@ -1192,7 +1257,7 @@ NetworkEngine::sendRefreshValue(Sp<Node> n,
                 RequestCb&& on_done,
                 RequestExpiredCb&& on_expired)
 {
-    TransId tid (n->getNewTid());
+    Tid tid (n->getNewTid());
     msgpack::sbuffer buffer;
     msgpack::packer<msgpack::sbuffer> pk(&buffer);
     pk.pack_map(5+(config.network?1:0));
@@ -1204,15 +1269,14 @@ NetworkEngine::sendRefreshValue(Sp<Node> n,
       pk.pack(KEY_REQ_TOKEN);    pk.pack(token);
 
     pk.pack(KEY_Q); pk.pack(QUERY_REFRESH);
-    pk.pack(KEY_TID); pk.pack_bin(tid.size());
-                               pk.pack_bin_body((const char*)tid.data(), tid.size());
+    pk.pack(KEY_TID); pk.pack(tid);
     pk.pack(KEY_Y); pk.pack(KEY_Q);
     pk.pack(KEY_UA); pk.pack(my_v);
     if (config.network) {
         pk.pack(KEY_NETID); pk.pack(config.network);
     }
 
-    auto req = std::make_shared<Request>(MessageType::Refresh, tid.toInt(), n,
+    auto req = std::make_shared<Request>(MessageType::Refresh, tid, n,
         Blob(buffer.data(), buffer.data() + buffer.size()),
         [=](const Request& req_status, ParsedMessage&& msg) { /* on done */
             if (msg.value_id == Value::INVALID_ID) {
@@ -1248,9 +1312,7 @@ NetworkEngine::sendValueAnnounced(const SockAddr& addr, Tid tid, Value::Id vid)
       pk.pack(KEY_REQ_VALUE_ID); pk.pack(vid);
       insertAddr(pk, addr);
 
-    TransId t(tid);
-    pk.pack(KEY_TID); pk.pack_bin(t.size());
-                               pk.pack_bin_body((const char*)t.data(), t.size());
+    pk.pack(KEY_TID); pk.pack(tid);
     pk.pack(KEY_Y); pk.pack(KEY_R);
     pk.pack(KEY_UA); pk.pack(my_v);
     if (config.network) {
@@ -1280,9 +1342,7 @@ NetworkEngine::sendError(const SockAddr& addr,
           pk.pack(KEY_REQ_ID); pk.pack(myid);
     }
 
-    TransId t(tid);
-    pk.pack(KEY_TID); pk.pack_bin(t.size());
-                               pk.pack_bin_body((const char*)t.data(), t.size());
+    pk.pack(KEY_TID); pk.pack(tid);
     pk.pack(KEY_Y); pk.pack(KEY_E);
     pk.pack(KEY_UA); pk.pack(my_v);
     if (config.network) {

src/parsed_message.h

@@ -54,10 +54,12 @@ static const std::string KEY_REQ_EXPIRED {"exp"};
 static const std::string KEY_REQ_REFRESHED {"re"};
 static const std::string KEY_REQ_FIELDS {"fileds"};
 static const std::string KEY_REQ_WANT {"w"};
+static const std::string KEY_VERSION {"ve"};
 
 static const std::string QUERY_PING {"ping"};
 static const std::string QUERY_FIND {"find"};
 static const std::string QUERY_GET {"get"};
+static const std::string QUERY_UPDATE {"update"};
 static const std::string QUERY_PUT {"put"};
 static const std::string QUERY_LISTEN {"listen"};
 static const std::string QUERY_REFRESH {"refresh"};
@@ -113,6 +115,7 @@ struct ParsedMessage {
     uint16_t error_code;
     /* reported address by the distant node */
     std::string ua;
+    int version {0};
     SockAddr addr;
     void msgpack_unpack(const msgpack::object& o);
 
@@ -224,6 +227,8 @@ ParsedMessage::msgpack_unpack(const msgpack::object& msg)
         type = MessageType::AnnounceValue;
     else if (parsed.q == QUERY_REFRESH)
         type = MessageType::Refresh;
+    else if (parsed.q == QUERY_UPDATE)
+        type = MessageType::UpdateValue;
     else
         throw msgpack::type_error();
 
@@ -295,6 +300,8 @@ ParsedMessage::msgpack_unpack(const msgpack::object& msg)
             parsedReq.fields = &o.val;
         else if (key == KEY_REQ_WANT)
             parsedReq.want = &o.val;
+        else if (key == KEY_VERSION)
+            version = o.val.as<int>();
     }
 
     if (parsedReq.sa) {

src/utils.cpp

@@ -277,28 +277,16 @@ unpackMsg(Blob b) {
 }
 
 msgpack::object*
-findMapValue(const msgpack::object& map, const char* key) {
+findMapValue(const msgpack::object& map, const char* key, size_t key_length) {
     if (map.type != msgpack::type::MAP) throw msgpack::type_error();
     for (unsigned i = 0; i < map.via.map.size; i++) {
         auto& o = map.via.map.ptr[i];
         if (o.key.type == msgpack::type::STR
+            && key_length == o.key.via.str.size
             && std::strncmp(o.key.via.str.ptr, key, o.key.via.str.size) == 0)
             return &o.val;
     }
     return nullptr;
 }
 
-msgpack::object*
-findMapValue(const msgpack::object& map, const std::string& key) {
-    if (map.type != msgpack::type::MAP) throw msgpack::type_error();
-    for (unsigned i = 0; i < map.via.map.size; i++) {
-        auto& o = map.via.map.ptr[i];
-        if (o.key.type == msgpack::type::STR
-            && key.size() == o.key.via.str.size
-            && std::strncmp(o.key.via.str.ptr, key.data(), o.key.via.str.size) == 0)
-            return &o.val;
-    }
-    return nullptr;
-}
-
 }

tests/dhtproxytester.cpp

@@ -44,12 +44,12 @@ DhtProxyTester::setUp() {
     nodeProxy->bootstrap(nodePeer.getBound());
 
     auto serverCAIdentity = dht::crypto::generateEcIdentity("DHT Node CA");
-    auto serverIdentity = dht::crypto::generateIdentity("DHT Node", serverCAIdentity);
 
-    serverProxy = std::make_unique<dht::DhtProxyServer>(
-            //dht::crypto::Identity{}, // http
-            serverIdentity,            // https
-            nodeProxy, 8080, /*pushServer*/"127.0.0.1:8090");
+    dht::ProxyServerConfig serverConfig;
+    serverConfig.identity = dht::crypto::generateIdentity("DHT Node", serverCAIdentity);
+    serverConfig.port = 8080;
+    serverConfig.pushServer = "127.0.0.1:8090";
+    serverProxy = std::make_unique<dht::DhtProxyServer>(nodeProxy, serverConfig);
 
     clientConfig.server_ca = serverCAIdentity.second;
     clientConfig.client_identity = dht::crypto::generateIdentity("DhtProxyTester");

tests/dhtrunnertester.cpp

@@ -167,4 +167,51 @@ DhtRunnerTester::testListen() {
     node1.cancelListen(d, tokend);
 }
 
+void
+DhtRunnerTester::testListenLotOfBytes() {
+    std::mutex mutex;
+    std::condition_variable cv;
+    std::atomic_uint valueCount(0);
+    unsigned putCount(0);
+    unsigned putOkCount(0);
+
+    std::string data(10000, 'a');
+
+    auto foo = dht::InfoHash::get("foo");
+    constexpr unsigned N = 50;
+
+    for (unsigned i=0; i<N; i++) {
+        node2.put(foo, data, [&](bool ok) {
+            std::lock_guard<std::mutex> lock(mutex);
+            putCount++;
+            if (ok) putOkCount++;
+            cv.notify_all();
+        });
+    }
+    {
+        std::unique_lock<std::mutex> lk(mutex);
+        CPPUNIT_ASSERT(cv.wait_for(lk, 30s, [&]{ return putCount == N; }));
+    }
+
+    dht::DhtRunner node3 {};
+    dht::DhtRunner::Config config;
+    config.dht_config.node_config.max_peer_req_per_sec = -1;
+    config.dht_config.node_config.max_req_per_sec = -1;
+    node3.run(42242, config);
+    node3.bootstrap(node1.getBound());
+
+    auto ftokenfoo = node3.listen(foo, [&](const std::shared_ptr<dht::Value>&) {
+        valueCount++;
+        cv.notify_all();
+        return true;
+    });
+
+    {
+        std::unique_lock<std::mutex> lk(mutex);
+        CPPUNIT_ASSERT(cv.wait_for(lk, 30s, [&]{ return valueCount == N; }));
+    }
+
+    node3.cancelListen(foo, ftokenfoo.get());
+}
+
 }  // namespace test

tests/dhtrunnertester.h

@@ -32,6 +32,7 @@ class DhtRunnerTester : public CppUnit::TestFixture {
     CPPUNIT_TEST(testConstructors);
     CPPUNIT_TEST(testGetPut);
     CPPUNIT_TEST(testListen);
+    CPPUNIT_TEST(testListenLotOfBytes);
     CPPUNIT_TEST_SUITE_END();
 
     dht::DhtRunner node1 {};
@@ -57,6 +58,10 @@ class DhtRunnerTester : public CppUnit::TestFixture {
      * Test listen method
      */
     void testListen();
+    /**
+     * Test listen method with lot of datas
+     */
+    void testListenLotOfBytes();
 };
 
 }  // namespace test

tests/httptester.cpp

@@ -39,9 +39,10 @@ HttpTester::setUp() {
     nodeProxy->run(0, /*identity*/{}, /*threaded*/true);
     nodeProxy->bootstrap(nodePeer->getBound());
 
-    serverProxy = std::unique_ptr<dht::DhtProxyServer>(
-        new dht::DhtProxyServer(
-            /*http*/dht::crypto::Identity{}, nodeProxy, 8080, /*pushServer*/"127.0.0.1:8090"));
+    dht::ProxyServerConfig config;
+    config.port = 8080;
+    config.pushServer = "127.0.0.1:8090";
+    serverProxy = std::make_unique<dht::DhtProxyServer>(nodeProxy, config);
 }
 
 void

tools/dhtchat.cpp

@@ -64,10 +64,10 @@ main(int argc, char **argv)
         auto dhtConf = getDhtConfig(params);
         dht.run(params.port, dhtConf.first, std::move(dhtConf.second));
 
-        if (not params.bootstrap.first.empty())
-            dht.bootstrap(params.bootstrap.first.c_str(), params.bootstrap.second.c_str());
+        if (not params.bootstrap.empty())
+            dht.bootstrap(params.bootstrap);
 
-        print_node_info(dht, params);
+        print_node_info(dht.getNodeInfo());
         std::cout << "  type 'c {hash}' to join a channel" << std::endl << std::endl;
 
         bool connected {false};

tools/dhtnode.cpp

@@ -114,7 +114,7 @@ void cmd_loop(std::shared_ptr<DhtRunner>& node, dht_params& params
 #endif
 )
 {
-    print_node_info(*node, params);
+    print_node_info(node->getNodeInfo());
     std::cout << " (type 'h' or 'help' for a list of possible commands)" << std::endl << std::endl;
 
 #ifndef WIN32_NATIVE
@@ -143,22 +143,23 @@ void cmd_loop(std::shared_ptr<DhtRunner>& node, dht_params& params
             print_help();
             continue;
         } else if (op == "ll") {
-            auto nodeInfo = node->getNodeInfo();
-            print_node_info(*node, params);
-            std::cout << nodeInfo.ongoing_ops << " ongoing operations" << std::endl;
-            std::cout << "IPv4 stats:" << std::endl;
-            std::cout << nodeInfo.ipv4.toString() << std::endl;
-            std::cout << "IPv6 stats:" << std::endl;
-            std::cout << nodeInfo.ipv6.toString() << std::endl;
+            node->getNodeInfo([&](const std::shared_ptr<dht::NodeInfo>& nodeInfo) {
+                print_node_info(*nodeInfo);
+                std::cout << nodeInfo->ongoing_ops << " ongoing operations" << std::endl;
+                std::cout << "IPv4 stats:" << std::endl;
+                std::cout << nodeInfo->ipv4.toString() << std::endl;
+                std::cout << "IPv6 stats:" << std::endl;
+                std::cout << nodeInfo->ipv6.toString() << std::endl;
 #ifdef OPENDHT_PROXY_SERVER
-            for (const auto& proxy : proxies) {
-                std::cout << "Stats for proxy server on port " << proxy.first << std::endl;
-                if (auto stats = proxy.second->stats())
-                    std::cout << "  " << stats->toString() << std::endl;
-                else
-                    std::cout << "  (stats not available yet)" << std::endl;                
-            }
+                for (const auto& proxy : proxies) {
+                    std::cout << "Stats for proxy server on port " << proxy.first << std::endl;
+                    if (auto stats = proxy.second->stats())
+                        std::cout << "  " << stats->toString() << std::endl;
+                    else
+                        std::cout << "  (stats not available yet)" << std::endl;                
+                }
 #endif
+            });
             continue;
         } else if (op == "lr") {
             std::cout << "IPv4 routing table:" << std::endl;
@@ -190,14 +191,10 @@ void cmd_loop(std::shared_ptr<DhtRunner>& node, dht_params& params
             continue;
         } else if (op == "b") {
             iss >> idstr;
-            try {
-                auto addr = splitPort(idstr);
-                if (not addr.first.empty() and addr.second.empty())
-                    addr.second = std::to_string(dht::net::DHT_DEFAULT_PORT);
-                node->bootstrap(addr.first.c_str(), addr.second.c_str());
-            } catch (const std::exception& e) {
-                std::cerr << e.what() << std::endl;
-            }
+            if (not idstr.empty())
+                node->bootstrap(idstr);
+            else
+                std::cerr << "No service provided. Syntax: b hostname:port" << std::endl;
             continue;
         } else if (op == "log") {
             iss >> idstr;
@@ -221,14 +218,11 @@ void cmd_loop(std::shared_ptr<DhtRunner>& node, dht_params& params
                 iss >> idstr;
 #endif // OPENDHT_PUSH_NOTIFICATIONS
             try {
-                unsigned int port = std::stoi(idstr);
-                proxies.emplace(port, std::unique_ptr<DhtProxyServer>(
-                    new DhtProxyServer(
-                        dht::crypto::Identity{}, node, port
-#ifdef OPENDHT_PUSH_NOTIFICATIONS
-                        ,pushServer
-#endif
-                )));
+                in_port_t port = std::stoi(idstr);
+                ProxyServerConfig serverConfig;
+                serverConfig.port = port;
+                serverConfig.pushServer = pushServer;
+                proxies.emplace(port, std::make_unique<DhtProxyServer>(node, serverConfig));
             } catch (...) { }
             continue;
         } else if (op == "psx") {
@@ -239,13 +233,12 @@ void cmd_loop(std::shared_ptr<DhtRunner>& node, dht_params& params
 #endif // OPENDHT_PUSH_NOTIFICATIONS
             try {
                 if (params.proxy_id.first and params.proxy_id.second){
-                    unsigned int port = std::stoi(idstr);
-                    proxies.emplace(port, std::unique_ptr<DhtProxyServer>(
-                        new DhtProxyServer(params.proxy_id, node, port
-#ifdef OPENDHT_PUSH_NOTIFICATIONS
-                                           ,pushServer
-#endif
-                    )));
+                    in_port_t port = std::stoi(idstr);
+                    ProxyServerConfig serverConfig;
+                    serverConfig.identity = params.proxy_id;
+                    serverConfig.port = port;
+                    serverConfig.pushServer = pushServer;
+                    proxies.emplace(port, std::make_unique<DhtProxyServer>(node, serverConfig));
                 }
                 else {
                     std::cerr << "Missing Identity private key or certificate" << std::endl;
@@ -537,25 +530,32 @@ main(int argc, char **argv)
         auto dhtConf = getDhtConfig(params);
         node->run(params.port, dhtConf.first, std::move(dhtConf.second));
 
-        if (not params.bootstrap.first.empty()) {
-            std::cout << "Bootstrap: " << params.bootstrap.first << ":" << params.bootstrap.second << std::endl;
-            node->bootstrap(params.bootstrap.first.c_str(), params.bootstrap.second.c_str());
+        if (not params.bootstrap.empty()) {
+            std::cout << "Bootstrap: " << params.bootstrap << std::endl;
+            node->bootstrap(params.bootstrap);
         }
 
 #ifdef OPENDHT_PROXY_SERVER
         std::map<in_port_t, std::unique_ptr<DhtProxyServer>> proxies;
 #endif
-        if (params.proxyserverssl and params.proxy_id.first and params.proxy_id.second){
+        if (params.proxyserver or params.proxyserverssl) {
 #ifdef OPENDHT_PROXY_SERVER
-            proxies.emplace(params.proxyserverssl, std::unique_ptr<DhtProxyServer>(
-                new DhtProxyServer(params.proxy_id,
-                                   node, params.proxyserverssl, params.pushserver, dhtConf.second.logger)));
-        }
-        if (params.proxyserver) {
-            proxies.emplace(params.proxyserver, std::unique_ptr<DhtProxyServer>(
-                new DhtProxyServer(
-                    dht::crypto::Identity{},
-                    node, params.proxyserver, params.pushserver, dhtConf.second.logger)));
+            ProxyServerConfig serverConfig;
+            serverConfig.pushServer = params.pushserver;
+            if (params.proxyserverssl and params.proxy_id.first and params.proxy_id.second){
+                serverConfig.identity = params.proxy_id;
+                serverConfig.port = params.proxyserverssl;
+                if (not params.persist_path.empty())
+                    serverConfig.persistStatePath = params.persist_path + '_' + std::to_string(serverConfig.port);
+                proxies.emplace(params.proxyserverssl, std::make_unique<DhtProxyServer>(node, serverConfig, dhtConf.second.logger));
+            }
+            if (params.proxyserver) {
+                serverConfig.identity = {};
+                serverConfig.port = params.proxyserver;
+                if (not params.persist_path.empty())
+                    serverConfig.persistStatePath = params.persist_path + '_' + std::to_string(serverConfig.port);
+                proxies.emplace(params.proxyserver, std::make_unique<DhtProxyServer>(node, serverConfig, dhtConf.second.logger));
+            }
 #else
             std::cerr << "DHT proxy server requested but OpenDHT built without proxy server support." << std::endl;
             exit(EXIT_FAILURE);

tools/dhtproxy_stats.py

@@ -48,10 +48,10 @@ for i in range(80,101):
         
         stats['timestamp'] = str(ts)
 
-        with open("stats_proxy_" + str(i), "a") as stat_file:
-            stat_file.write(str(stats))
-            stat_file.write('\n')
+        #with open("stats_proxy_" + str(i), "a") as stat_file:
+        #    stat_file.write(str(stats))
+        #    stat_file.write('\n')
 
-with open("stats_proxy_total", "a") as stat_file:
+with open("stats_proxy_total", "w") as stat_file:
     stat_file.write(str(stats_total))
     stat_file.write('\n')

tools/dhtscanner.cpp

@@ -91,10 +91,10 @@ main(int argc, char **argv)
         auto dhtConf = getDhtConfig(params);
         dht.run(params.port, dhtConf.first, std::move(dhtConf.second));
 
-        if (not params.bootstrap.first.empty())
-            dht.bootstrap(params.bootstrap.first.c_str(), params.bootstrap.second.c_str());
+        if (not params.bootstrap.empty())
+            dht.bootstrap(params.bootstrap);
 
-        print_node_info(dht, params);
+        print_node_info(dht.getNodeInfo());
         std::cout << "Scanning network..." << std::endl;
         auto all_nodes = std::make_shared<NodeSet>();
 

tools/tools_common.h

@@ -124,7 +124,7 @@ struct dht_params {
     bool log {false};
     bool syslog {false};
     std::string logfile {};
-    std::pair<std::string, std::string> bootstrap {};
+    std::string bootstrap {};
     dht::NetId network {0};
     in_port_t port {0};
     in_port_t proxyserver {0};
@@ -191,16 +191,14 @@ getDhtConfig(dht_params& params)
     return {std::move(config), std::move(context)};
 }
 
-void print_node_info(const dht::DhtRunner& node, const dht_params& params) {
-    std::cout << "OpenDHT node " << node.getNodeId() << " running on ";
-    auto port4 = node.getBoundPort(AF_INET);
-    auto port6 = node.getBoundPort(AF_INET6);
-    if (port4 == port6)
-        std::cout << "port " << port4 << std::endl;
+void print_node_info(const dht::NodeInfo& info) {
+    std::cout << "OpenDHT node " << info.node_id << " running on ";
+    if (info.bound4 == info.bound6)
+        std::cout << "port " << info.bound4 << std::endl;
     else
-        std::cout << "IPv4 port " << port4 << ", IPv6 port " << port6 << std::endl;
-    if (params.id.first)
-        std::cout << "Public key ID " << node.getId() << std::endl;
+        std::cout << "IPv4 port " << info.bound4 << ", IPv6 port " << info.bound6 << std::endl;
+    if (info.id)
+        std::cout << "Public key ID " << info.id << std::endl;
 }
 
 static const constexpr struct option long_options[] = {
@@ -291,10 +289,7 @@ parseArgs(int argc, char **argv) {
             params.public_stable = true;
             break;
         case 'b':
-            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::net::DHT_DEFAULT_PORT);
-            }
+            params.bootstrap = (optarg[0] == '=') ? optarg+1 : optarg;
             break;
         case 'V':
             params.version = true;