search: expire only after candidate expiration

move bootstrap logic from DhtRunner to Dht
2026-11-09 04:49:14 +00:00 · 2020-04-24 13:29:12 -04:00 · 2020-04-24 13:29:12 -04:00
38 changed files with 605 additions and 1042 deletions
@@ -16,12 +16,13 @@ env:

 before_install:
  - |
-    if [[ "$OPENDHT_TEST_JOB" == *"opendht.classic"* ]]; then
-      docker pull aberaud/opendht-deps-bionic;
-    fi
-    if [[ "$OPENDHT_TEST_JOB" == *"opendht.proxyclient"* ]] || [[ "$OPENDHT_TEST_JOB" == *"opendht.proxyserver"* ]] || [[ "$OPENDHT_TEST_JOB" == *"opendht.push"* ]]; then
+    # non llvm builds
+    if [[ "$OPENDHT_TEST_JOB" != *"opendht.llvm"* ]]; then
      docker pull aberaud/opendht-deps;
    fi
+
+  - |
+    # classic build
    if [[ "$OPENDHT_TEST_JOB" == *"opendht.llvm"* ]]; then
      docker pull aberaud/opendht-deps-llvm
    fi
@@ -4,7 +4,7 @@ include(CMakeDependentOption)

 project (opendht)
 set (opendht_VERSION_MAJOR 2)
-set (opendht_VERSION_MINOR 1.3)
+set (opendht_VERSION_MINOR 0.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 REQUIRED)
+find_package (PkgConfig)
 find_package (GnuTLS 3.3 REQUIRED)
 pkg_search_module (Nettle nettle)
 find_package (Msgpack 1.2 REQUIRED)
@@ -45,6 +45,7 @@ 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} "\")")
@@ -299,9 +300,9 @@ if (OPENDHT_STATIC)
    endif ()
    target_link_libraries(opendht-static
        PRIVATE  ${argon2_LIBRARIES}
-        PUBLIC ${CMAKE_THREAD_LIBS_INIT} ${GNUTLS_LIBRARIES} ${Nettle_STATIC_LIBRARIES}
-               ${Jsoncpp_STATIC_LIBRARIES} ${FMT_LIBRARY} ${HTTP_PARSER_LIBRARY}
-               ${OPENSSL_STATIC_LIBRARIES})
+        PUBLIC ${CMAKE_THREAD_LIBS_INIT} ${GNUTLS_LIBRARIES} ${Nettle_LIBRARIES}
+               ${Jsoncpp_LIBRARIES} ${FMT_LIBRARY} ${HTTP_PARSER_LIBRARY}
+               ${OPENSSL_LIBRARIES})
    install (TARGETS opendht-static DESTINATION ${CMAKE_INSTALL_LIBDIR} EXPORT opendht)
 endif ()

@@ -379,7 +380,7 @@ install (FILES ${CMAKE_CURRENT_BINARY_DIR}/opendhtConfigVersion.cmake DESTINATIO

 # Unit tests
 if (OPENDHT_TESTS)
-    pkg_search_module(Cppunit REQUIRED cppunit)
+    FIND_PACKAGE(Cppunit REQUIRED)
    # unit testing
    list (APPEND test_FILES
        tests/infohashtester.h
@@ -411,8 +412,7 @@ if (OPENDHT_TESTS)
        tests/tests_runner.cpp
        ${test_FILES}
    )
-    target_include_directories(opendht_unit_tests SYSTEM PRIVATE ${Cppunit_INCLUDE_DIR})
-    target_link_directories(opendht_unit_tests PRIVATE ${Cppunit_LIBRARY_DIRS})
+    target_include_directories(opendht_unit_tests SYSTEM PRIVATE ${CPPUNIT_INCLUDE_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}
    )
@@ -0,0 +1,28 @@
+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)
@@ -1,7 +1,7 @@
 dnl define macros
 m4_define([opendht_major_version], 2)
-m4_define([opendht_minor_version], 1)
-m4_define([opendht_patch_version], 3)
+m4_define([opendht_minor_version], 0)
+m4_define([opendht_patch_version], 0)
 m4_define([opendht_version],
 		  [opendht_major_version.opendht_minor_version.opendht_patch_version])

@@ -1,20 +1,33 @@
-FROM ubuntu:20.04
+FROM ubuntu:16.04
 MAINTAINER Adrien Béraud <adrien.beraud@savoirfairelinux.com>
-
 RUN apt-get update && apt-get install -y \
-        dialog apt-utils \
-    && apt-get clean \
-    && echo 'debconf debconf/frontend select Noninteractive' | debconf-set-selections
-
-RUN apt-get update && apt-get install -y \
-        build-essential pkg-config cmake git wget \
-        autotools-dev autoconf \
-        cython3 python3-dev python3-setuptools \
-        libncurses5-dev libreadline-dev nettle-dev libcppunit-dev \
-        libgnutls28-dev libuv1-dev libjsoncpp-dev libargon2-dev \
-        libssl-dev libfmt-dev libhttp-parser-dev libasio-dev libmsgpack-dev \
+        build-essential cmake git wget libncurses5-dev libreadline-dev nettle-dev \
+        libgnutls28-dev libuv1-dev cython3 python3-dev python3-setuptools libcppunit-dev libjsoncpp-dev \
+        autotools-dev autoconf libssl-dev \
    && apt-get clean

+RUN echo "*** Installing libfmt ***" \
+    && wget https://github.com/fmtlib/fmt/archive/5.3.0.tar.gz \
+    && tar -xzf 5.3.0.tar.gz && cd fmt-5.3.0/ \
+    && cmake -DCMAKE_INSTALL_PREFIX=/usr -DBUILD_SHARED_LIBS=On -DFMT_TEST=Off -DFMT_DOC=Off . \
+    && make -j8 && make install \
+    && cd ../ && rm -rf fmt*
+
+# libhttp-parser-dev is 2.1-2 which is too old
+RUN echo "*** Building nodejs/http_parser dependency ***" \
+    && wget https://github.com/nodejs/http-parser/archive/v2.9.2.tar.gz \
+    && tar xvf v2.9.2.tar.gz && cd http-parser-2.9.2/ \
+    && PREFIX=/usr make -j8 install \
+    && cd ../ && rm -rf http-parser*
+
+# libasio-dev (1.10.6-3) is too old
+RUN echo "** Building a recent version of asio ***" \
+    && wget https://github.com/aberaud/asio/archive/a7d66ef4017d8f1b7f2cef1bb4ba8e23b0961571.tar.gz \
+    && tar -xvf a7d66ef4017d8f1b7f2cef1bb4ba8e23b0961571.tar.gz && cd asio-a7d66ef4017d8f1b7f2cef1bb4ba8e23b0961571/asio \
+    && ./autogen.sh && ./configure --prefix=/usr --without-boost --disable-examples --disable-tests  \
+    && make install \
+    && cd ../../ && rm -rf asio*
+
 RUN echo "*** Downloading RESTinio ***" \
    && mkdir restinio && cd restinio \
    && wget https://github.com/aberaud/restinio/archive/a7a10e419d9089c5b8ee63f5e3098c892f22fae4.tar.gz \
@@ -25,3 +38,19 @@ RUN echo "*** Downloading RESTinio ***" \
             -DRESTINIO_FIND_DEPS=ON -DRESTINIO_ALLOW_SOBJECTIZER=Off -DRESTINIO_USE_BOOST_ASIO=none . \
    && make -j8 && make install \
    && cd ../../ && rm -rf restinio*
+
+
+# build restbed from sources
+RUN git clone --recursive https://github.com/corvusoft/restbed.git \
+	&& cd restbed && mkdir build && cd build \
+	&& cmake -DBUILD_TESTS=NO -DBUILD_EXAMPLES=NO -DBUILD_SSL=NO -DBUILD_SHARED=YES -DCMAKE_INSTALL_PREFIX=/usr -DCMAKE_INSTALL_LIBDIR=lib .. \
+	&& make -j8 install \
+	&& cd .. && rm -rf restbed
+
+#build msgpack from source
+RUN wget https://github.com/msgpack/msgpack-c/releases/download/cpp-2.1.5/msgpack-2.1.5.tar.gz \
+    && tar -xzf msgpack-2.1.5.tar.gz \
+    && cd msgpack-2.1.5 && mkdir build && cd build \
+    && cmake -DMSGPACK_CXX11=ON -DMSGPACK_BUILD_EXAMPLES=OFF -DCMAKE_INSTALL_PREFIX=/usr .. \
+    && make -j8 && make install \
+    && cd ../.. && rm -rf msgpack-2.1.5 msgpack-2.1.5.tar.gz
@@ -3,13 +3,11 @@ MAINTAINER Adrien Béraud <adrien.beraud@savoirfairelinux.com>

 RUN echo "APT::Acquire::Retries \"3\";" > /etc/apt/apt.conf.d/80-retries
 RUN apt-get update && apt-get install -y \
-        apt-transport-https build-essential pkg-config git wget libncurses5-dev libreadline-dev nettle-dev \
+        apt-transport-https build-essential cmake git wget libncurses5-dev libreadline-dev nettle-dev \
        libgnutls28-dev libuv1-dev cython3 python3-dev python3-setuptools libcppunit-dev libjsoncpp-dev \
        autotools-dev autoconf libfmt-dev libhttp-parser-dev libmsgpack-dev libssl-dev \
    && apt-get clean

-RUN apt-get update && apt-get install -y python3-pip && pip3 install --upgrade cmake
-
 # libasio-dev (1.10) is too old
 RUN echo "** Building a recent version of asio ***" \
    && wget https://github.com/aberaud/asio/archive/a7d66ef4017d8f1b7f2cef1bb4ba8e23b0961571.tar.gz \
@@ -1,21 +1,40 @@
-FROM ubuntu:20.04
+FROM ubuntu:16.04
 MAINTAINER Adrien Béraud <adrien.beraud@savoirfairelinux.com>
-RUN apt-get update && apt-get install -y \
-        dialog apt-utils \
-    && apt-get clean \
-    && echo 'debconf debconf/frontend select Noninteractive' | debconf-set-selections
-
 RUN apt-get update \
