proxy server: add missing listener type

The rest is not needed.

CMakeLists.txt

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

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

@@ -256,8 +256,8 @@ public:
             std::vector<Sp<Node>>&& nodes, std::vector<Sp<Node>>&& nodes6,
             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; }
@@ -516,8 +516,8 @@ private:
         unsigned get          {0};
         unsigned put          {0};
         unsigned listen       {0};
-        unsigned updateValue  {0};
         unsigned refresh      {0};
+        unsigned updateValue  {0};
     };
 
 

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

@@ -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);
                 }
             }
         }
@@ -2480,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_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/network_engine.cpp

@@ -127,7 +127,7 @@ NetworkEngine::tellListener(Sp<Node> node, Tid socket_id, const InfoHash& hash,
 {
     auto nnodes = bufferNodes(node->getFamily(), hash, want, nodes, nodes6);
     try {
-        if (version == 1) {
+        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);
@@ -139,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));
@@ -165,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));
@@ -196,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());
 }

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/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/dhtnode.cpp

@@ -218,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") {
@@ -236,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;
@@ -542,17 +538,24 @@ main(int argc, char **argv)
 #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')