-    && apt-get install -y llvm llvm-dev clang make cmake pkg-config git wget libncurses5-dev libreadline-dev \
+    && apt-get install -y llvm llvm-dev clang make cmake git wget libncurses5-dev libreadline-dev \
       nettle-dev libgnutls28-dev libuv1-dev libmsgpack-dev libjsoncpp-dev cython3 python3-dev \
       python3-setuptools libcppunit-dev python3-pip \
-       autotools-dev autoconf libssl-dev libargon2-dev \
-       libfmt-dev libhttp-parser-dev libasio-dev \
+       autotools-dev autoconf libssl-dev \
    && apt-get remove -y gcc g++ && apt-get autoremove -y && apt-get clean

 ENV CC cc
 ENV CXX c++

+RUN echo "*** Installing libfmt ***" \
+    && wget https://github.com/fmtlib/fmt/archive/5.3.0.tar.gz \
+    && tar -xzf 5.3.0.tar.gz && cd fmt-5.3.0/ \
+    && cmake -DCMAKE_INSTALL_PREFIX=/usr -DBUILD_SHARED_LIBS=On -DFMT_TEST=Off -DFMT_DOC=Off . \
+    && make -j8 && make install \
+    && cd ../ && rm -rf fmt*
+
+# libhttp-parser-dev is 2.1-2 which is too old
+RUN echo "*** Building nodejs/http_parser dependency ***" \
+    && wget https://github.com/nodejs/http-parser/archive/v2.9.2.tar.gz \
+    && tar xvf v2.9.2.tar.gz && cd http-parser-2.9.2/ \
+    && PREFIX=/usr make -j8 install \
+    && cd ../ && rm -rf http-parser*
+
+# libasio-dev (1.10.6-3) is too old
+RUN echo "** Building a recent version of asio ***" \
+    && wget https://github.com/aberaud/asio/archive/a7d66ef4017d8f1b7f2cef1bb4ba8e23b0961571.tar.gz \
+    && tar -xvf a7d66ef4017d8f1b7f2cef1bb4ba8e23b0961571.tar.gz && cd asio-a7d66ef4017d8f1b7f2cef1bb4ba8e23b0961571/asio \
+    && ./autogen.sh && ./configure --prefix=/usr --without-boost --disable-examples --disable-tests  \
+    && make install \
+    && cd ../../ && rm -rf asio*
+
+#patch for https://github.com/Stiffstream/restinio-conan-example/issues/2
+RUN apt-get update && apt-get install -y python3-pip && pip3 install --upgrade cmake
+
 RUN echo "*** Downloading RESTinio ***" \
    && mkdir restinio && cd restinio \
    && wget https://github.com/aberaud/restinio/archive/a7a10e419d9089c5b8ee63f5e3098c892f22fae4.tar.gz \
@@ -26,3 +45,18 @@ RUN echo "*** Downloading RESTinio ***" \
             -DRESTINIO_FIND_DEPS=ON -DRESTINIO_ALLOW_SOBJECTIZER=Off -DRESTINIO_USE_BOOST_ASIO=none . \
    && make -j8 && make install \
    && cd ../../ && rm -rf restinio*
+
+# build restbed from sources
+RUN git clone --recursive https://github.com/corvusoft/restbed.git \
+	&& cd restbed && mkdir build && cd build \
+	&& cmake -DBUILD_TESTS=NO -DBUILD_EXAMPLES=NO -DBUILD_SSL=NO -DBUILD_SHARED=YES -DCMAKE_INSTALL_PREFIX=/usr -DCMAKE_INSTALL_LIBDIR=lib .. \
+	&& make -j8 install \
+	&& cd .. && rm -rf restbed
+
+#build msgpack from source
+RUN wget https://github.com/msgpack/msgpack-c/releases/download/cpp-2.1.5/msgpack-2.1.5.tar.gz \
+    && tar -xzf msgpack-2.1.5.tar.gz \
+    && cd msgpack-2.1.5 && mkdir build && cd build \
+    && cmake -DMSGPACK_CXX11=ON -DMSGPACK_BUILD_EXAMPLES=OFF -DCMAKE_INSTALL_PREFIX=/usr .. \
+    && make -j8 && make install \
+    && cd ../.. && rm -rf msgpack-2.1.5 msgpack-2.1.5.tar.gz
@@ -1,6 +0,0 @@
-FROM aberaud/opendht-deps-llvm
-MAINTAINER Adrien Béraud <adrien.beraud@savoirfairelinux.com>
-RUN git clone https://github.com/savoirfairelinux/opendht.git \
-	&& cd opendht && mkdir build && cd build \
-	&& cmake .. -DCMAKE_INSTALL_PREFIX=/usr -DOPENDHT_PYTHON=On -DOPENDHT_LTO=On && make -j8 && make install \
-	&& cd ../.. && rm -rf opendht
@@ -1,8 +1,6 @@
-FROM aberaud/opendht-deps-bionic
+FROM aberaud/opendht-deps
 MAINTAINER Adrien Béraud <adrien.beraud@savoirfairelinux.com>

-RUN apt-get update && apt-get install -y python3-pip && pip3 install --upgrade cmake
-
 COPY . /root/opendht
 RUN cd /root/opendht && mkdir build && cd build \
 	&& cmake -DCMAKE_INSTALL_PREFIX=/usr -DOPENDHT_PYTHON=On -DOPENDHT_C=On -DOPENDHT_LTO=On -DOPENDHT_TESTS=ON ..  \
@@ -82,8 +82,6 @@ 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
    /**
@@ -371,8 +371,6 @@ private:
    Dht(const Dht&) = delete;
    Dht& operator=(const Dht&) = delete;

-    std::mt19937_64 rd {crypto::getSeededRandomEngine<std::mt19937_64>()};
-
    InfoHash myid {};

    uint64_t secret {};
@@ -414,6 +412,7 @@ private:
    std::map<size_t, std::tuple<size_t, size_t, size_t>> listeners {};
    size_t listener_token {1};

+    std::mt19937_64 rd {crypto::getSeededRandomEngine<std::mt19937_64>()};

    // timing
    Scheduler scheduler;
@@ -446,7 +445,7 @@ private:
    void reportedAddr(const SockAddr&);

    // Storage
-    void storageAddListener(const InfoHash& id, const Sp<Node>& node, size_t tid, Query&& = {}, int version = 0);
+    void storageAddListener(const InfoHash& id, const Sp<Node>& node, size_t tid, Query&& = {});
    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);
@@ -601,8 +600,7 @@ private:
            const InfoHash& hash,
            const Blob& token,
            size_t socket_id,
-            const Query& query,
-            int version = 0);
+            const Query& query);
    void onListenDone(const Sp<Node>& status,
            net::RequestAnswer& a,
            Sp<Search>& sr);
@@ -36,15 +36,6 @@
 #include <memory>
 #include <mutex>

-namespace dht {
-enum class PushType {
-    None = 0,
-    Android,
-    iOS
-};
-}
-MSGPACK_ADD_ENUM(dht::PushType)
-
 namespace http {
 class Request;
 struct ListenerSession;
@@ -61,13 +52,6 @@ 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
 */
@@ -82,9 +66,10 @@ public:
     * @note if the server fails to start (if port is already used or reserved),
     * it will fails silently
     */
-    DhtProxyServer(const std::shared_ptr<DhtRunner>& dht,
-        const ProxyServerConfig& config = {},
-        const std::shared_ptr<dht::Logger>& logger = {});
+    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 = {});

    virtual ~DhtProxyServer();

@@ -98,10 +83,8 @@ public:
    struct ServerStats {
        /** Current number of listen operations */
        size_t listenCount {0};
-        /** Current number of permanent put operations (hash used) */
+        /** Current number of permanent put operations */
        size_t putCount {0};
-        /** Current number of permanent put values */
-        size_t totalPermanentPuts {0};
        /** Current number of push tokens with at least one listen operation */
        size_t pushListenersCount {0};
        /** Average requests per second */
@@ -131,7 +114,6 @@ public:
            Json::Value result;
            result["listenCount"] = static_cast<Json::UInt64>(listenCount);
            result["putCount"] = static_cast<Json::UInt64>(putCount);
-            result["totalPermanentPuts"] = static_cast<Json::UInt64>(totalPermanentPuts);
            result["pushListenersCount"] = static_cast<Json::UInt64>(pushListenersCount);
            result["requestRate"] = requestRate;
            if (nodeInfo)
@@ -300,17 +282,17 @@ private:
     * @param key of the device
     * @param json, the content to send
     */
-    void sendPushNotification(const std::string& key, Json::Value&& json, PushType type, bool highPriority);
+    void sendPushNotification(const std::string& key, Json::Value&& json, bool isAndroid, bool highPriority);

    /**
     * Send push notification with an expire timeout.
     * @param ec
     * @param pushToken
     * @param json
-     * @param type
+     * @param isAndroid
     */
    void handleNotifyPushListenExpire(const asio::error_code &ec, const std::string pushToken,
-                                      std::function<Json::Value()> json, PushType type);
+                                      std::function<Json::Value()> json, const bool isAndroid);

    /**
     * Remove a push listener between a client and a hash
@@ -327,12 +309,6 @@ 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;

@@ -340,9 +316,6 @@ 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_;
@@ -371,7 +344,6 @@ private:
    struct PushSessionContext {
        std::mutex lock;
        std::string sessionId;
-        PushSessionContext(const std::string& id) : sessionId(id) {}
    };
    struct PermanentPut {
        time_point expiration;
@@ -381,32 +353,9 @@ 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_;
@@ -418,32 +367,14 @@ 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_;
@@ -389,7 +389,7 @@ public:
     * @param threaded: If false, loop() must be called periodically. Otherwise a thread is launched.
     * @param cb: Optional callback to receive general state information.
     */
-    void run(SockAddr& local4, SockAddr& local6, const Config& config, Context&& context = {});
+    void run(const SockAddr& local4, const SockAddr& local6, const Config& config, Context&& context = {});

    /**
     * Same as @run(sockaddr_in, sockaddr_in6, Identity, bool, StatusCallback), but with string IP addresses and service (port).
@@ -237,9 +237,6 @@ public:

    static Hash getRandom();

-    template <typename Rd>
-    static Hash getRandom(Rd&);
-
    template <size_t M>
    OPENDHT_PUBLIC friend std::ostream& operator<< (std::ostream& s, const Hash<M>& h);

@@ -327,19 +324,6 @@ Hash<N>::getRandom()
    return h;
 }

-template <size_t N>
-template <typename Rd>
-Hash<N>
-Hash<N>::getRandom(Rd& rdev)
-{
-    Hash h;
-    std::uniform_int_distribution<uint32_t> rand_int;
-    auto a = reinterpret_cast<uint32_t*>(h.data());
-    auto b = reinterpret_cast<uint32_t*>(h.data() + h.size());
-    std::generate(a, b, std::bind(rand_int, std::ref(rdev)));
-    return h;
-}
-
 struct HexMap : public std::array<std::array<char, 2>, 256> {
    HexMap() {
        for (size_t i=0; i<size(); i++) {
@@ -180,8 +180,7 @@ private:
            const InfoHash&,
            const Blob&,
            Tid,
-            const Query&,
-            int)> onListen {};
+            const Query&)> onListen {};
    /**
     * Called on announce request.
     *
@@ -211,7 +210,6 @@ private:

 public:
    using RequestCb = std::function<void(const Request&, RequestAnswer&&)>;
-    using RequestErrorCb = std::function<bool(const Request&, DhtProtocolException&&)>;
    using RequestExpiredCb = std::function<void(const Request&, bool)>;

    NetworkEngine(const Sp<Logger>& log, std::mt19937_64& rd, Scheduler& scheduler, std::unique_ptr<DatagramSocket>&& sock);
@@ -239,10 +237,9 @@ public:
    void clear();

    /**
-     * Sends values (with closest nodes) to a listener.
+     * Sends values (with closest nodes) to a listenner.
     *
-     * @deprecated
-     * @param sa          The address of the listener.
+     * @param sa          The address of the listenner.
     * @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.
@@ -251,14 +248,13 @@ 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, int version);
+            std::vector<Sp<Value>>&& values, const Query& q);

-    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);
+    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);

    bool isRunning(sa_family_t af) const;
    inline want_t want () const { return dht_socket->hasIPv4() and dht_socket->hasIPv6() ? (WANT4 | WANT6) : -1; }
@@ -362,9 +358,10 @@ public:
                           const InfoHash& hash,
                           const Query& query,
                           const Blob& token,
-                           Tid socketId,
+                           Sp<Request> previous,
                           RequestCb&& on_done,
-                           RequestExpiredCb&& on_expired);
+                           RequestExpiredCb&& on_expired,
+                           SocketCb&& socket_cb);
    /**
     * Send a "announce" request to a given node.
     *
@@ -403,26 +400,7 @@ public:
                                 const Value::Id& vid,
                                 const Blob& token,
                                 RequestCb&& on_done,
-                                 RequestErrorCb&& on_error,
                                 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.
@@ -435,8 +413,8 @@ public:
     */
    void processMessage(const uint8_t *buf, size_t buflen, SockAddr addr);

-    Sp<Node> insertNode(const InfoHash& id, const SockAddr& addr) {
-        auto n = cache.getNode(id, addr, scheduler.time(), 0);
+    Sp<Node> insertNode(const InfoHash& myid, const SockAddr& addr) {
+        auto n = cache.getNode(myid, addr, scheduler.time(), 0);
        onNewNode(n, 0);
        return n;
    }
@@ -512,20 +490,19 @@ 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 updateValue  {0};
+        unsigned ping    {0};
+        unsigned find    {0};
+        unsigned get     {0};
+        unsigned put     {0};
+        unsigned listen  {0};
+        unsigned refresh {0};
    };


    // basic wrapper for socket sendto function
    int send(const SockAddr& addr, const char *buf, size_t len, bool confirmed = false);

-    void sendValueParts(Tid tid, const std::vector<Blob>& svals, const SockAddr& addr);
+    void sendValueParts(const TransId& tid, const std::vector<Blob>& svals, const SockAddr& addr);
    std::vector<Blob> packValueHeader(msgpack::sbuffer&, const std::vector<Sp<Value>>&);
    void maintainRxBuffer(Tid tid);

@@ -21,8 +21,6 @@
 #include <random>
 #include <algorithm>
 #include <functional>
-#include <thread>
-#include <stdexcept>

 namespace dht {
 namespace crypto {
@@ -113,18 +111,11 @@ using random_device = std::random_device;
 template<class T = std::mt19937, std::size_t N = T::state_size>
 auto getSeededRandomEngine () -> typename std::enable_if<!!N, T>::type {
    typename T::result_type random_data[N];
-    for (unsigned i=0; i<256; i++) {
-        try {
-            random_device source;
-            std::generate(std::begin(random_data), std::end(random_data), std::ref(source));
-            std::seed_seq seeds(std::begin(random_data), std::end(random_data));
-            T seededEngine (seeds);
-            return seededEngine;
-        } catch (...) {
-            std::this_thread::sleep_for(std::chrono::milliseconds(1));
-        }
-    }
-    throw std::runtime_error("Can't seed random engine");
+    random_device source;
+    std::generate(std::begin(random_data), std::end(random_data), std::ref(source));
+    std::seed_seq seeds(std::begin(random_data), std::end(random_data));
+    T seededEngine (seeds);
+    return seededEngine;
 }

 }} // dht::crypto
@@ -173,13 +173,7 @@ unpackMsg(Blob b) {

 msgpack::unpacked unpackMsg(Blob b);

-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());
-}
+msgpack::object* findMapValue(const msgpack::object& map, const char* key);
+msgpack::object* findMapValue(const msgpack::object& map, const std::string& key);

 } // namespace dht
@@ -490,29 +490,7 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr) {
                    logger_->w(sr->id, sn->node->id, "[search %s] [node %s] sending 'refresh' (vid: %d)",
                        sr->id.toString().c_str(), sn->node->toString().c_str(), a.value->id);
                sn->acked[a.value->id] = {
-                    network_engine.sendRefreshValue(sn->node, sr->id, a.value->id, sn->token, onDone,
-                    [this, ws, node=sn->node, v=a.value, 
-                     onDone, 
-                     onExpired, 
-                     created = a.permanent ? time_point::max() : a.created,
-                     next_refresh_time
-                    ](const net::Request& req, net::DhtProtocolException&& e){
-                        if (e.getCode() == net::DhtProtocolException::NOT_FOUND) {
-                            if (logger_)
-                                logger_->e(node->id, "[node %s] returned error 404: storage not found", node->toString().c_str());
-                            if (auto sr = ws.lock()) {
-                                if (auto sn = sr->getNode(node)) {
-                                    sn->acked[v->id] = {
-                                        network_engine.sendAnnounceValue(sn->node, sr->id, v, created, sn->token, onDone, onExpired),
-                                        next_refresh_time
-                                    };
-                                    scheduler.edit(sr->nextSearchStep, scheduler.time());
-                                    return true;
-                                }
-                            }
-                        }
-                        return false;
-                    }, onExpired),
+                    network_engine.sendRefreshValue(sn->node, sr->id, a.value->id, sn->token, onDone, onExpired),
                    next_refresh_time
                };
            } else {
@@ -591,30 +569,33 @@ Dht::searchSynchedNodeListen(const Sp<Search>& sr, SearchNode& n)
    std::weak_ptr<Search> ws = sr;
    for (const auto& l : sr->listeners) {
        const auto& query = l.second.query;
-        
-        auto r = n.listenStatus.find(query);
-        if (n.getListenTime(r, listenExp) > scheduler.time())
+        auto list_token = l.first;
+        if (n.getListenTime(query, listenExp) > scheduler.time())
            continue;
        // if (logger_)
        //     logger_->d(sr->id, n.node->id, "[search %s] [node %s] sending 'listen'",
        //        sr->id.toString().c_str(), n.node->toString().c_str());

+        auto r = n.listenStatus.find(query);
        if (r == n.listenStatus.end()) {
            r = n.listenStatus.emplace(std::piecewise_construct,
                std::forward_as_tuple(query),
                std::forward_as_tuple(
-                    l.second.get_cb,
-                    l.second.sync_cb,
-                    n.node->openSocket([this,ws,query](const Sp<Node>& node, net::RequestAnswer&& answer) mutable {
-                        /* on new values */
-                        if (auto sr = ws.lock()) {
-                            scheduler.edit(sr->nextSearchStep, scheduler.time());
-                            sr->insertNode(node, scheduler.time(), answer.ntoken);
-                            if (auto sn = sr->getNode(node)) {
-                                sn->onValues(query, std::move(answer), types, scheduler);
-                            }
+                [ws,list_token](const std::vector<Sp<Value>>& values, bool expired){
+                    if (auto sr = ws.lock()) {
+                        auto l = sr->listeners.find(list_token);
+                        if (l != sr->listeners.end()) {
+                            l->second.get_cb(values, expired);
                        }
-                    }))).first;
+                    }
+                }, [ws,list_token] (ListenSyncStatus status) {
+                    if (auto sr = ws.lock()) {
+                        auto l = sr->listeners.find(list_token);
+                        if (l != sr->listeners.end()) {
+                            l->second.sync_cb(status);
+                        }
+                    }
+                })).first;
            r->second.cacheExpirationJob = scheduler.add(time_point::max(), [this,ws,query,node=n.node]{
                if (auto sr = ws.lock()) {
                    if (auto sn = sr->getNode(node)) {
@@ -623,7 +604,8 @@ Dht::searchSynchedNodeListen(const Sp<Search>& sr, SearchNode& n)
                }
            });
        }
-        auto new_req = network_engine.sendListen(n.node, sr->id, *query, n.token, r->second.socketId,
+        auto prev_req = r != n.listenStatus.end() ? r->second.req : nullptr;
+        auto new_req = network_engine.sendListen(n.node, sr->id, *query, n.token, prev_req,
            [this,ws,query](const net::Request& req, net::RequestAnswer&& answer) mutable
            { /* on done */
                if (auto sr = ws.lock()) {
@@ -643,6 +625,16 @@ Dht::searchSynchedNodeListen(const Sp<Search>& sr, SearchNode& n)
                        if (auto sn = sr->getNode(req.node))
                            sn->listenStatus.erase(query);
                }
+            },
+            [this,ws,query](const Sp<Node>& node, net::RequestAnswer&& answer) mutable
+            { /* on new values */
+                if (auto sr = ws.lock()) {
+                    scheduler.edit(sr->nextSearchStep, scheduler.time());
+                    sr->insertNode(node, scheduler.time(), answer.ntoken);
+                    if (auto sn = sr->getNode(node)) {
+                        sn->onValues(query, std::move(answer), types, scheduler);
+                    }
+                }
            }
        );
        // Here the request may have failed and the CachedListenStatus removed
@@ -1209,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, l.second.version);
+                        std::move(vals), l.second.query);
            }
        }
    }
@@ -1267,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, int version)
+Dht::storageAddListener(const InfoHash& id, const Sp<Node>& node, size_t socket_id, Query&& query)
 {
    const auto& now = scheduler.time();
    auto st = store.find(id);
@@ -1283,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, version);
+                    std::move(vals), query);
        }
-        node_listeners.emplace(socket_id, Listener {now, std::forward<Query>(query), version});
+        node_listeners.emplace(socket_id, Listener {now, std::forward<Query>(query)});
    }
    else
        l->second.refresh(now, std::forward<Query>(query));
@@ -1320,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, l.second.version);
+                    network_engine.tellListenerExpired(node_listeners.first, l.first, id, ntoken, ids);
                }
            }
        }
@@ -1751,7 +1743,7 @@ Dht::Dht() : store(), network_engine(logger_, rd, scheduler, {}) {}

 Dht::Dht(std::unique_ptr<net::DatagramSocket>&& sock, const Config& config, const Sp<Logger>& l)
    : DhtInterface(l),
-    myid(config.node_id ? config.node_id : InfoHash::getRandom(rd)),
+    myid(config.node_id ? config.node_id : InfoHash::getRandom()),
    store(),
    store_quota(),
    max_store_keys(config.max_store_size ? (int)config.max_store_size : MAX_HASHES),
@@ -1763,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, _6),
+            std::bind(&Dht::onListen, this, _1, _2, _3, _4, _5),
            std::bind(&Dht::onAnnounce, this, _1, _2, _3, _4, _5),
            std::bind(&Dht::onRefresh, this, _1, _2, _3, _4)),
    persistPath(config.persist_path),
@@ -1793,13 +1785,13 @@ Dht::Dht(std::unique_ptr<net::DatagramSocket>&& sock, const Config& config, cons
    secret = std::uniform_int_distribution<uint64_t>{}(rd);
    rotateSecrets();

-    if (not persistPath.empty())
-        loadState(persistPath);
-
    expire();

    if (logger_)
        logger_->d("DHT node initialised with ID %s", myid.toString().c_str());
+
+    if (not persistPath.empty())
+        loadState(persistPath);
 }

 bool
@@ -1990,19 +1982,14 @@ Dht::onDisconnected()
        return;
    if (logger_)
        logger_->d(myid, "Bootstraping");
-    for (const auto& boootstrap : bootstrap_nodes) {
+    for (const auto& boootstrap : bootstrap_nodes)
        try {
-            auto ips = network_engine.getSocket()->resolve(boootstrap.first, boootstrap.second);
-            for (auto& ip : ips) {
-                if (ip.getPort() == 0)
-                    ip.setPort(net::DHT_DEFAULT_PORT);
+            for (const auto& ip : network_engine.getSocket()->resolve(boootstrap.first, boootstrap.second))
                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));
@@ -2078,13 +2065,13 @@ Dht::importValues(const std::vector<ValuesExport>& import)
 {
    const auto& now = scheduler.time();

-    for (const auto& value : import) {
-        if (value.second.empty())
+    for (const auto& node : import) {
+        if (node.second.empty())
            continue;

        try {
            msgpack::unpacked msg;
-            msgpack::unpack(msg, (const char*)value.second.data(), value.second.size());
+            msgpack::unpack(msg, (const char*)node.second.data(), node.second.size());
            auto valarr = msg.get();
            if (valarr.type != msgpack::type::ARRAY)
                throw msgpack::type_error();
@@ -2099,15 +2086,15 @@ Dht::importValues(const std::vector<ValuesExport>& import)
                    tmp_val.msgpack_unpack(valel.via.array.ptr[1]);
                } catch (const std::exception&) {
                    if (logger_)
-                        logger_->e(value.first, "Error reading value at %s", value.first.toString().c_str());
+                        logger_->e(node.first, "Error reading value at %s", node.first.toString().c_str());
                    continue;
                }
                val_time = std::min(val_time, now);
-                storageStore(value.first, std::make_shared<Value>(std::move(tmp_val)), val_time);
+                storageStore(node.first, std::make_shared<Value>(std::move(tmp_val)), val_time);
            }
        } catch (const std::exception&) {
            if (logger_)
-                logger_->e(value.first, "Error reading values at %s", value.first.toString().c_str());
+                logger_->e(node.first, "Error reading values at %s", node.first.toString().c_str());
            continue;
        }
    }
@@ -2183,6 +2170,7 @@ Dht::onError(Sp<net::Request> req, net::DhtProtocolException e) {
        if (logger_)
            logger_->e(node->id, "[node %s] token flush", node->toString().c_str());
        node->authError();
+        node->cancelRequest(req);
        for (auto& srp : searches(node->getFamily())) {
            auto& sr = srp.second;
            for (auto& n : sr->nodes) {
@@ -2332,7 +2320,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, int version)
+Dht::onListen(Sp<Node> node, const InfoHash& hash, const Blob& token, size_t socket_id, const Query& query)
 {
    if (not hash) {
        if (logger_)
@@ -2348,7 +2336,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), version);
+    storageAddListener(hash, node, socket_id, std::move(q));
    return {};
 }

@@ -2487,7 +2475,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, l.second.version);
+                    network_engine.tellListenerRefreshed(node_listeners.first, l.first, id, ntoken, ids);
                }
            }
        }
@@ -2511,24 +2499,12 @@ Dht::onAnnounceDone(const Sp<Node>& node, net::RequestAnswer& answer, Sp<Search>
 }


-struct DhtState {
-    unsigned v {1};
-    InfoHash id;
-    std::vector<NodeExport> nodes;
-    std::vector<ValuesExport> values;
-
-    MSGPACK_DEFINE_MAP(v, id, nodes, values)
-};
-
 void
 Dht::saveState(const std::string& path) const
 {
-    DhtState state;
-    state.id = myid;
-    state.nodes = exportNodes();
-    state.values = exportValues();
    std::ofstream file(path);
-    msgpack::pack(file, state);    
+    msgpack::pack(file, exportNodes());
+    msgpack::pack(file, exportValues());
 }

 void
@@ -2554,16 +2530,19 @@ Dht::loadState(const std::string& path)
        // Import nodes
        msgpack::object_handle oh;
        if (pac.next(oh)) {
-            auto state = oh.get().as<DhtState>();
-            if (logger_)
-                logger_->d("Importing %zu nodes", state.nodes.size());
-            if (state.id)
-                myid = state.id;
-            std::vector<Sp<Node>> tmpNodes;
-            tmpNodes.reserve(state.nodes.size());
-            for (const auto& node : state.nodes)
-                tmpNodes.emplace_back(network_engine.insertNode(node.id, SockAddr(node.ss, node.sslen)));
-            importValues(state.values);
+            {
+                auto imported_nodes = oh.get().as<std::vector<NodeExport>>();
+                if (logger_)
+                    logger_->d("Importing %zu nodes", imported_nodes.size());
+                for (const auto& node : imported_nodes)
+                    insertNode(node);
+            }
+            if (pac.next(oh)) {
+                auto imported_values = oh.get().as<std::vector<ValuesExport>>();
+                if (logger_)
+                    logger_->d("Importing %zu values", imported_values.size());
+                importValues(imported_values);
+            }
        }
    } catch (const std::exception& e) {
        if (logger_)
@@ -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)) - 2);
+    std::uniform_int_distribution<> dist(0, (sizeof(chars)/sizeof(char)) - 1);
    std::generate_n( str.begin(), length, [&]{ return chars[dist(rdev)]; } );
    return str;
 }
@@ -34,7 +34,6 @@
 #include <functional>
 #include <limits>
 #include <iostream>
-#include <fstream>

 using namespace std::placeholders;
 using namespace std::chrono_literals;
@@ -174,73 +173,26 @@ struct DhtProxyServer::RestRouterTraits : public restinio::default_traits_t
    using connection_state_listener_t = ConnectionListener;
 };

-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
+DhtProxyServer::DhtProxyServer(
+    crypto::Identity identity,
+    std::shared_ptr<DhtRunner> dht, in_port_t port, const std::string& pushServer,
+    std::shared_ptr<Logger> logger
 )
    :   ioContext_(std::make_shared<asio::io_context>()),
-        dht_(dht), persistPath_(config.persistStatePath), logger_(logger),
+        dht_(dht), 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_(config.pushServer)
+        pushServer_(pushServer)
 {
    if (not dht_)
        throw std::invalid_argument("A DHT instance must be provided");

    if (logger_)
-        logger_->d("[proxy:server] [init] running on %i", config.port);
-    if (not pushServer_.empty()){
+        logger_->d("[proxy:server] [init] running on %i", 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");
@@ -250,9 +202,9 @@ DhtProxyServer::DhtProxyServer(const std::shared_ptr<DhtRunner>& dht,
    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_);
@@ -260,7 +212,7 @@ DhtProxyServer::DhtProxyServer(const std::shared_ptr<DhtRunner>& dht,
            logger_->d("Using push server for notifications: %s:%s", pushHostPort_.first.c_str(),
                                                                     pushHostPort_.second.c_str());
    }
-    if (config.identity.first and config.identity.second) {
+    if (identity.first and identity.second) {
        asio::error_code ec;
        // define tls context
        asio::ssl::context tls_context { asio::ssl::context::sslv23 };
@@ -274,13 +226,13 @@ DhtProxyServer::DhtProxyServer(const std::shared_ptr<DhtRunner>& dht,
        SSL_CTX_set_options(tls_context.native_handle(), SSL_OP_NO_RENEGOTIATION); // CVE-2009-3555
 #endif
        // node private key
-        auto key = config.identity.first->serialize();
+        auto key = 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 = config.identity.second->toString(true/*chain*/);
+        auto certchain = 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());
@@ -289,7 +241,7 @@ DhtProxyServer::DhtProxyServer(const std::shared_ptr<DhtRunner>& dht,
        // build http server
        auto settings = restinio::run_on_this_thread_settings_t<RestRouterTraitsTls>();
        addServerSettings(settings);
-        settings.port(config.port);
+        settings.port(port);
        settings.tls_context(std::move(tls_context));
        httpsServer_ = std::make_unique<restinio::http_server_t<RestRouterTraitsTls>>(
            ioContext_,
@@ -306,7 +258,7 @@ DhtProxyServer::DhtProxyServer(const std::shared_ptr<DhtRunner>& dht,
    else {
        auto settings = restinio::run_on_this_thread_settings_t<RestRouterTraits>();
        addServerSettings(settings);
-        settings.port(config.port);
+        settings.port(port);
        httpServer_ = std::make_unique<restinio::http_server_t<RestRouterTraits>>(
            ioContext_,
            std::forward<restinio::run_on_this_thread_settings_t<RestRouterTraits>>(std::move(settings))
@@ -322,158 +274,8 @@ DhtProxyServer::DhtProxyServer(const std::shared_ptr<DhtRunner>& dht,
    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_);
-    }
-#ifdef OPENDHT_PUSH_NOTIFICATIONS
-    {
-        std::lock_guard<std::mutex> lock(lockListener_);
-        pk.pack("pushListeners");
-        pk.pack(pushListeners_);
-    }
-#endif
-}
-
-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));
-#ifdef OPENDHT_PUSH_NOTIFICATIONS
-                        if (not pput.second.pushToken.empty()) {
-                            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);
-                                if (sessionCtx) {
-                                    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));
-                        }
-#endif
-                        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
 {
@@ -482,12 +284,6 @@ 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_) {
@@ -563,10 +359,6 @@ DhtProxyServer::updateStats(std::shared_ptr<NodeInfo> info) const
 #ifdef OPENDHT_PUSH_NOTIFICATIONS
    stats.pushListenersCount = pushListeners_.size();
 #endif
-    stats.totalPermanentPuts = 0;
-    std::for_each(puts_.begin(), puts_.end(), [&stats](const auto& put) {
-        stats.totalPermanentPuts += put.second.puts.size();
-    });
    stats.putCount = puts_.size();
    stats.listenCount = listeners_.size();
    stats.nodeInfo = std::move(info);
@@ -793,11 +585,12 @@ DhtProxyServer::subscribe(restinio::request_handle_t request,
                          restinio::router::route_params_t params)
 {
    requestNum_++;
-    try {
-        InfoHash infoHash(params["hash"].to_string());
-        if (!infoHash)
-            infoHash = InfoHash::get(params["hash"].to_string());

+    InfoHash infoHash(params["hash"].to_string());
+    if (!infoHash)
+        infoHash = InfoHash::get(params["hash"].to_string());
+
+    try {
        std::string err;
        Json::Value r;
        auto* char_data = reinterpret_cast<const char*>(request->body().data());
@@ -814,115 +607,117 @@ DhtProxyServer::subscribe(restinio::request_handle_t request,
            response.set_body(RESP_MSG_NO_TOKEN);
            return response.done();
        }
-        auto type = root["platform"].asString() == "android" ? PushType::Android : PushType::iOS;
+        auto isAndroid = root["platform"].asString() == "android";
        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];

-        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;
+        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& listener = *listIt;
+        // =========== 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;

-        // 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);
-        else
+        // 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);
-        auto jsonProvider = [h=infoHash.toString(), clientId, sessionCtx = listener.sessionCtx](){
+        else
+            listener.expireNotifyTimer->expires_at(timeout - proxy::OP_MARGIN);
+        auto jsonProvider = [infoHash, clientId, sessionCtx = listener.sessionCtx](){
            Json::Value json;
-            json["timeout"] = h;
+            json["timeout"] = infoHash.toString();
            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), listener.type));
+                                               std::placeholders::_1, pushToken, std::move(jsonProvider), isAndroid));
+        // cancel push listen
        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));
-
-        // 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();
-        }
+                                         std::placeholders::_1, pushToken, infoHash, clientId));
+        auto response = initHttpResponse(request->create_response());
+        response.set_body("{}\n");
+        return response.done();
    }
    catch (...) {
        return serverError(*request);
@@ -971,7 +766,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, PushType type)
+                                             std::function<Json::Value()> jsonProvider, const bool isAndroid)
 {
    if (ec == asio::error::operation_aborted)
        return;
@@ -981,7 +776,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(), type, false);
+    sendPushNotification(pushToken, jsonProvider(), isAndroid, false);
 }

 void
@@ -1023,7 +818,7 @@ DhtProxyServer::handleCancelPushListen(const asio::error_code &ec, const std::st
 }

 void
-DhtProxyServer::sendPushNotification(const std::string& token, Json::Value&& json, PushType type, bool highPriority)
+DhtProxyServer::sendPushNotification(const std::string& token, Json::Value&& json, bool isAndroid, bool highPriority)
 {
    if (pushServer_.empty())
        return;
@@ -1045,7 +840,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"] = type == PushType::Android ? 2 : 1;
+        notification["platform"] = isAndroid ? 2 : 1;
        notification["data"] = std::move(json);
        notification["priority"] = highPriority ? "high" : "normal";
        notification["time_to_live"] = 600;
@@ -1094,7 +889,6 @@ 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;
@@ -1104,8 +898,6 @@ 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);
@@ -1149,13 +941,13 @@ DhtProxyServer::put(restinio::request_handle_t request,
                    platform = pVal["platform"].asString();
                    sessionId = pVal["session_id"].asString();
                }
-                std::lock_guard<std::mutex> lock(lockSearchPuts_);
+                std::unique_lock<std::mutex> lock(lockSearchPuts_);
                auto timeout = std::chrono::steady_clock::now() + proxy::OP_TIMEOUT;
                auto& sPuts = puts_[infoHash];
                if (value->id == Value::INVALID_ID) {
                    for (auto& pp : sPuts.puts) {
-                        if (pp.second.pushToken == pushToken
-                            and pp.second.clientId == clientId
+                        if (pp.second.pushToken == pushToken 
+                            and pp.second.clientId == clientId 
                            and pp.second.value->contentEquals(*value))
                        {
                            pp.second.expireTimer->expires_at(timeout);
@@ -1163,36 +955,32 @@ 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>(sessionId);
-                                else {
-                                    std::lock_guard<std::mutex> l(pp.second.sessionCtx->lock);
-                                    pp.second.sessionCtx->sessionId = sessionId;
-                                }
+                                    pp.second.sessionCtx = std::make_shared<PushSessionContext>();
+                                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.type = isAndroid ? PushType::Android : PushType::iOS;
-                        pput.sessionCtx = std::make_shared<PushSessionContext>(sessionId);
+                        pput.sessionCtx = std::make_shared<PushSessionContext>();
+                        pput.sessionCtx->sessionId = sessionId;
                        // notify push listen expire
+                        bool isAndroid = platform == "android";
                        auto jsonProvider = [infoHash, clientId, vid, sessionCtx = pput.sessionCtx](){
                            Json::Value json;
                            json["timeout"] = infoHash.toString();
@@ -1209,17 +997,15 @@ 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), pput.type));
+                            std::placeholders::_1, pushToken, std::move(jsonProvider), isAndroid));
                    }
 #endif
                } else {
                    if (not sessionId.empty()) {
                        if (not pput.sessionCtx)
-                            pput.sessionCtx = std::make_shared<PushSessionContext>(sessionId);
-                        else {
-                            std::lock_guard<std::mutex> l(pput.sessionCtx->lock);
-                            pput.sessionCtx->sessionId = sessionId;
-                        }
+                            pput.sessionCtx = std::make_shared<PushSessionContext>();
+                        std::lock_guard<std::mutex> l(pput.sessionCtx->lock);
+                        pput.sessionCtx->sessionId = sessionId;
                    }
                    pput.expireTimer->expires_at(timeout);
                    if (pput.expireNotifyTimer)
@@ -32,8 +32,6 @@
 #include "dht_proxy_client.h"
 #endif

-#include <fstream>
-
 namespace dht {

 constexpr std::chrono::seconds DhtRunner::BOOTSTRAP_PERIOD;
@@ -92,46 +90,16 @@ DhtRunner::run(const char* ip4, const char* ip6, const char* service, const Conf
 {
    auto res4 = SockAddr::resolve(ip4, service);
    auto res6 = SockAddr::resolve(ip6, service);
-    if (res4.empty())
-        res4.emplace_back();
-    if (res6.empty())
-        res6.emplace_back();
-    run(res4.front(), res6.front(), config, std::move(context));
+    run(res4.empty() ? SockAddr() : res4.front(),
+        res6.empty() ? SockAddr() : res6.front(), config, std::move(context));
 }

 void
-DhtRunner::run(SockAddr& local4, SockAddr& local6, const Config& config, Context&& context)
+DhtRunner::run(const SockAddr& local4, const SockAddr& local6, const Config& config, Context&& context)
 {
    if (running == State::Idle) {
-        auto state_path = config.dht_config.node_config.persist_path;
-        if (not state_path.empty()) {
-            state_path += "_port.txt";
-            std::ifstream inConfig(state_path);
-            if (inConfig.is_open()) {
-                in_port_t port;
-                if (inConfig >> port) {
-                    if (context.logger)
-                        context.logger->d("[runner %p] Using IPv4 port %hu from saved configuration", this, port);
-                    if (local4.getPort() == 0)
-                        local4.setPort(port);
-                }
-                if (inConfig >> port) {
-                    if (context.logger)
-                        context.logger->d("[runner %p] Using IPv6 port %hu from saved configuration", this, port);
-                    if (local6.getPort() == 0)
-                        local6.setPort(port);
-                }
-            }
-        }
-
        if (not context.sock)
            context.sock.reset(new net::UdpSocket(local4, local6, context.logger));
-
-        if (not state_path.empty()) {
-            std::ofstream outConfig(state_path);
-            outConfig << context.sock->getBoundRef(AF_INET).getPort() << std::endl;
-            outConfig << context.sock->getBoundRef(AF_INET6).getPort() << std::endl;
-        }
        run(config, std::move(context));
    }
 }
@@ -507,10 +475,6 @@ 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;
@@ -528,10 +492,6 @@ 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();
@@ -30,9 +30,8 @@ namespace dht {
 struct Listener {
    time_point time;
    Query query;
-    int version;

-    Listener(time_point t, Query&& q, int version = 0) : time(t), query(std::move(q)), version(version) {}
+    Listener(time_point t, Query&& q) : time(t), query(std::move(q)) {}

    void refresh(time_point t, Query&& q) {
        time = t;
@@ -31,8 +31,7 @@ enum class MessageType {
    Refresh,
    Listen,
    ValueData,
-    ValueUpdate,
-    UpdateValue
+    ValueUpdate
 };

 } /* namespace net */
@@ -48,6 +48,23 @@ 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;
@@ -123,15 +140,11 @@ 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, int version)
+        const Query& query)
 {
    auto nnodes = bufferNodes(node->getFamily(), hash, want, nodes, nodes6);
    try {
-        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);
-        }
+        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 !");
@@ -139,103 +152,63 @@ 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, int version)
+NetworkEngine::tellListenerRefreshed(Sp<Node> n, Tid socket_id, const InfoHash&, const Blob& token, const std::vector<Value::Id>& values)
 {
    msgpack::sbuffer buffer;
    msgpack::packer<msgpack::sbuffer> pk(&buffer);
+    pk.pack_map(4+(config.network?1:0));

-    pk.pack_map(4 + (version >= 1 ? 1 : 0) + (config.network?1:0));
+    pk.pack(KEY_U);
+    pk.pack_map(1 + (not values.empty()?1:0) + (not token.empty()?1:0));
+    pk.pack(KEY_REQ_ID); pk.pack(myid);
+    if (not token.empty()) {
+        pk.pack(KEY_REQ_TOKEN); packToken(pk, token);
+    }
+    if (not values.empty()) {
+        pk.pack(KEY_REQ_REFRESHED);
+        pk.pack(values);
+        if (logger_)
+            logger_->d(n->id, "[node %s] sending %zu refreshed values", n->toString().c_str(), values.size());
+    }

-    pk.pack(version >= 1 ? KEY_A : KEY_U);
-        pk.pack_map(1 + (version >= 1 ? 1 : 0) + (not values.empty()?1:0) + (not token.empty()?1:0));
-        pk.pack(KEY_REQ_ID); pk.pack(myid);
-        if (version >= 1) {
-            pk.pack(KEY_REQ_SID);   pk.pack(socket_id);
-        }
-        if (not token.empty()) {
-            pk.pack(KEY_REQ_TOKEN); packToken(pk, token);
-        }
-        if (not values.empty()) {
-            pk.pack(KEY_REQ_REFRESHED);
-            pk.pack(values);
-            if (logger_)
-                logger_->d(n->id, "[node %s] sending %zu refreshed values", n->toString().c_str(), values.size());
-        }
-
-    pk.pack(KEY_Y); pk.pack(version >= 1 ? KEY_Q : KEY_R);
+    pk.pack(KEY_TID); pk.pack(socket_id);
+    pk.pack(KEY_Y); pk.pack(KEY_R);
    pk.pack(KEY_UA); pk.pack(my_v);
    if (config.network) {
        pk.pack(KEY_NETID); pk.pack(config.network);
    }

-    if (version >= 1) {
-        Tid tid (n->getNewTid());
-
-        pk.pack(KEY_Q);   pk.pack(QUERY_UPDATE);
-        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;
-    }
-    pk.pack(KEY_TID); pk.pack(socket_id);
-
    // 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, int version)
+NetworkEngine::tellListenerExpired(Sp<Node> n, Tid socket_id, const InfoHash&, const Blob& token, const std::vector<Value::Id>& values)
 {
    msgpack::sbuffer buffer;
    msgpack::packer<msgpack::sbuffer> pk(&buffer);
+    pk.pack_map(4+(config.network?1:0));

-    pk.pack_map(4 + (version >= 1 ? 1 : 0) + (config.network?1:0));
+    pk.pack(KEY_U);
+    pk.pack_map(1 + (not values.empty()?1:0) + (not token.empty()?1:0));
+    pk.pack(KEY_REQ_ID); pk.pack(myid);
+    if (not token.empty()) {
+        pk.pack(KEY_REQ_TOKEN); packToken(pk, token);
+    }
+    if (not values.empty()) {
+        pk.pack(KEY_REQ_EXPIRED);
+        pk.pack(values);
+        if (logger_)
+            logger_->d(n->id, "[node %s] sending %zu expired values", n->toString().c_str(), values.size());
+    }

-    pk.pack(version >= 1 ? KEY_A : KEY_U);
-        pk.pack_map(1 + (version >= 1 ? 1 : 0) + (not values.empty()?1:0) + (not token.empty()?1:0));
-        pk.pack(KEY_REQ_ID); pk.pack(myid);
-        if (version >= 1) {
-            pk.pack(KEY_REQ_SID);   pk.pack(socket_id);
-        }
-        if (not token.empty()) {
-            pk.pack(KEY_REQ_TOKEN); packToken(pk, token);
-        }
-        if (not values.empty()) {
-            pk.pack(KEY_REQ_EXPIRED);
-            pk.pack(values);
-            if (logger_)
-                logger_->d(n->id, "[node %s] sending %zu expired values", n->toString().c_str(), values.size());
-        }
-
-    pk.pack(KEY_Y); pk.pack(version >= 1 ? KEY_Q : KEY_R);
+    pk.pack(KEY_TID); pk.pack(socket_id);
+    pk.pack(KEY_Y); pk.pack(KEY_R);
    pk.pack(KEY_UA); pk.pack(my_v);
    if (config.network) {
        pk.pack(KEY_NETID); pk.pack(config.network);
    }

-    if (version >= 1) {
-        Tid tid (n->getNewTid());
-
-        pk.pack(KEY_Q);   pk.pack(QUERY_UPDATE);
-        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;
-    }
-    pk.pack(KEY_TID); pk.pack(socket_id);
-
    // send response
    send(n->getAddr(), buffer.data(), buffer.size());
 }
@@ -560,8 +533,7 @@ NetworkEngine::process(std::unique_ptr<ParsedMessage>&& msg, const SockAddr& fro
            {
                req->last_try = time_point::min();
                req->reply_time = time_point::min();
-                if (not req->setError(DhtProtocolException {msg->error_code}))
-                    onError(req, DhtProtocolException {msg->error_code});
+                onError(req, DhtProtocolException {msg->error_code});
            } else {
                if (logIncoming_)
                    if (logger_)
@@ -573,11 +545,8 @@ NetworkEngine::process(std::unique_ptr<ParsedMessage>&& msg, const SockAddr& fro
        case MessageType::Reply:
            if (req) { /* request reply */
                auto& r = *req;
-                if (r.getType() == MessageType::AnnounceValue
-                 or r.getType() == MessageType::Listen
-                 or r.getType() == MessageType::Refresh) {
+                if (r.getType() == MessageType::AnnounceValue or r.getType() == MessageType::Listen)
                    r.node->authSuccess();
-                }
                r.reply_time = scheduler.time();

                deserializeNodes(*msg, from);
@@ -648,29 +617,18 @@ 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), msg->version);
+                RequestAnswer answer = onListen(node, msg->info_hash, msg->token, msg->socket_id, std::move(msg->query));
                auto nnodes = bufferNodes(from.getFamily(), msg->info_hash, msg->want, answer.nodes4, answer.nodes6);
                sendListenConfirmation(from, msg->tid);
                break;
            }
-            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());
-                sendListenConfirmation(from, msg->tid);
-                break;
-            }
            default:
                break;
            }
        } catch (const std::overflow_error& e) {
            if (logger_)
                logger_->e("Can't send value: buffer not large enough !");
-        } catch (const DhtProtocolException& e) {
+        } catch (DhtProtocolException& e) {
            sendError(from, msg->tid, e.getCode(), e.getMsg().c_str(), true);
        }
    }
@@ -696,7 +654,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) {
-    Tid tid (node->getNewTid());
+    TransId tid (node->getNewTid());
    msgpack::sbuffer buffer;
    msgpack::packer<msgpack::sbuffer> pk(&buffer);
    pk.pack_map(5+(config.network?1:0));
@@ -705,14 +663,15 @@ 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(tid);
+    pk.pack(KEY_TID); pk.pack_bin(tid.size());
+                              pk.pack_bin_body((const char*)tid.data(), tid.size());
    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, node,
+    auto req = std::make_shared<Request>(MessageType::Ping, tid.toInt(), node,
        Blob(buffer.data(), buffer.data() + buffer.size()),
        [=](const Request& req_status, ParsedMessage&&) {
            if (logger_)
@@ -742,7 +701,9 @@ NetworkEngine::sendPong(const SockAddr& addr, Tid tid) {
      pk.pack(KEY_REQ_ID); pk.pack(myid);
      insertAddr(pk, addr);

-    pk.pack(KEY_TID); pk.pack(tid);
+    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_Y); pk.pack(KEY_R);
    pk.pack(KEY_UA); pk.pack(my_v);
    if (config.network) {
@@ -755,7 +716,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) {
-    Tid tid (n->getNewTid());
+    TransId tid (n->getNewTid());
    msgpack::sbuffer buffer;
    msgpack::packer<msgpack::sbuffer> pk(&buffer);
    pk.pack_map(5+(config.network?1:0));
@@ -771,14 +732,15 @@ 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(tid);
+    pk.pack(KEY_TID); pk.pack_bin(tid.size());
+                               pk.pack_bin_body((const char*)tid.data(), tid.size());
    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, n,
+    auto req = std::make_shared<Request>(MessageType::FindNode, tid.toInt(), n,
        Blob(buffer.data(), buffer.data() + buffer.size()),
        [=](const Request& req_status, ParsedMessage&& msg) { /* on done */
            if (on_done) {
@@ -800,7 +762,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) {
-    Tid tid (n->getNewTid());
+    TransId tid (n->getNewTid());
    msgpack::sbuffer buffer;
    msgpack::packer<msgpack::sbuffer> pk(&buffer);
    pk.pack_map(5+(config.network?1:0));
@@ -824,14 +786,15 @@ 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(tid);
+    pk.pack(KEY_TID); pk.pack_bin(tid.size());
+                      pk.pack_bin_body((const char*)tid.data(), tid.size());
    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, n,
+    auto req = std::make_shared<Request>(MessageType::GetValues, tid.toInt(), n,
        Blob(buffer.data(), buffer.data() + buffer.size()),
        [=](const Request& req_status, ParsedMessage&& msg) { /* on done */
            if (on_done) {
@@ -933,7 +896,7 @@ NetworkEngine::packValueHeader(msgpack::sbuffer& buffer, const std::vector<Sp<Va
 }

 void
-NetworkEngine::sendValueParts(Tid tid, const std::vector<Blob>& svals, const SockAddr& addr)
+NetworkEngine::sendValueParts(const TransId& tid, const std::vector<Blob>& svals, const SockAddr& addr)
 {
    msgpack::sbuffer buffer;
    unsigned i=0;
@@ -948,7 +911,8 @@ NetworkEngine::sendValueParts(Tid tid, const std::vector<Blob>& svals, const Soc
                pk.pack(KEY_NETID); pk.pack(config.network);
            }
            pk.pack(KEY_Y); pk.pack(KEY_V);
-            pk.pack(KEY_TID); pk.pack(tid);
+            pk.pack(KEY_TID); pk.pack_bin(tid.size());
+                                       pk.pack_bin_body((const char*)tid.data(), tid.size());
            pk.pack(KEY_V); pk.pack_map(1);
                pk.pack(i); pk.pack_map(2);
                    pk.pack(std::string("o")); pk.pack(start);
@@ -1003,7 +967,9 @@ NetworkEngine::sendNodesValues(const SockAddr& addr, Tid tid, const Blob& nodes,
        }
    }

-    pk.pack(KEY_TID); pk.pack(tid);
+    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_Y); pk.pack(KEY_R);
    pk.pack(KEY_UA); pk.pack(my_v);
    if (config.network) {
@@ -1073,35 +1039,54 @@ NetworkEngine::sendListen(Sp<Node> n,
        const InfoHash& hash,
        const Query& query,
        const Blob& token,
-        Tid socketId,
+        Sp<Request> previous,
        RequestCb&& on_done,
-        RequestExpiredCb&& on_expired)
+        RequestExpiredCb&& on_expired,
+        SocketCb&& socket_cb)
 {
-    Tid tid (n->getNewTid());
+    Tid socket;
+    TransId tid (n->getNewTid());
+    if (previous and previous->node == n) {
+        socket = previous->getSocket();
+    } else {
+        if (previous)
+            if (logger_)
+                logger_->e(hash, "[node %s] trying refresh listen contract with wrong node", previous->node->toString().c_str());
+        socket = n->openSocket(std::move(socket_cb));
+    }
+
+    if (not socket) {
+        if (logger_)
+            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(5 + has_query);
+    pk.pack(KEY_A); pk.pack_map(4 + 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(socketId);
+      pk.pack(KEY_REQ_SID);   pk.pack_bin(sid.size());
+                              pk.pack_bin_body((const char*)sid.data(), sid.size());
      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(tid);
+    pk.pack(KEY_TID); pk.pack_bin(tid.size());
+                               pk.pack_bin_body((const char*)tid.data(), tid.size());
    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, n,
+    auto req = std::make_shared<Request>(MessageType::Listen, tid.toInt(), n,
        Blob(buffer.data(), buffer.data() + buffer.size()),
        [=](const Request& req_status, ParsedMessage&& msg) { /* on done */
            if (on_done)
@@ -1110,7 +1095,8 @@ NetworkEngine::sendListen(Sp<Node> n,
        [=](const Request& req_status, bool done) { /* on expired */
            if (on_expired)
                on_expired(req_status, done);
-        }
+        },
+        socket
    );
    sendRequest(req);
    ++out_stats.listen;
@@ -1127,7 +1113,9 @@ NetworkEngine::sendListenConfirmation(const SockAddr& addr, Tid tid) {
      pk.pack(KEY_REQ_ID); pk.pack(myid);
      insertAddr(pk, addr);

-    pk.pack(KEY_TID); pk.pack(tid);
+    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_Y); pk.pack(KEY_R);
    pk.pack(KEY_UA); pk.pack(my_v);
    if (config.network) {
@@ -1146,7 +1134,7 @@ NetworkEngine::sendAnnounceValue(Sp<Node> n,
        RequestCb&& on_done,
        RequestExpiredCb&& on_expired)
 {
-    Tid tid (n->getNewTid());
+    TransId tid (n->getNewTid());
    msgpack::sbuffer buffer;
    msgpack::packer<msgpack::sbuffer> pk(&buffer);
    pk.pack_map(5+(config.network?1:0));
@@ -1162,14 +1150,15 @@ 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(tid);
+    pk.pack(KEY_TID); pk.pack_bin(tid.size());
+                      pk.pack_bin_body((const char*)tid.data(), tid.size());
    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, n,
+    auto req = std::make_shared<Request>(MessageType::AnnounceValue, tid.toInt(), n,
        Blob(buffer.data(), buffer.data() + buffer.size()),
        [=](const Request& req_status, ParsedMessage&& msg) { /* on done */
            if (msg.value_id == Value::INVALID_ID) {
@@ -1195,62 +1184,15 @@ 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,
                const Value::Id& vid,
                const Blob& token,
                RequestCb&& on_done,
-                RequestErrorCb&& on_error,
                RequestExpiredCb&& on_expired)
 {
-    Tid tid (n->getNewTid());
+    TransId tid (n->getNewTid());
    msgpack::sbuffer buffer;
    msgpack::packer<msgpack::sbuffer> pk(&buffer);
    pk.pack_map(5+(config.network?1:0));
@@ -1262,14 +1204,15 @@ 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(tid);
+    pk.pack(KEY_TID); pk.pack_bin(tid.size());
+                               pk.pack_bin_body((const char*)tid.data(), tid.size());
    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, n,
+    auto req = std::make_shared<Request>(MessageType::Refresh, tid.toInt(), n,
        Blob(buffer.data(), buffer.data() + buffer.size()),
        [=](const Request& req_status, ParsedMessage&& msg) { /* on done */
            if (msg.value_id == Value::INVALID_ID) {
@@ -1283,7 +1226,6 @@ NetworkEngine::sendRefreshValue(Sp<Node> n,
                }
            }
        },
-        on_error,
        [=](const Request& req_status, bool done) { /* on expired */
            if (on_expired) {
                on_expired(req_status, done);
@@ -1306,7 +1248,9 @@ NetworkEngine::sendValueAnnounced(const SockAddr& addr, Tid tid, Value::Id vid)
      pk.pack(KEY_REQ_VALUE_ID); pk.pack(vid);
      insertAddr(pk, addr);

-    pk.pack(KEY_TID); pk.pack(tid);
+    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_Y); pk.pack(KEY_R);
    pk.pack(KEY_UA); pk.pack(my_v);
    if (config.network) {
@@ -1325,7 +1269,7 @@ NetworkEngine::sendError(const SockAddr& addr,
 {
    msgpack::sbuffer buffer;
    msgpack::packer<msgpack::sbuffer> pk(&buffer);
-    pk.pack_map(4 + (include_id?1:0) + (config.network?1:0));
+    pk.pack_map(4 + (include_id?1:0));

    pk.pack(KEY_E); pk.pack_array(2);
      pk.pack(code);
@@ -1336,7 +1280,9 @@ NetworkEngine::sendError(const SockAddr& addr,
          pk.pack(KEY_REQ_ID); pk.pack(myid);
    }

-    pk.pack(KEY_TID); pk.pack(tid);
+    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_Y); pk.pack(KEY_E);
    pk.pack(KEY_UA); pk.pack(my_v);
    if (config.network) {
@@ -116,6 +116,7 @@ Node::cancelRequest(const Sp<net::Request>& req)
 {
    if (req) {
        req->cancel();
+        closeSocket(req->closeSocket());
        requests_.erase(req->getTid());
    }
 }
@@ -54,12 +54,10 @@ 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"};
@@ -115,7 +113,6 @@ 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);

@@ -227,8 +224,6 @@ 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();

@@ -300,8 +295,6 @@ 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) {
@@ -27,7 +27,6 @@ struct Node;
 namespace net {

 class NetworkEngine;
-class DhtProtocolException;
 struct ParsedMessage;

 /*!
@@ -73,19 +72,16 @@ struct Request {
            Sp<Node> node,
            Blob&& msg,
            std::function<void(const Request&, ParsedMessage&&)> on_done,
-            std::function<void(const Request&, bool)> on_expired) :
-        node(node), tid(tid), type(type), on_done(on_done), on_expired(on_expired), msg(std::move(msg)) { }
-    Request(MessageType type, Tid tid,
-            Sp<Node> node,
-            Blob&& msg,
-            std::function<void(const Request&, ParsedMessage&&)> on_done,
-            std::function<bool(const Request&, DhtProtocolException&&)> on_error,
-            std::function<void(const Request&, bool)> on_expired) :
-        node(node), tid(tid), type(type), on_done(on_done), on_error(on_error), on_expired(on_expired), msg(std::move(msg)) { }
+            std::function<void(const Request&, bool)> on_expired,
+            Tid socket = 0) :
+        node(node), tid(tid), type(type), on_done(on_done), on_expired(on_expired), msg(std::move(msg)), socket(socket) { }

    Tid getTid() const { return tid; }
    MessageType getType() const { return type; }

+    Tid getSocket() { return socket; }
+    Tid closeSocket() { auto ret = socket; socket = 0; return ret; }
+
    void setExpired() {
        if (pending()) {
            state_ = Request::State::EXPIRED;
@@ -100,15 +96,6 @@ struct Request {
            clear();
        }
    }
-    bool setError(DhtProtocolException&& e) {
-        if (pending()) {
-            state_ = Request::State::EXPIRED;
-            bool handled = on_error and on_error(*this, std::forward<DhtProtocolException>(e));
-            clear();
-            return handled;
-        }
-        return true;
-    }

    void cancel() {
        if (pending()) {
@@ -126,7 +113,6 @@ private:

    void clear() {
        on_done = {};
-        on_error = {};
        on_expired = {};
        msg = {};
        parts = {};
@@ -142,11 +128,11 @@ private:
    time_point last_try {time_point::min()};   /* time of the last attempt to process the request. */

    std::function<void(const Request&, ParsedMessage&&)> on_done {};
-    std::function<bool(const Request&, DhtProtocolException&&)> on_error {};
    std::function<void(const Request&, bool)> on_expired {};

    Blob msg {};                      /* the serialized message. */
    std::vector<Blob> parts;
+    Tid socket;   /* the socket used for further reponses. */
 };

 } /* namespace net  */
@@ -65,16 +65,10 @@ struct Dht::SearchNode {
        ValueCache cache;
        Sp<Scheduler::Job> cacheExpirationJob {};
        Sp<net::Request> req {};
-        Tid socketId {0};
-        CachedListenStatus(ValueStateCallback&& cb, SyncCallback scb, Tid sid)
-         : cache(std::forward<ValueStateCallback>(cb), std::forward<SyncCallback>(scb)), socketId(sid) {}
+        CachedListenStatus(ValueStateCallback&& cb, SyncCallback&& scb)
+         : cache(std::forward<ValueStateCallback>(cb), std::forward<SyncCallback>(scb)) {}
        CachedListenStatus(CachedListenStatus&&) = delete;
        CachedListenStatus(const CachedListenStatus&) = delete;
-        ~CachedListenStatus() {
-            if (socketId and req and req->node) {
-                req->node->closeSocket(socketId);
-            }
-        }
    };
    using NodeListenerStatus = std::map<Sp<Query>, CachedListenStatus>;

@@ -350,15 +344,13 @@ struct Dht::SearchNode {
     * Assuming the node is synced, should the "listen" request with Query q be
     * sent to this node now ?
     */
-    time_point getListenTime(const decltype(listenStatus)::const_iterator listen_status, duration listen_expire) const {
-        if (listen_status == listenStatus.cend() or not listen_status->second.req)
+    time_point getListenTime(const Sp<Query>& q, duration listen_expire) const {
+        auto listen_status = listenStatus.find(q);
+        if (listen_status == listenStatus.end() or not listen_status->second.req)
            return time_point::min();
        return listen_status->second.req->pending() ? time_point::max() :
            listen_status->second.req->reply_time + listen_expire - REANNOUNCE_MARGIN;
    }
-    time_point getListenTime(const Sp<Query>& q, duration listen_expire) const {
-        return getListenTime(listenStatus.find(q), listen_expire);
-    }

    /**
     * Is this node expired or candidate
@@ -277,16 +277,28 @@ unpackMsg(Blob b) {
 }

 msgpack::object*
-findMapValue(const msgpack::object& map, const char* key, size_t key_length) {
+findMapValue(const msgpack::object& map, const char* 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_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;
+}
+
 }
@@ -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);

-    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);
+    serverProxy = std::make_unique<dht::DhtProxyServer>(
+            //dht::crypto::Identity{}, // http
+            serverIdentity,            // https
+            nodeProxy, 8080, /*pushServer*/"127.0.0.1:8090");

    clientConfig.server_ca = serverCAIdentity.second;
    clientConfig.client_identity = dht::crypto::generateIdentity("DhtProxyTester");
@@ -167,51 +167,4 @@ 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
@@ -32,7 +32,6 @@ class DhtRunnerTester : public CppUnit::TestFixture {
    CPPUNIT_TEST(testConstructors);
    CPPUNIT_TEST(testGetPut);
    CPPUNIT_TEST(testListen);
-    CPPUNIT_TEST(testListenLotOfBytes);
    CPPUNIT_TEST_SUITE_END();

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

 }  // namespace test
@@ -39,10 +39,9 @@ HttpTester::setUp() {
    nodeProxy->run(0, /*identity*/{}, /*threaded*/true);
    nodeProxy->bootstrap(nodePeer->getBound());

-    dht::ProxyServerConfig config;
-    config.port = 8080;
-    config.pushServer = "127.0.0.1:8090";
-    serverProxy = std::make_unique<dht::DhtProxyServer>(nodeProxy, config);
+    serverProxy = std::unique_ptr<dht::DhtProxyServer>(
+        new dht::DhtProxyServer(
+            /*http*/dht::crypto::Identity{}, nodeProxy, 8080, /*pushServer*/"127.0.0.1:8090"));
 }

 void
@@ -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.empty())
-            dht.bootstrap(params.bootstrap);
+        if (not params.bootstrap.first.empty())
+            dht.bootstrap(params.bootstrap.first.c_str(), params.bootstrap.second.c_str());

-        print_node_info(dht.getNodeInfo());
+        print_node_info(dht, params);
        std::cout << "  type 'c {hash}' to join a channel" << std::endl << std::endl;

        bool connected {false};
@@ -114,7 +114,7 @@ void cmd_loop(std::shared_ptr<DhtRunner>& node, dht_params& params
 #endif
 )
 {
-    print_node_info(node->getNodeInfo());
+    print_node_info(*node, params);
    std::cout << " (type 'h' or 'help' for a list of possible commands)" << std::endl << std::endl;

 #ifndef WIN32_NATIVE
@@ -143,23 +143,22 @@ void cmd_loop(std::shared_ptr<DhtRunner>& node, dht_params& params
            print_help();
            continue;
        } else if (op == "ll") {
-            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;
+            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;
 #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;
@@ -191,10 +190,14 @@ void cmd_loop(std::shared_ptr<DhtRunner>& node, dht_params& params
            continue;
        } else if (op == "b") {
            iss >> idstr;
-            if (not idstr.empty())
-                node->bootstrap(idstr);
-            else
-                std::cerr << "No service provided. Syntax: b hostname:port" << std::endl;
+            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;
+            }
            continue;
        } else if (op == "log") {
            iss >> idstr;
@@ -218,11 +221,14 @@ void cmd_loop(std::shared_ptr<DhtRunner>& node, dht_params& params
                iss >> idstr;
 #endif // OPENDHT_PUSH_NOTIFICATIONS
            try {
-                in_port_t port = std::stoi(idstr);
-                ProxyServerConfig serverConfig;
-                serverConfig.port = port;
-                serverConfig.pushServer = pushServer;
-                proxies.emplace(port, std::make_unique<DhtProxyServer>(node, serverConfig));
+                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
+                )));
            } catch (...) { }
            continue;
        } else if (op == "psx") {
@@ -233,12 +239,13 @@ 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){
-                    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));
+                    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
+                    )));
                }
                else {
                    std::cerr << "Missing Identity private key or certificate" << std::endl;
@@ -530,32 +537,25 @@ main(int argc, char **argv)
        auto dhtConf = getDhtConfig(params);
        node->run(params.port, dhtConf.first, std::move(dhtConf.second));

-        if (not params.bootstrap.empty()) {
-            std::cout << "Bootstrap: " << params.bootstrap << std::endl;
-            node->bootstrap(params.bootstrap);
+        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());
        }

 #ifdef OPENDHT_PROXY_SERVER
        std::map<in_port_t, std::unique_ptr<DhtProxyServer>> proxies;
 #endif
-        if (params.proxyserver or params.proxyserverssl) {
+        if (params.proxyserverssl and params.proxy_id.first and params.proxy_id.second){
 #ifdef OPENDHT_PROXY_SERVER
-            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));
-            }
+            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)));
 #else
            std::cerr << "DHT proxy server requested but OpenDHT built without proxy server support." << std::endl;
            exit(EXIT_FAILURE);
@@ -7,7 +7,7 @@ stats_total = {"users":0, "pushListenersCount":0, "listenCount":0, "totalListene

 for i in range(80,101):
    print("Collecting stats for proxy " + str(i))
-    response = requests.request('STATS', 'http://127.0.0.1:' + str(i))
+    response = requests.request('STATS', 'http://dhtproxy.jami.net:' + str(i))

    if response.status_code == 200:
        result = response.json()
@@ -38,17 +38,20 @@ for i in range(80,101):
        except:
            pass
        try:
-            stats['totalPermanentPuts'] = int(result["totalPermanentPuts"])
-            stats_total['totalPermanentPuts'] += int(result["totalPermanentPuts"])
+            total = 0
+            for h,v in result["puts"].items():
+                total += int(v)
+            stats['totalPermanentPuts'] = total
+            stats_total['totalPermanentPuts'] += total
        except:
            pass
-
+        
        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", "w") as stat_file:
+with open("stats_proxy_total", "a") as stat_file:
    stat_file.write(str(stats_total))
    stat_file.write('\n')
@@ -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.empty())
-            dht.bootstrap(params.bootstrap);
+        if (not params.bootstrap.first.empty())
+            dht.bootstrap(params.bootstrap.first.c_str(), params.bootstrap.second.c_str());

-        print_node_info(dht.getNodeInfo());
+        print_node_info(dht, params);
        std::cout << "Scanning network..." << std::endl;
        auto all_nodes = std::make_shared<NodeSet>();

@@ -124,7 +124,7 @@ struct dht_params {
    bool log {false};
    bool syslog {false};
    std::string logfile {};
-    std::string bootstrap {};
+    std::pair<std::string, std::string> bootstrap {};
    dht::NetId network {0};
    in_port_t port {0};
    in_port_t proxyserver {0};
@@ -191,14 +191,16 @@ getDhtConfig(dht_params& params)
    return {std::move(config), std::move(context)};
 }

-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;
+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;
    else
-        std::cout << "IPv4 port " << info.bound4 << ", IPv6 port " << info.bound6 << std::endl;
-    if (info.id)
-        std::cout << "Public key ID " << info.id << std::endl;
+        std::cout << "IPv4 port " << port4 << ", IPv6 port " << port6 << std::endl;
+    if (params.id.first)
+        std::cout << "Public key ID " << node.getId() << std::endl;
 }

 static const constexpr struct option long_options[] = {
@@ -289,7 +291,10 @@ parseArgs(int argc, char **argv) {
            params.public_stable = true;
            break;
        case 'b':
-            params.bootstrap = (optarg[0] == '=') ? optarg+1 : optarg;
+            params.bootstrap = dht::splitPort((optarg[0] == '=') ? optarg+1 : optarg);
+            if (not params.bootstrap.first.empty() and params.bootstrap.second.empty()) {
+                params.bootstrap.second = std::to_string(dht::net::DHT_DEFAULT_PORT);
+            }
            break;
        case 'V':
            params.version = true;
Author	SHA1	Message	Date
Adrien Béraud	b0536d9e02	search: expire only after candidate expiration	2020-04-24 13:29:12 -04:00
Adrien Béraud	60a4f369fa	move bootstrap logic from DhtRunner to Dht	2020-04-24 13:29:12 -04:00