wip

2026-11-11 13:57:04 +00:00 · 2018-04-21 17:16:37 -04:00
33 changed files with 538 additions and 1224 deletions
@@ -37,7 +37,7 @@ script:
    # proxy builds
    if [[ "$OPENDHT_TEST_JOB" != *"opendht.llvm"* ]] && [[ "$OPENDHT_TEST_JOB" != *"opendht.classic"* ]]; then
      docker build -t opendht-proxy -f docker/DockerfileTravisProxy .;
-      options='-DOPENDHT_SANITIZE=On ';
+      options='';
      if [[ "$OPENDHT_TEST_JOB" == *"opendht.proxyserver"* ]] || [[ "$OPENDHT_TEST_JOB" == *"opendht.push"* ]]; then
        options+='-DOPENDHT_PROXY_SERVER=ON ';
      else
@@ -1,7 +1,7 @@
 cmake_minimum_required (VERSION 3.1)
 project (opendht)
 set (opendht_VERSION_MAJOR 1)
-set (opendht_VERSION_MINOR 7.4)
+set (opendht_VERSION_MINOR 7.1)
 set (opendht_VERSION ${opendht_VERSION_MAJOR}.${opendht_VERSION_MINOR})
 set (PACKAGE_VERSION ${opendht_VERSION})
 set (VERSION "${opendht_VERSION}")
@@ -47,21 +47,9 @@ if (NOT OPENDHT_ARGON2)
        set(OPENDHT_ARGON2 ON)
    endif()
 endif ()
-
-pkg_search_module(Jsoncpp jsoncpp)
-if (Jsoncpp_FOUND)
-    add_definitions(-DOPENDHT_JSONCPP)
-    list (APPEND opendht_SOURCES
-      src/base64.h
-      src/base64.cpp
-    )
-endif()
-
 if (OPENDHT_PROXY_SERVER OR OPENDHT_PROXY_CLIENT)
    find_package(Restbed REQUIRED)
-    if (!Jsoncpp_FOUND)
-        message(SEND_ERROR "Jsoncpp is required for DHT proxy support")
-    endif()
+    pkg_search_module(Jsoncpp jsoncpp)
 endif()

 # Build flags
@@ -194,6 +182,11 @@ if (OPENDHT_PROXY_SERVER)
  list (APPEND opendht_SOURCES
    src/dht_proxy_server.cpp
  )
+  if (OPENDHT_PUSH_NOTIFICATIONS)
+    add_definitions(-DOPENDHT_PUSH_NOTIFICATIONS=true)
+  else ()
+    add_definitions(-DOPENDHT_PUSH_NOTIFICATIONS=false)
+  endif ()
 else ()
    add_definitions(-DENABLE_PROXY_SERVER=false)
 endif ()
@@ -211,17 +204,16 @@ else ()
 endif ()

 if (OPENDHT_PROXY_SERVER OR OPENDHT_PROXY_CLIENT)
-  if (OPENDHT_PUSH_NOTIFICATIONS)
-    message("Using push notification")
-    add_definitions(-DOPENDHT_PUSH_NOTIFICATIONS=true)
-  else ()
-    add_definitions(-DOPENDHT_PUSH_NOTIFICATIONS=false)
-  endif ()
  list (APPEND opendht_HEADERS
    include/opendht/proxy.h
  )
+  list (APPEND opendht_SOURCES
+    src/base64.h
+    src/base64.cpp
+  )
 endif ()

+
 if(OPENDHT_ARGON2)
    # make sure argon2 submodule is up to date and initialized
    message("Initializing Argon2 submodule")
@@ -312,23 +304,12 @@ install (FILES ${CMAKE_CURRENT_BINARY_DIR}/opendhtConfigVersion.cmake DESTINATIO
 IF(OPENDHT_TESTS)
  FIND_PACKAGE(Cppunit REQUIRED)
  # unit testing
-    list (APPEND test_FILES
-      tests/infohashtester.h
-      tests/infohashtester.cpp
-      tests/cryptotester.h
-      tests/cryptotester.cpp
-      tests/dhtrunnertester.h
-      tests/dhtrunnertester.cpp
-    )
-    if (OPENDHT_PROXY_SERVER AND OPENDHT_PROXY_CLIENT)
-      list (APPEND test_FILES
-        tests/dhtproxytester.h
-        tests/dhtproxytester.cpp
-      )
-    endif()
    add_executable(opendht_unit_tests
      tests/tests_runner.cpp
-      ${test_FILES}
+      tests/infohashtester.h
+      tests/infohashtester.cpp
+      tests/dhtrunnertester.h
+      tests/dhtrunnertester.cpp
    )
    if (OPENDHT_SHARED)
      TARGET_LINK_LIBRARIES(opendht_unit_tests opendht)
@@ -146,7 +146,7 @@
      <Optimization>Disabled</Optimization>
      <SDLCheck>true</SDLCheck>
      <AdditionalIncludeDirectories>..\..\include;..\include;..\include\opendht;..\..\msgpack-c\include;</AdditionalIncludeDirectories>
-      <PreprocessorDefinitions>_CRT_SECURE_NO_WARNINGS;WIN32_LEAN_AND_MEAN;_MBCS;%(PreprocessorDefinitions)</PreprocessorDefinitions>
+      <PreprocessorDefinitions>_CRT_SECURE_NO_WARNINGS;WIN32_NATIVE;WIN32_LEAN_AND_MEAN;_MBCS;%(PreprocessorDefinitions)</PreprocessorDefinitions>
      <DisableSpecificWarnings>4804;4800;4101;4267;4244;4503;</DisableSpecificWarnings>
      <AdditionalOptions>-D_SCL_SECURE_NO_WARNINGS %(AdditionalOptions)</AdditionalOptions>
      <SuppressStartupBanner>false</SuppressStartupBanner>
@@ -169,7 +169,7 @@ copy ..\include\opendht.h $(OutDir)\include\</Command>
      <Optimization>Disabled</Optimization>
      <SDLCheck>true</SDLCheck>
      <AdditionalIncludeDirectories>..\..\include;..\include;..\include\opendht;..\..\msgpack-c\include;</AdditionalIncludeDirectories>
-      <PreprocessorDefinitions>_CRT_SECURE_NO_WARNINGS;WIN32_LEAN_AND_MEAN;_MBCS;%(PreprocessorDefinitions)</PreprocessorDefinitions>
+      <PreprocessorDefinitions>_CRT_SECURE_NO_WARNINGS;WIN32_NATIVE;WIN32_LEAN_AND_MEAN;_MBCS;%(PreprocessorDefinitions)</PreprocessorDefinitions>
      <DisableSpecificWarnings>4804;4800;4101;4267;4244;4503;</DisableSpecificWarnings>
      <AdditionalOptions>-D_SCL_SECURE_NO_WARNINGS %(AdditionalOptions)</AdditionalOptions>
    </ClCompile>
@@ -193,7 +193,7 @@ copy ..\include\opendht.h $(OutDir)\include\</Command>
      <IntrinsicFunctions>true</IntrinsicFunctions>
      <SDLCheck>true</SDLCheck>
      <AdditionalIncludeDirectories>..\..\include;..\include;..\include\opendht;..\..\msgpack-c\include;</AdditionalIncludeDirectories>
-      <PreprocessorDefinitions>_CRT_SECURE_NO_WARNINGS;WIN32_LEAN_AND_MEAN;_MBCS;%(PreprocessorDefinitions)</PreprocessorDefinitions>
+      <PreprocessorDefinitions>_CRT_SECURE_NO_WARNINGS;WIN32_NATIVE;WIN32_LEAN_AND_MEAN;_MBCS;%(PreprocessorDefinitions)</PreprocessorDefinitions>
      <DisableSpecificWarnings>4804;4800;4101;4267;4244;4503;</DisableSpecificWarnings>
      <AdditionalOptions>-D_SCL_SECURE_NO_WARNINGS %(AdditionalOptions)</AdditionalOptions>
      <ProgramDataBaseFileName>$(OutDir)\lib\x86\$(TargetName).pdb</ProgramDataBaseFileName>
@@ -223,7 +223,7 @@ copy ..\include\opendht.h $(OutDir)\include\</Command>
      <IntrinsicFunctions>true</IntrinsicFunctions>
      <SDLCheck>true</SDLCheck>
      <AdditionalIncludeDirectories>$(ProjectDir)contrib\build\include;$(ProjectDir)..\include;$(ProjectDir)..\include\opendht;$(ProjectDir)contrib\build\msgpack-c\include;$(ProjectDir)contrib\build\argon2\include;$(ProjectDir)..\..\include;$(ProjectDir)..\..\argon2\include</AdditionalIncludeDirectories>
-      <PreprocessorDefinitions>_CRT_SECURE_NO_WARNINGS;WIN32_LEAN_AND_MEAN;_MBCS;%(PreprocessorDefinitions)</PreprocessorDefinitions>
+      <PreprocessorDefinitions>_CRT_SECURE_NO_WARNINGS;WIN32_NATIVE;WIN32_LEAN_AND_MEAN;_MBCS;%(PreprocessorDefinitions)</PreprocessorDefinitions>
      <DisableSpecificWarnings>4804;4800;4101;4267;4244;4503;4273;</DisableSpecificWarnings>
      <AdditionalOptions>-D_SCL_SECURE_NO_WARNINGS %(AdditionalOptions)</AdditionalOptions>
      <ProgramDataBaseFileName>$(OutDir)\lib\x64\$(TargetName).pdb</ProgramDataBaseFileName>
@@ -1,7 +1,7 @@
 dnl define macros
 m4_define([opendht_major_version], 1)
 m4_define([opendht_minor_version], 7)
-m4_define([opendht_patch_version], 4)
+m4_define([opendht_patch_version], 1)
 m4_define([opendht_version],
 		  [opendht_major_version.opendht_minor_version.opendht_patch_version])

@@ -101,51 +101,9 @@ LT_LANG(C++)
 AX_CXX_COMPILE_STDCXX(11,[noext],[mandatory])

 PKG_PROG_PKG_CONFIG()
-
-AC_ARG_ENABLE([proxy_server], AS_HELP_STRING([--enable-proxy-server], [Enable proxy server ability]), proxy_server=yes, proxy_server=no)
-AM_CONDITIONAL(ENABLE_PROXY_SERVER, test x$proxy_server == xyes)
-
-AC_ARG_ENABLE([push_notifications], AS_HELP_STRING([--enable-push-notifications], [Enable push notifications support]), push_notifications=yes, push_notifications=no)
-AM_CONDITIONAL(ENABLE_PUSH_NOTIFICATIONS, test x$push_notifications == xyes)
-
-AC_ARG_ENABLE([proxy_server_identity], AS_HELP_STRING([--enable-proxy-server-identity],
-	[Enable proxy server ability]), proxy_server_identity=yes, proxy_server_identity=no)
-AM_CONDITIONAL(ENABLE_PROXY_SERVER_IDENTITY, test x$proxy_server_identity == xyes -a x$proxy_server == xyes)
-AC_ARG_ENABLE([proxy_server_identity], AS_HELP_STRING([--enable-proxy-server-identity],
-	[Enable proxy server ability]), proxy_server_identity=yes, proxy_server_identity=no)
-
-AC_ARG_ENABLE([proxy_client], AS_HELP_STRING([--enable-proxy-client], [Enable proxy client ability]), proxy_client=yes, proxy_client=no)
-AM_CONDITIONAL(ENABLE_PROXY_CLIENT, test x$proxy_client == xyes)
-
-AC_ARG_ENABLE([tests], AS_HELP_STRING([--enable-tests], [Enable tests]), build_tests=yes, build_tests=no)
-AM_CONDITIONAL(ENABLE_TESTS, test x$build_tests == xyes)
-AM_COND_IF([ENABLE_TESTS], [
-	PKG_CHECK_MODULES([CppUnit], [cppunit >= 1.12])
-])
-
-AM_CONDITIONAL(PROXY_CLIENT_OR_SERVER, test x$proxy_client == xyes | test x$proxy_server == xyes)
-
 PKG_CHECK_MODULES([Nettle], [nettle >= 2.4])
 PKG_CHECK_MODULES([GnuTLS], [gnutls >= 3.3])
 PKG_CHECK_MODULES([MsgPack], [msgpack >= 1.2])
-
-AC_ARG_WITH([jsoncpp], AS_HELP_STRING([--without-jsoncpp], [Build without JsonCpp support]))
-AS_IF([test "x$with_jsoncpp" != "xno"], 
-      [PKG_CHECK_MODULES([JsonCpp], [jsoncpp >= 1.7.2], [have_jsoncpp=yes], [have_jsoncpp=no])],
-      [have_jsoncpp=no])
-AS_IF([test "x$have_jsoncpp" = "xyes"], [
-    AC_MSG_NOTICE([Using JsonCpp])
-    CPPFLAGS+=" -DOPENDHT_JSONCPP=1"
-], [
-    AC_MSG_NOTICE([Not using JsonCpp])
-    AM_COND_IF(PROXY_CLIENT_OR_SERVER, AC_MSG_ERROR(["JsonCpp is required for proxy/push notification support"]))
-])
-
-AM_COND_IF([PROXY_CLIENT_OR_SERVER], [
-	AC_CHECK_LIB(restbed, exit,, AC_MSG_ERROR([Missing restbed files]))
-	LDFLAGS="${LDFLAGS} -lrestbed"
-])
-
 CXXFLAGS="${CXXFLAGS} -DMSGPACK_DISABLE_LEGACY_NIL -DMSGPACK_DISABLE_LEGACY_CONVERT"

 dnl Check for Argon2
@@ -177,21 +135,58 @@ AM_COND_IF([ENABLE_TOOLS], [
  ])
 ])

-AM_COND_IF(ENABLE_PROXY_SERVER,
-    [CPPFLAGS+=" -DOPENDHT_PROXY_SERVER=true"],
-    [CPPFLAGS+=" -DOPENDHT_PROXY_SERVER=false"])
+AC_ARG_ENABLE([proxy_server], AS_HELP_STRING([--enable-proxy-server], [Enable proxy server ability]), proxy_server=yes, proxy_server=no)
+AM_CONDITIONAL(ENABLE_PROXY_SERVER, test x$proxy_server == xyes)
+AC_ARG_ENABLE([push_notifications], AS_HELP_STRING([--enable-push-notifications], [Enable push notifications support]), push_notifications=yes, push_notifications=no)
+AM_CONDITIONAL(ENABLE_PUSH_NOTIFICATIONS, test x$push_notifications == xyes)
+AC_ARG_ENABLE([proxy_server_identity], AS_HELP_STRING([--enable-proxy-server-identity],
+	[Enable proxy server ability]), proxy_server_identity=yes, proxy_server_identity=no)
+AM_CONDITIONAL(ENABLE_PROXY_SERVER_IDENTITY, test x$proxy_server_identity == xyes)
+AC_ARG_ENABLE([proxy_client], AS_HELP_STRING([--enable-proxy-client], [Enable proxy client ability]), proxy_client=yes, proxy_client=no)
+AM_CONDITIONAL(ENABLE_PROXY_CLIENT, test x$proxy_client == xyes)

-AM_COND_IF(ENABLE_PROXY_CLIENT,
-    [CPPFLAGS+=" -DOPENDHT_PROXY_CLIENT=true"],
-    [CPPFLAGS+=" -DOPENDHT_PROXY_CLIENT=false"])
+AM_COND_IF([ENABLE_PROXY_SERVER], [
+	AC_CHECK_LIB(restbed, exit,, AC_MSG_ERROR([Missing restbed files]))
+	PKG_CHECK_MODULES([Jsoncpp], [jsoncpp >= 1.7.2])
+	CPPFLAGS+=" -DOPENDHT_PROXY_SERVER=true -ljsoncpp -lrestbed"
+], [
+	CPPFLAGS+=" -DOPENDHT_PROXY_SERVER=false"
+])

-AM_COND_IF(ENABLE_PUSH_NOTIFICATIONS,
-    [CPPFLAGS+=" -DOPENDHT_PUSH_NOTIFICATIONS=true"],
-    [CPPFLAGS+=" -DOPENDHT_PUSH_NOTIFICATIONS=false"])
+AM_COND_IF([ENABLE_PUSH_NOTIFICATIONS], [
+	CPPFLAGS+=" -DOPENDHT_PUSH_NOTIFICATIONS=true"
+], [
+	CPPFLAGS+=" -DOPENDHT_PUSH_NOTIFICATIONS=false"
+])

-AM_COND_IF(ENABLE_PROXY_SERVER_IDENTITY,
-    [CPPFLAGS+=" -DOPENDHT_PROXY_SERVER_IDENTITY=true"],
-    [CPPFLAGS+=" -DOPENDHT_PROXY_SERVER_IDENTITY=false"])
+AC_ARG_ENABLE([proxy_server_identity], AS_HELP_STRING([--enable-proxy-server-identity],
+    [Enable proxy server ability]), proxy_server_identity=yes, proxy_server_identity=no)
+AM_CONDITIONAL(ENABLE_PROXY_SERVER_IDENTITY, test x$proxy_server_identity == xyes -a x$proxy_server == xyes)
+AM_COND_IF([ENABLE_PROXY_SERVER_IDENTITY], [
+    CPPFLAGS+=" -DOPENDHT_PROXY_SERVER_IDENTITY=true"
+], [
+    CPPFLAGS+=" -DOPENDHT_PROXY_SERVER_IDENTITY=false"
+])
+
+AM_COND_IF([ENABLE_PROXY_CLIENT], [
+	CPPFLAGS+=" -DOPENDHT_PROXY_CLIENT=true"
+], [
+	CPPFLAGS+=" -DOPENDHT_PROXY_CLIENT=false"
+])
+
+AM_CONDITIONAL(PROXY_CLIENT_OR_SERVER, test x$proxy_client == xyes | test x$proxy_server == xyes)
+AM_COND_IF([PROXY_CLIENT_OR_SERVER], [
+	AC_CHECK_LIB(restbed, exit,, AC_MSG_ERROR([Missing restbed files]))
+	PKG_CHECK_MODULES([Jsoncpp], [jsoncpp >= 1.7.4])
+	CPPFLAGS="${CPPFLAGS} ${Jsoncpp_CFLAGS}"
+	LDFLAGS="${LDFLAGS} ${Jsoncpp_LIBS} -lrestbed"
+], [])
+
+AC_ARG_ENABLE([tests], AS_HELP_STRING([--enable-tests], [Enable tests]), build_tests=yes, build_tests=no)
+AM_CONDITIONAL(ENABLE_TESTS, test x$build_tests == xyes)
+AM_COND_IF([ENABLE_TESTS], [
+	PKG_CHECK_MODULES([CppUnit], [cppunit >= 1.12])
+])

 AC_CONFIG_FILES([doc/Doxyfile doc/Makefile])

@@ -207,6 +202,6 @@ AM_COND_IF([USE_CYTHON], [
 AC_CONFIG_FILES([Makefile
                 src/Makefile
                 tools/Makefile
-                 tests/Makefile
+								 tests/Makefile
                 opendht.pc])
 AC_OUTPUT
@@ -27,9 +27,9 @@
 #include <memory>
 #include <functional>

-#ifdef OPENDHT_JSONCPP
+#if OPENDHT_PROXY_SERVER
 #include <json/json.h>
-#endif
+#endif //OPENDHT_PROXY_SERVER

 namespace dht {

@@ -51,37 +51,15 @@ struct OPENDHT_PUBLIC NodeStats {
             incoming_nodes {0};
    unsigned table_depth {0};
    unsigned getKnownNodes() const { return good_nodes + dubious_nodes; }
-    unsigned long getNetworkSizeEstimation() const { return 8 * std::exp2(table_depth); }
    std::string toString() const;
-
-#ifdef OPENDHT_JSONCPP
+#if OPENDHT_PROXY_SERVER || OPENDHT_PROXY_CLIENT
    /**
     * Build a json object from a NodeStats
     */
    Json::Value toJson() const;
    NodeStats() {};
    explicit NodeStats(const Json::Value& v);
-#endif
-
-    MSGPACK_DEFINE_MAP(good_nodes, dubious_nodes, cached_nodes, incoming_nodes, table_depth)
-};
-
-struct OPENDHT_PUBLIC NodeInfo {
-    InfoHash id;
-    InfoHash node_id;
-    NodeStats ipv4;
-    NodeStats ipv6;
-
-#ifdef OPENDHT_JSONCPP
-    /**
-     * Build a json object from a NodeStats
-     */
-    Json::Value toJson() const;
-    NodeInfo() {};
-    explicit NodeInfo(const Json::Value& v);
-#endif
-
-    MSGPACK_DEFINE_MAP(id, node_id, ipv4, ipv6)
+#endif //OPENDHT_PROXY_SERVER
 };

 /**
@@ -114,8 +114,6 @@ public:
    ImMessage() {}
    ImMessage(dht::Value::Id id, std::string&& m, long d = 0)
        : id(id), msg(std::move(m)), date(d) {}
-    ImMessage(dht::Value::Id id, std::string&& dt, std::string&& m, long d = 0)
-        : id(id), msg(std::move(m)), datatype(std::move(dt)), date(d) {}

    virtual void unpackValue(const Value& v) override {
        to = v.recipient;
@@ -125,11 +123,10 @@ public:
    dht::InfoHash to;
    dht::Value::Id id {0};
    std::string msg;
-    std::string datatype;
    long date {0};
    ImStatus status {ImStatus::NONE};

-    MSGPACK_DEFINE_MAP(id, msg, date, status, datatype)
+    MSGPACK_DEFINE_MAP(id, msg, date, status)
 };

 class OPENDHT_PUBLIC TrustRequest : public EncryptedValue<TrustRequest>
@@ -409,8 +409,7 @@ private:

    // Storage
    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 storageStore(const InfoHash& id, const Sp<Value>& value, time_point created, const SockAddr& sa = {});
    void expireStore();
    void expireStorage(InfoHash h);
    void expireStore(decltype(store)::iterator);
@@ -269,7 +269,6 @@ private:
     * Get informations from the proxy node
     * @return the JSON returned by the proxy
     */
-    struct InfoState;
    void getProxyInfos();
    void onProxyInfos(const Json::Value& val, sa_family_t family);
    SockAddr parsePublicAddress(const Json::Value& val);
@@ -279,10 +278,6 @@ private:
    size_t doListen(const InfoHash& key, ValueCallback, Value::Filter);
    bool doCancelListen(const InfoHash& key, size_t token);

-    struct ListenState;
-    void sendListen(const std::shared_ptr<restbed::Request>& request, const ValueCallback&, const Value::Filter& filter, const Sp<ListenState>& state);
-    void sendSubscribe(const std::shared_ptr<restbed::Request>& request, const Sp<proxy::ListenToken>&, const Sp<ListenState>& state);
-
    void doPut(const InfoHash&, Sp<Value>, DoneCallback, time_point created, bool permanent);

    /**
@@ -301,7 +296,7 @@ private:
    std::string serverHost_;
    std::string pushClientId_;

-    mutable std::mutex lockCurrentProxyInfos_;
+    std::atomic_flag ongoingStatusUpdate_ = ATOMIC_FLAG_INIT;
    NodeStatus statusIpv4_ {NodeStatus::Disconnected};
    NodeStatus statusIpv6_ {NodeStatus::Disconnected};
    NodeStats stats4_ {};
@@ -341,9 +336,8 @@ private:
    std::vector<std::function<void()>> callbacks_;
    std::mutex lockCallbacks;

-    Sp<InfoState> infoState_;
    std::thread statusThread_;
-    mutable std::mutex statusLock_;
+    mutable std::mutex lockCurrentProxyInfos_;

    Scheduler scheduler;
    /**
@@ -21,23 +21,18 @@

 #pragma once

-#include "callbacks.h"
 #include "def.h"
-#include "infohash.h"
-#include "proxy.h"
-#include "scheduler.h"
 #include "sockaddr.h"
+#include "infohash.h"
+#include "scheduler.h"
 #include "value.h"
+#include "proxy.h"

 #include <thread>
 #include <memory>
 #include <mutex>
 #include <restbed>

-#ifdef OPENDHT_JSONCPP
-#include <json/json.h>
-#endif
-
 namespace Json {
    class Value;
 }
@@ -77,44 +72,15 @@ public:
        size_t pushListenersCount;
        /** Average requests per second */
        double requestRate;
-        /** Node Info **/
-        NodeInfo nodeInfo;

        std::string toString() const {
            std::ostringstream ss;
            ss << "Listens: " << listenCount << " Puts: " << putCount << " PushListeners: " << pushListenersCount << std::endl;
            ss << "Requests: " << requestRate << " per second." << std::endl;
-            auto& ni = nodeInfo;
-            auto& ipv4 = ni.ipv4;
-            if (ipv4.table_depth > 1) {
-                ss << "IPv4 Network estimation: " << ipv4.getNetworkSizeEstimation() << std::endl;;
-            }
-            auto& ipv6 = ni.ipv6;
-            if (ipv6.table_depth > 1) {
-                ss << "IPv6 Network estimation: " << ipv6.getNetworkSizeEstimation() << std::endl;;
-            }
            return ss.str();
        }
-
-#ifdef OPENDHT_JSONCPP
-        /**
-         * Build a json object from a NodeStats
-         */
-        Json::Value toJson() const {
-            Json::Value result;
-            result["listenCount"] = static_cast<Json::UInt64>(listenCount);
-            result["putCount"] = static_cast<Json::UInt64>(putCount);
-            result["pushListenersCount"] = static_cast<Json::UInt64>(pushListenersCount);
-            result["requestRate"] = requestRate;
-            result["nodeInfo"] = nodeInfo.toJson();
-            return result;
-        }
-#endif
    };
-
-    ServerStats stats() const { return stats_; }
-
-    void updateStats() const;
+    ServerStats getStats() const;

    std::shared_ptr<DhtRunner> getNode() const { return dht_; }

@@ -133,14 +99,6 @@ private:
     */
    void getNodeInfo(const std::shared_ptr<restbed::Session>& session) const;

-    /**
-     * Return ServerStats in JSON format
-     * Method: STATS "/"
-     * Result: HTTP 200, body: Node infos in JSON format
-     * @param session
-     */
-    void getStats(const std::shared_ptr<restbed::Session>& session) const;
-
    /**
     * Return Values of an infoHash
     * Method: GET "/{InfoHash: .*}"
@@ -279,8 +237,6 @@ private:
    std::thread schedulerThread_;

    Sp<Scheduler::Job> printStatsJob_;
-    mutable std::mutex statsMutex_;
-    mutable NodeInfo nodeInfo_ {};

    // Handle client quit for listen.
    // NOTE: can be simplified when we will supports restbed 5.0
@@ -303,8 +259,6 @@ private:

    const std::string pushServer_;

-    mutable ServerStats stats_;
-
 #if OPENDHT_PUSH_NOTIFICATIONS
    struct Listener;
    struct PushListener;
@@ -293,7 +293,6 @@ public:

    NodeStats getNodesStats(sa_family_t af) const;
    unsigned getNodesStats(sa_family_t af, unsigned *good_return, unsigned *dubious_return, unsigned *cached_return, unsigned *incoming_return) const;
-    NodeInfo getNodeInfo() const;

    std::vector<unsigned> getNodeMessageStats(bool in = false) const;
    std::string getStorageLog() const;
@@ -381,7 +380,15 @@ public:
     */
    void join();

-    void setProxyServer(const std::string& proxy, const std::string& pushNodeId = "");
+    void setProxyServer(const std::string& proxy, const std::string& pushNodeId = "") {
+#if OPENDHT_PROXY_CLIENT
+        if (config_.proxy_server == proxy and config_.push_node_id == pushNodeId)
+            return;
+        config_.proxy_server = proxy;
+        config_.push_node_id = pushNodeId;
+        enableProxy(use_proxy and not config_.proxy_server.empty());
+#endif
+    }

    /**
     * Start or stop the proxy
@@ -416,7 +423,6 @@ private:
     */
    void tryBootstrapContinuously();

-    void stopNetwork();
    void startNetwork(const SockAddr sin4, const SockAddr sin6);
    time_point loop_();

@@ -458,13 +464,7 @@ private:

    std::thread rcv_thread {};
    std::mutex sock_mtx {};
-
-    struct ReceivedPacket {
-        Blob data;
-        SockAddr from;
-        time_point received;
-    };
-    std::queue<ReceivedPacket> rcv {};
+    std::vector<std::pair<Blob, SockAddr>> rcv {};

    /** true if currently actively boostraping */
    std::atomic_bool bootstraping {false};
@@ -487,7 +487,6 @@ private:
               status6 {NodeStatus::Disconnected};
    StatusCallback statusCb {nullptr};

-    int stop_writefd {-1};
    int s4 {-1}, s6 {-1};
    SockAddr bound4 {};
    SockAddr bound6 {};
@@ -43,23 +43,8 @@ inline void NOLOG(char const*, va_list) {}
 struct LogMethod {
    LogMethod() = default;

-    LogMethod(LogMethod&& l) : func(std::move(l.func)) {}
-    LogMethod(const LogMethod& l) : func(l.func) {}
-
-    LogMethod& operator=(dht::LogMethod&& l) {
-        func = std::forward<LogMethod>(l.func);
-        return *this;
-    }
-    LogMethod& operator=(const dht::LogMethod& l) {
-        func = l.func;
-        return *this;
-    }
-
    template<typename T>
-    explicit LogMethod(T&& t) : func(std::forward<T>(t)) {}
-
-    template<typename T>
-    LogMethod(const T& t) : func(t) {}
+    LogMethod(T&& t) : func(std::forward<T>(t)) {}

    void operator()(char const* format, ...) const {
        va_list args;
@@ -22,8 +22,8 @@
 namespace dht {
 namespace proxy {

-constexpr const std::chrono::seconds OP_TIMEOUT {1 * 60 * 60}; // one hour
-constexpr const std::chrono::seconds OP_MARGIN {5 * 60}; // 5 minutes
+constexpr const std::chrono::minutes OP_TIMEOUT {1 * 10}; // one hour
+constexpr const std::chrono::minutes OP_MARGIN {5}; // 5 minutes
 constexpr const char* const HTTP_PROTO {"http://"};
 using ListenToken = uint64_t;

@@ -25,7 +25,7 @@
 namespace dht {
 namespace crypto {

-#ifndef _MSC_VER
+#ifndef WIN32_NATIVE
 #ifdef _WIN32

 /**
@@ -315,7 +315,7 @@ public:
        DHT_LOG.DEBUG = debug;
        DHT_LOG.WARN = warn;
        DHT_LOG.ERR = error;
-        dht_->setLoggers(error, warn, debug);
+        dht_->setLoggers(std::forward<LogMethod>(error), std::forward<LogMethod>(warn), std::forward<LogMethod>(debug));
    }

    /**
@@ -37,9 +37,9 @@
 #include <chrono>
 #include <set>

-#ifdef OPENDHT_JSONCPP
+#if OPENDHT_PROXY_SERVER || OPENDHT_PROXY_CLIENT
 #include <json/json.h>
-#endif
+#endif //OPENDHT_PROXY_SERVER

 namespace dht {

@@ -372,13 +372,13 @@ struct OPENDHT_PUBLIC Value
    Value(ValueType::Id t, const uint8_t* dat_ptr, size_t dat_len, Id id = INVALID_ID)
     : id(id), type(t), data(dat_ptr, dat_ptr+dat_len) {}

-#ifdef OPENDHT_JSONCPP
+#if OPENDHT_PROXY_SERVER || OPENDHT_PROXY_CLIENT
    /**
     * Build a value from a json object
     * @param json
     */
    Value(Json::Value& json);
-#endif
+#endif //OPENDHT_PROXY_SERVER

    template <typename Type>
    Value(ValueType::Id t, const Type& d, Id id = INVALID_ID)
@@ -451,7 +451,7 @@ struct OPENDHT_PUBLIC Value
        return ss.str();
    }

-#ifdef OPENDHT_JSONCPP
+#if OPENDHT_PROXY_SERVER || OPENDHT_PROXY_CLIENT
    /**
     * Build a json object from a value
     * Example:
@@ -461,7 +461,7 @@ struct OPENDHT_PUBLIC Value
     * }
     */
    Json::Value toJson() const;
-#endif
+#endif //OPENDHT_PROXY_SERVER

    /** Return the size in bytes used by this value in memory (minimum). */
    size_t size() const;
@@ -996,7 +996,7 @@ unpackVector(const std::vector<std::shared_ptr<Value>>& vals) {
    return ret;
 }

-#ifdef OPENDHT_JSONCPP
+#if OPENDHT_PROXY_SERVER || OPENDHT_PROXY_CLIENT
 uint64_t unpackId(const Json::Value& json, const std::string& key);
 #endif

@@ -1,7 +1,7 @@
 lib_LTLIBRARIES = libopendht.la

-libopendht_la_CPPFLAGS = @CPPFLAGS@ -I$(top_srcdir)/include/opendht @Argon2_CFLAGS@ @JsonCpp_CFLAGS@
-libopendht_la_LIBADD   = @Argon2_LIBS@ @JsonCpp_LIBS@ @GnuTLS_LIBS@ @Nettle_LIBS@
+libopendht_la_CPPFLAGS = @CPPFLAGS@ -I$(top_srcdir)/include/opendht @Argon2_CFLAGS@
+libopendht_la_LIBADD   = @Argon2_LIBS@ @GnuTLS_LIBS@ @Nettle_LIBS@
 libopendht_la_LDFLAGS  = @LDFLAGS@ @Argon2_LDFLAGS@
 libopendht_la_SOURCES  = \
        dht.cpp \
@@ -62,12 +62,13 @@ NodeStats::toString() const
    ss << "Known nodes: " << good_nodes << " good, " << dubious_nodes << " dubious, " << incoming_nodes << " incoming." << std::endl;
    if (table_depth > 1) {
        ss << "Routing table depth: " << table_depth << std::endl;
-        ss << "Network size estimation: " << getNetworkSizeEstimation() << " nodes" << std::endl;
+        unsigned long tot_nodes = 8 * std::exp2(table_depth);
+        ss << "Network size estimation: " << tot_nodes << " nodes" << std::endl;
    }
    return ss.str();
 }

-#ifdef OPENDHT_JSONCPP
+#if OPENDHT_PROXY_SERVER || OPENDHT_PROXY_CLIENT
 /**
 * Build a json object from a NodeStats
 */
@@ -80,7 +81,8 @@ NodeStats::toJson() const
    val["incoming"] = static_cast<Json::LargestUInt>(incoming_nodes);
    if (table_depth > 1) {
        val["table_depth"] = static_cast<Json::LargestUInt>(table_depth);
-        val["network_size_estimation"] = static_cast<Json::LargestUInt>(getNetworkSizeEstimation());
+        unsigned long tot_nodes = 8 * std::exp2(table_depth);
+        val["network_size_estimation"] = static_cast<Json::LargestUInt>(tot_nodes);
    }
    return val;
 }
@@ -96,32 +98,6 @@ NodeStats::NodeStats(const Json::Value& val)
    if (val.isMember("table_depth"))
        table_depth = static_cast<unsigned>(val["table_depth"].asLargestUInt());
 }
-
-/**
- * Build a json object from a NodeStats
- */
-Json::Value
-NodeInfo::toJson() const
-{
-    Json::Value val;
-    if (id)
-        val["id"] = id.toString();
-    val["node_id"] = node_id.toString();
-    val["ipv4"] = ipv4.toJson();
-    val["ipv6"] = ipv6.toJson();
-    return val;
-}
-
-NodeInfo::NodeInfo(const Json::Value& v)
-{
-    if (v.isMember("id"))
-        id = InfoHash(v["id"].asString());
-    node_id = InfoHash(v["node_id"].asString());
-    ipv4 = NodeStats(v["ipv4"]);
-    ipv6 = NodeStats(v["ipv6"]);
-}
-
-#endif
-
+#endif //OPENDHT_PROXY_SERVER

 }
@@ -261,11 +261,11 @@ void Dht::paginate(std::weak_ptr<Search> ws, Sp<Query> query, SearchNode* n) {
    auto select_q = std::make_shared<Query>(Select {}.field(Value::Field::Id), query ? query->where : Where {});
    auto onSelectDone = [this,ws,query](const net::Request& status,
                                        net::RequestAnswer&& answer) mutable {
-        // Retrieve search
+        // retreive search
        auto sr = ws.lock();
        if (not sr) return;
        const auto& id = sr->id;
-        // Retrieve search node
+        // retreive search node
        auto sn = sr->getNode(status.node);
        if (not sn) return;
        // backward compatibility
@@ -319,12 +319,11 @@ Dht::searchSendGetValues(Sp<Search> sr, SearchNode* pn, bool update)

    std::weak_ptr<Search> ws = sr;
    auto cb = sr->callbacks.begin();
-    static const auto ANY_QUERY = std::make_shared<Query>(Select {}, Where {}, true);
    do { /* for all requests to send */
        SearchNode* n = nullptr;
-        auto& query = sr->callbacks.empty() ? ANY_QUERY : cb->second.query;
-        const time_point up = (not sr->callbacks.empty() and update)
-                                ? sr->getLastGetTime(*query)
+        auto query = not sr->callbacks.empty() ? cb->second.query : std::make_shared<Query>(Select {}, Where {}, true);
+        const time_point up = not sr->callbacks.empty() and update
+                                ? sr->getLastGetTime(query)
                                : time_point::min();

        if (pn and pn->canGet(now, up, query)) {
@@ -383,94 +382,6 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr) {
    unsigned i = 0;
    auto probe_query = std::make_shared<Query>(Select {}.field(Value::Field::Id).field(Value::Field::SeqNum));
    std::weak_ptr<Search> ws = sr;
-
-    auto onDone = [this,ws](const net::Request& req, net::RequestAnswer&& answer)
-    { /* when put done */
-        if (auto sr = ws.lock()) {
-            onAnnounceDone(req.node, answer, sr);
-            searchStep(sr);
-        }
-    };
-
-    auto onExpired = [this,ws](const net::Request&, bool over)
-    { /* when put expired */
-        if (over)
-            if (auto sr = ws.lock())
-                scheduler.edit(sr->nextSearchStep, scheduler.time());
-    };
-
-    auto onSelectDone =
-    [this,ws,onDone,onExpired](const net::Request& req, net::RequestAnswer&& answer) mutable
-    { /* on probing done */
-        auto sr = ws.lock();
-        if (not sr) return;
-        const auto& now = scheduler.time();
-        sr->insertNode(req.node, scheduler.time(), answer.ntoken);
-        auto sn = sr->getNode(req.node);
-        if (not sn) return;
-
-        if (not sn->isSynced(now)) {
-            /* Search is now unsynced. Let's call searchStep to sync again. */
-            scheduler.edit(sr->nextSearchStep, now);
-            return;
-        }
-        for (auto& a : sr->announce) {
-            if (sn->getAnnounceTime(a.value->id) > now)
-                continue;
-            bool hasValue {false};
-            uint16_t seq_no = 0;
-            try {
-                const auto& f = std::find_if(answer.fields.cbegin(), answer.fields.cend(),
-                        [&a](const Sp<FieldValueIndex>& i){
-                            return i->index.at(Value::Field::Id).getInt() == a.value->id;
-                        });
-                if (f != answer.fields.cend() and *f) {
-                    hasValue = true;
-                    seq_no = static_cast<uint16_t>((*f)->index.at(Value::Field::SeqNum).getInt());
-                }
-            } catch (std::out_of_range&) { }
-
-            auto next_refresh_time = now + getType(a.value->type).expiration;
-            /* only put the value if the node doesn't already have it */
-            if (not hasValue or seq_no < a.value->seq) {
-                DHT_LOG.w(sr->id, sn->node->id, "[search %s] [node %s] sending 'put' (vid: %d)",
-                        sr->id.toString().c_str(), sn->node->toString().c_str(), a.value->id);
-                sn->acked[a.value->id] = std::make_pair(network_engine.sendAnnounceValue(sn->node,
-                                                sr->id,
-                                                a.value,
-                                                a.permanent ? time_point::max() : a.created,
-                                                sn->token,
-                                                onDone,
-                                                onExpired), next_refresh_time);
-            } else if (hasValue and a.permanent) {
-                DHT_LOG.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] = std::make_pair(network_engine.sendRefreshValue(sn->node,
-                                                sr->id,
-                                                a.value->id,
-                                                sn->token,
-                                                onDone,
-                                                onExpired), next_refresh_time);
-            } else {
-                DHT_LOG.w(sr->id, sn->node->id, "[search %s] [node %s] already has value (vid: %d). Aborting.",
-                        sr->id.toString().c_str(), sn->node->toString().c_str(), a.value->id);
-                auto ack_req = std::make_shared<net::Request>(net::Request::State::COMPLETED);
-                ack_req->reply_time = now;
-                sn->acked[a.value->id] = std::make_pair(std::move(ack_req), next_refresh_time);
-
-                /* step to clear announces */
-                scheduler.edit(sr->nextSearchStep, now);
-            }
-            if (a.permanent) {
-                scheduler.add(next_refresh_time - REANNOUNCE_MARGIN, [this,ws] {
-                    if (auto sr = ws.lock()) {
-                        searchStep(sr);
-                    }
-                });
-            }
-        }
-    };
-
    const auto& now = scheduler.time();
    for (auto& n : sr->nodes) {
        if (not n.isSynced(now))
@@ -481,6 +392,82 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr) {
            }) == sr->announce.cend())
            continue;

+        auto onDone = [this,ws](const net::Request& req, net::RequestAnswer&& answer)
+        { /* when put done */
+            if (auto sr = ws.lock()) {
+                onAnnounceDone(req.node, answer, sr);
+                searchStep(sr);
+            }
+        };
+        auto onExpired = [this,ws](const net::Request&, bool over)
+        { /* when put expired */
+            if (over)
+                if (auto sr = ws.lock())
+                    scheduler.edit(sr->nextSearchStep, scheduler.time());
+        };
+        auto onSelectDone =
+            [this,ws,onDone,onExpired](const net::Request& req, net::RequestAnswer&& answer) mutable
+            { /* on probing done */
+                auto sr = ws.lock();
+                if (not sr) return;
+                auto sn = sr->getNode(req.node);
+                if (not sn) return;
+
+                const auto& now = scheduler.time();
+                if (not sn->isSynced(now)) {
+                    /* Search is now unsynced. Let's call searchStep to sync again. */
+                    scheduler.edit(sr->nextSearchStep, now);
+                    return;
+                }
+                for (auto& a : sr->announce) {
+                    if (sn->getAnnounceTime(a.value->id) > now)
+                        continue;
+                    bool hasValue {false};
+                    uint16_t seq_no = 0;
+                    try {
+                        const auto& f = std::find_if(answer.fields.cbegin(), answer.fields.cend(),
+                                [&a](const Sp<FieldValueIndex>& i){
+                                    return i->index.at(Value::Field::Id).getInt() == a.value->id;
+                                });
+                        if (f != answer.fields.cend() and *f) {
+                            hasValue = true;
+                            seq_no = static_cast<uint16_t>((*f)->index.at(Value::Field::SeqNum).getInt());
+                        }
+                    } catch (std::out_of_range&) { }
+
+                    auto next_refresh_time = now + getType(a.value->type).expiration;
+                    /* only put the value if the node doesn't already have it */
+                    if (not hasValue or seq_no < a.value->seq) {
+                        DHT_LOG.w(sr->id, sn->node->id, "[search %s] [node %s] sending 'put' (vid: %d)",
+                                sr->id.toString().c_str(), sn->node->toString().c_str(), a.value->id);
+                        sn->acked[a.value->id] = std::make_pair(network_engine.sendAnnounceValue(sn->node,
+                                                        sr->id,
+                                                        a.value,
+                                                        a.permanent ? time_point::max() : a.created,
+                                                        sn->token,
+                                                        onDone,
+                                                        onExpired), next_refresh_time);
+                    } else if (hasValue and a.permanent) {
+                        DHT_LOG.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] = std::make_pair(network_engine.sendRefreshValue(sn->node,
+                                                        sr->id,
+                                                        a.value->id,
+                                                        sn->token,
+                                                        onDone,
+                                                        onExpired), next_refresh_time);
+                    } else {
+                        DHT_LOG.w(sr->id, sn->node->id, "[search %s] [node %s] already has value (vid: %d). Aborting.",
+                                sr->id.toString().c_str(), sn->node->toString().c_str(), a.value->id);
+                        auto ack_req = std::make_shared<net::Request>(net::Request::State::COMPLETED);
+                        ack_req->reply_time = now;
+                        sn->acked[a.value->id] = std::make_pair(std::move(ack_req), next_refresh_time);
+
+                        /* step to clear announces */
+                        scheduler.edit(sr->nextSearchStep, now);
+                    }
+                }
+            };
        DHT_LOG.w(sr->id, n.node->id, "[search %s] [node %s] sending %s",
                sr->id.toString().c_str(), n.node->toString().c_str(), probe_query->toString().c_str());
        n.probe_query = probe_query;
@@ -529,7 +516,7 @@ Dht::searchSynchedNodeListen(const Sp<Search>& sr, SearchNode& n)
            });
        }
        auto prev_req = r != n.listenStatus.end() ? r->second.req : nullptr;
-        auto new_req = network_engine.sendListen(n.node, sr->id, *query, n.token, prev_req,
+        r->second.req = network_engine.sendListen(n.node, sr->id, *query, n.token, prev_req,
            [this,ws,query](const net::Request& req, net::RequestAnswer&& answer) mutable
            { /* on done */
                if (auto sr = ws.lock()) {
@@ -552,18 +539,12 @@ Dht::searchSynchedNodeListen(const Sp<Search>& sr, SearchNode& n)
            { /* 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
-        r = n.listenStatus.find(query);
-        if (r != n.listenStatus.end()) {
-            r->second.req = new_req;
-        }
    }
 }

@@ -764,6 +745,15 @@ Dht::announce(const InfoHash& id,
        time_point created,
        bool permanent)
 {
+    const auto& now = scheduler.time();
+    if (!value) {
+        if (callback)
+            callback(false, {});
+        return;
+    }
+    created = std::min(now, created);
+    storageStore(id, value, created);
+
    auto& srs = searches(af);
    auto srp = srs.find(id);
    auto sr = srp == srs.end() ? search(id, af) : srp->second;
@@ -915,20 +905,13 @@ struct GetStatus : public OpStatus {
 void
 Dht::put(const InfoHash& id, Sp<Value> val, DoneCallback callback, time_point created, bool permanent)
 {
-    if (not val) {
-        if (callback)
-            callback(false, {});
-        return;
-    }
+    scheduler.syncTime();
+
    if (val->id == Value::INVALID_ID) {
        crypto::random_device rdev;
        std::uniform_int_distribution<Value::Id> rand_id {};
        val->id = rand_id(rdev);
    }
-    scheduler.syncTime();
-    const auto& now = scheduler.time();
-    created = std::min(now, created);
-    storageStore(id, val, created, {}, true);

    DHT_LOG.d(id, "put: adding %s -> %s", id.toString().c_str(), val->toString().c_str());

@@ -1132,8 +1115,6 @@ bool
 Dht::cancelPut(const InfoHash& id, const Value::Id& vid)
 {
    bool canceled {false};
-    if (storageErase(id, vid))
-        canceled = true;
    auto sr_cancel_put = [&](std::map<InfoHash, Sp<Search>> srs) {
        auto srp = srs.find(id);
        if (srp == srs.end())
@@ -1202,11 +1183,11 @@ Dht::storageChanged(const InfoHash& id, Storage& st, ValueStorage& v, bool newVa
 }

 bool
-Dht::storageStore(const InfoHash& id, const Sp<Value>& value, time_point created, const SockAddr& sa, bool permanent)
+Dht::storageStore(const InfoHash& id, const Sp<Value>& value, time_point created, const SockAddr& sa)
 {
    const auto& now = scheduler.time();
    created = std::min(created, now);
-    auto expiration = permanent ? time_point::max() : created + getType(value->type).expiration;
+    auto expiration = created + getType(value->type).expiration;
    if (expiration < now)
        return false;

@@ -1228,9 +1209,7 @@ Dht::storageStore(const InfoHash& id, const Sp<Value>& value, time_point created
    if (auto vs = store.first) {
        total_store_size += store.second.size_diff;
        total_values += store.second.values_diff;
-        if (not permanent) {
-            scheduler.add(expiration, std::bind(&Dht::expireStorage, this, id));
-        }
+        scheduler.add(expiration, std::bind(&Dht::expireStorage, this, id));
        if (total_store_size > max_store_size) {
            expireStore();
        }
@@ -1240,18 +1219,6 @@ Dht::storageStore(const InfoHash& id, const Sp<Value>& value, time_point created
    return std::get<0>(store);
 }

-bool
-Dht::storageErase(const InfoHash& id, Value::Id vid)
-{
-    auto st = store.find(id);
-    if (st == store.end())
-        return false;
-    auto ret = st->second.remove(id, vid);
-    total_store_size += ret.size_diff;
-    total_values += ret.values_diff;
-    return ret.values_diff;
-}
-
 void
 Dht::storageAddListener(const InfoHash& id, const Sp<Node>& node, size_t socket_id, Query&& query)
 {
@@ -1568,7 +1535,10 @@ Dht::dumpSearch(const Search& sr, std::ostream& out) const
                    if (ack == n.acked.end() or not ack->second.first) {
                        out << ' ';
                    } else {
-                        out << ack->second.first->getStateChar();
+                        if (ack->second.first->completed())
+                            out << 'a';
+                        else if (ack->second.first->pending())
+                            out << 'f';
                    }
                }
                out << "] ";
@@ -1610,6 +1580,11 @@ Dht::getStorageLog() const
    std::stringstream out;
    for (const auto& s : store)
        out << printStorageLog(s);
+    out << "Total " << store.size() << " storages, " << total_values << " values (";
+    if (total_store_size < 1024)
+        out << total_store_size << " bytes)";
+    else
+        out << (total_store_size/1024) << " KB)";
    out << std::endl << std::endl;
    std::multimap<size_t, const SockAddr*> q_map;
    for (const auto& ip : store_quota)
@@ -1618,12 +1593,6 @@ Dht::getStorageLog() const
    for (auto ip = q_map.rbegin(); ip != q_map.rend(); ++ip)
        out << "IP " << ip->second->toString() << " uses " << ip->first << " bytes" << std::endl;
    out << std::endl;
-    out << "Total " << store.size() << " storages, " << total_values << " values (";
-    if (total_store_size < 1024)
-        out << total_store_size << " bytes)";
-    else
-        out << (total_store_size/1024) << " / " << (max_store_size/1024) << " KB)";
-    out << std::endl;
    return out.str();
 }

@@ -2374,10 +2343,7 @@ Dht::onRefresh(Sp<Node> node, const InfoHash& hash, const Blob& token, const Val
    }

    auto s = store.find(hash);
-    if (s != store.end()) {
-        auto expiration = s->second.refresh(now, vid, types);
-        if (expiration != time_point::max())
-            scheduler.add(expiration, std::bind(&Dht::expireStorage, this, hash));
+    if (s != store.end() and s->second.refresh(now, vid)) {
        DHT_LOG.d(hash, node->id, "[store %s] [node %s] refreshed value %s", hash.toString().c_str(), node->toString().c_str(), std::to_string(vid).c_str());
    } else {
        DHT_LOG.d(hash, node->id, "[store %s] [node %s] got refresh for unknown value",
@@ -2393,6 +2359,9 @@ Dht::onAnnounceDone(const Sp<Node>& node, net::RequestAnswer& answer, Sp<Search>
    DHT_LOG.d(sr->id, node->id, "[search %s] [node %s] got reply to put!",
            sr->id.toString().c_str(), node->toString().c_str());
    searchSendGetValues(sr);
+    /* if (auto sn = sr->getNode(req->node)) { */
+    /*     sn->setRefreshTime(answer.vid, now + answer) */
+    /* } */
    sr->checkAnnounced(answer.vid);
 }

@@ -33,18 +33,12 @@

 namespace dht {

-struct DhtProxyClient::InfoState {
-    std::atomic_uint ipv4 {0}, ipv6 {0};
-    std::atomic_bool cancel {false};
-};
-
-struct DhtProxyClient::ListenState {
-    std::atomic_bool ok {true};
-    std::atomic_bool cancel {false};
-};
-
 struct DhtProxyClient::Listener
 {
+    struct State {
+        std::atomic_bool ok {true};
+        std::atomic_bool cancel {false};
+    };
    ValueCache cache;
    Sp<Scheduler::Job> cacheExpirationJob {};
    ValueCallback cb;
@@ -52,7 +46,7 @@ struct DhtProxyClient::Listener
    Sp<restbed::Request> req;
    std::thread thread;
    unsigned callbackId;
-    Sp<ListenState> state;
+    Sp<State> state;
    Sp<proxy::ListenToken> pushNotifToken;
    Sp<Scheduler::Job> refreshJob;
    Listener(ValueCache&& c, Sp<Scheduler::Job>&& j, const Sp<restbed::Request>& r, Value::Filter&& f)
@@ -66,7 +60,6 @@ struct DhtProxyClient::Listener
 struct PermanentPut {
    Sp<Value> value;
    Sp<Scheduler::Job> refreshJob;
-    Sp<std::atomic_bool> ok;
 };

 struct DhtProxyClient::ProxySearch {
@@ -106,8 +99,6 @@ DhtProxyClient::~DhtProxyClient()
    isDestroying_ = true;
    cancelAllOperations();
    cancelAllListeners();
-    if (infoState_)
-        infoState_->cancel = true;
    if (statusThread_.joinable())
        statusThread_.join();
 }
@@ -331,29 +322,28 @@ DhtProxyClient::put(const InfoHash& key, Sp<Value> val, DoneCallback cb, time_po
        std::lock_guard<std::mutex> lock(searchLock_);
        auto id = val->id;
        auto search = searches_.emplace(key, ProxySearch{}).first;
-        auto nextRefresh = scheduler.time() + proxy::OP_TIMEOUT - proxy::OP_MARGIN;
-        auto ok = std::make_shared<std::atomic_bool>(false);
+        auto nextRefresh = scheduler.time() + proxy::OP_TIMEOUT;
        search->second.puts.erase(id);
-        search->second.puts.emplace(id, PermanentPut {val, scheduler.add(nextRefresh, [this, key, id, ok]{
+        search->second.puts.emplace(id, PermanentPut {val, scheduler.add(nextRefresh, [this, key, id]{
+            DHT_LOG.w(key, "[search %s]: PermanentPut refresh callback vid %llu", key.to_c_str(), id);
            std::lock_guard<std::mutex> lock(searchLock_);
            auto s = searches_.find(key);
            if (s != searches_.end()) {
                auto p = s->second.puts.find(id);
                if (p != s->second.puts.end()) {
-                    doPut(key, p->second.value,
-                    [ok](bool result, const std::vector<std::shared_ptr<dht::Node> >&){
-                        *ok = result;
+                    doPut(key, p->second.value, [this](bool ok, const std::vector<Sp<Node>>&) {
+                        DHT_LOG.w("PermanentPut refresh %s", (ok ? "ok" : "failed"));
                    }, time_point::max(), true);
-                    scheduler.edit(p->second.refreshJob, scheduler.time() + proxy::OP_TIMEOUT - proxy::OP_MARGIN);
+                    scheduler.edit(p->second.refreshJob, scheduler.time() + proxy::OP_TIMEOUT);
                }
            }
-        }), ok});
+        })});
    }
    doPut(key, val, std::move(cb), created, permanent);
 }

 void
-DhtProxyClient::doPut(const InfoHash& key, Sp<Value> val, DoneCallback cb, time_point /*created*/, bool permanent)
+DhtProxyClient::doPut(const InfoHash& key, Sp<Value> val, DoneCallback cb, time_point created, bool permanent)
 {
    DHT_LOG.d(key, "[search %s] performing put of %s", key.to_c_str(), val->toString().c_str());
    restbed::Uri uri(proxy::HTTP_PROTO + serverHost_ + "/" + key.toString());
@@ -485,12 +475,9 @@ void
 DhtProxyClient::getProxyInfos()
 {
    DHT_LOG.d("Requesting proxy server node information");
-    std::lock_guard<std::mutex> l(statusLock_);

-    auto infoState = std::make_shared<InfoState>();
-    if (infoState_)
-        infoState_->cancel = true;
-    infoState_ = infoState;
+    if (ongoingStatusUpdate_.test_and_set())
+        return;

    {
        std::lock_guard<std::mutex> l(lockCurrentProxyInfos_);
@@ -505,15 +492,11 @@ DhtProxyClient::getProxyInfos()

    // Try to contact the proxy and set the status to connected when done.
    // will change the connectivity status
-    if (statusThread_.joinable()) {
-        try {
-            statusThread_.detach();
-            statusThread_ = {};
-        } catch (const std::exception& e) {
-            DHT_LOG.e("Error detaching thread: %s", e.what());
-        }
-    }
-    statusThread_ = std::thread([this, serverHost, infoState]{
+    if (statusThread_.joinable())
+        statusThread_.join();
+    statusThread_ = std::thread([this, serverHost]{
+        struct SuccessCount {std::atomic_uint ipv4 {0}, ipv6 {0};};
+        auto successCount = std::make_shared<SuccessCount>();
        try {
            auto hostAndService = splitPort(serverHost);
            auto resolved_proxies = SockAddr::resolve(hostAndService.first, hostAndService.second);
@@ -528,10 +511,8 @@ DhtProxyClient::getProxyInfos()
                }
                restbed::Uri uri(proxy::HTTP_PROTO + server + "/");
                auto req = std::make_shared<restbed::Request>(uri);
-                if (infoState->cancel)
-                    return;
                reqs.emplace_back(restbed::Http::async(req,
-                    [this, resolved_proxy, infoState](
+                    [this, resolved_proxy, successCount](
                                const std::shared_ptr<restbed::Request>&,
                                const std::shared_ptr<restbed::Response>& reply)
                {
@@ -539,8 +520,6 @@ DhtProxyClient::getProxyInfos()
                    Json::Value proxyInfos;
                    if (code == 200) {
                        restbed::Http::fetch("\n", reply);
-                        auto& state = *infoState;
-                        if (state.cancel) return;
                        std::string body;
                        reply->get_body(body);

@@ -549,14 +528,13 @@ DhtProxyClient::getProxyInfos()
                        auto reader = std::unique_ptr<Json::CharReader>(rbuilder.newCharReader());
                        try {
                            reader->parse(body.data(), body.data() + body.size(), &proxyInfos, &err);
-                        } catch (...) {
-                            return;
+                        } catch (const std::exception& e) {
+                            DHT_LOG.w("Error parsing proxy infos: %s", e.what());
                        }
                        auto family = resolved_proxy.getFamily();
-                        if      (family == AF_INET)  state.ipv4++;
-                        else if (family == AF_INET6) state.ipv6++;
-                        if (not state.cancel)
-                            onProxyInfos(proxyInfos, family);
+                        if      (family == AF_INET)  successCount->ipv4++;
+                        else if (family == AF_INET6) successCount->ipv6++;
+                        onProxyInfos(proxyInfos, family);
                    }
                }));
            }
@@ -566,24 +544,22 @@ DhtProxyClient::getProxyInfos()
        } catch (const std::exception& e) {
            DHT_LOG.e("Error sending proxy info request: %s", e.what());
        }
-        const auto& state = *infoState;
-        if (state.cancel) return;
-        if (state.ipv4 == 0) onProxyInfos(Json::Value{}, AF_INET);
-        if (state.ipv6 == 0) onProxyInfos(Json::Value{}, AF_INET6);
+        if (successCount->ipv4 == 0) onProxyInfos(Json::Value{}, AF_INET);
+        if (successCount->ipv6 == 0) onProxyInfos(Json::Value{}, AF_INET6);
+        ongoingStatusUpdate_.clear();
    });
 }

 void
 DhtProxyClient::onProxyInfos(const Json::Value& proxyInfos, sa_family_t family)
 {
+    DHT_LOG.d("Got proxy infos %s", family == AF_INET ? "IPv4" : "IPv6");
    std::lock_guard<std::mutex> l(lockCurrentProxyInfos_);
    auto oldStatus = std::max(statusIpv4_, statusIpv6_);
    auto& status = family == AF_INET ? statusIpv4_ : statusIpv6_;
    if (not proxyInfos.isMember("node_id")) {
-        DHT_LOG.e("Proxy info request failed for %s", family == AF_INET ? "IPv4" : "IPv6");
        status = NodeStatus::Disconnected;
    } else {
-        DHT_LOG.d("Got proxy reply for %s", family == AF_INET ? "IPv4" : "IPv6");
        try {
            myid = InfoHash(proxyInfos["node_id"].asString());
            stats4_ = NodeStats(proxyInfos["ipv4"]);
@@ -647,7 +623,7 @@ DhtProxyClient::listen(const InfoHash& key, ValueCallback cb, Value::Filter filt
        it = searches_.emplace(key, ProxySearch{}).first;
    }
    auto query = std::make_shared<Query>(Select{}, where);
-    auto token = it->second.ops.listen(cb, query, filter, [&](Sp<Query> /*q*/, ValueCallback vcb){
+    auto token = it->second.ops.listen(cb, query, filter, [&](Sp<Query> q, ValueCallback vcb){
        return doListen(key, vcb, filter);
    });
    return token;
@@ -680,104 +656,14 @@ DhtProxyClient::cancelListen(const InfoHash& key, size_t gtoken) {
    return canceled;
 }

-
-void
-DhtProxyClient::sendListen(const std::shared_ptr<restbed::Request>& req, const ValueCallback& cb, const Value::Filter& filter, const Sp<ListenState>& state) {
-    auto settings = std::make_shared<restbed::Settings>();
-    std::chrono::milliseconds timeout(std::numeric_limits<int>::max());
-    settings->set_connection_timeout(timeout); // Avoid the client to close the socket after 5 seconds.
-    req->set_method("LISTEN");
-    restbed::Http::async(req,
-        [this, filter, cb, state](const std::shared_ptr<restbed::Request>& req,
-                                  const std::shared_ptr<restbed::Response>& reply) {
-        auto code = reply->get_status_code();
-        if (code == 200) {
-            try {
-                while (restbed::Http::is_open(req) and not state->cancel) {
-                    restbed::Http::fetch("\n", reply);
-                    if (state->cancel)
-                        break;
-                    std::string body;
-                    reply->get_body(body);
-                    reply->set_body(""); // Reset the body for the next fetch
-
-                    Json::Value json;
-                    std::string err;
-                    Json::CharReaderBuilder rbuilder;
-                    auto reader = std::unique_ptr<Json::CharReader>(rbuilder.newCharReader());
-                    if (reader->parse(body.data(), body.data() + body.size(), &json, &err)) {
-                        auto value = std::make_shared<Value>(json);
-                        if ((not filter or filter(*value)) and cb)  {
-                            std::lock_guard<std::mutex> lock(lockCallbacks);
-                            callbacks_.emplace_back([cb, value, state]() {
-                                if (not state->cancel and not cb({value}, false))
-                                    state->cancel = true;
-                            });
-                            loopSignal_();
-                        }
-                    }
-                }
-            } catch (const std::exception& e) {
-                if (not state->cancel) {
-                    DHT_LOG.w("Listen closed by the proxy server: %s", e.what());
-                    state->ok = false;
-                }
-            }
-        } else {
-            state->ok = false;
-        }
-    }, settings).get();
-    auto& s = *state;
-    if (not s.ok and not s.cancel)
-        opFailed();
-}
-
-void
-DhtProxyClient::sendSubscribe(const std::shared_ptr<restbed::Request>& req, const Sp<proxy::ListenToken>& token, const Sp<ListenState>& state) {
-#if OPENDHT_PUSH_NOTIFICATIONS
-    req->set_method("SUBSCRIBE");
-    fillBodyToGetToken(req);
-    restbed::Http::async(req, [this,state, token](const std::shared_ptr<restbed::Request>&,
-                                             const std::shared_ptr<restbed::Response>& reply) {
-        auto code = reply->get_status_code();
-        if (code == 200) {
-            try {
-                restbed::Http::fetch("\n", reply);
-                std::string body;
-                reply->get_body(body);
-
-                std::string err;
-                Json::Value json;
-                Json::CharReaderBuilder rbuilder;
-                auto reader = std::unique_ptr<Json::CharReader>(rbuilder.newCharReader());
-                if (reader->parse(body.data(), body.data() + body.size(), &json, &err)) {
-                    if (not json.isMember("token")) {
-                        state->ok = false;
-                        return;
-                    }
-                    *token = unpackId(json, "token");
-                }
-            } catch (const std::exception& e) {
-                if (not state->cancel) {
-                    DHT_LOG.e("sendSubscribe: error: %s", e.what());
-                    state->ok = false;
-                }
-            }
-        } else {
-            state->ok = false;
-        }
-    }).get();
-    auto& s = *state;
-    if (not s.ok and not s.cancel)
-        opFailed();
-#endif
-}
-
 size_t
 DhtProxyClient::doListen(const InfoHash& key, ValueCallback cb, Value::Filter filter/*, Where where*/)
 {
    scheduler.syncTime();
    restbed::Uri uri(proxy::HTTP_PROTO + serverHost_ + "/" + key.toString());
+    auto req = std::make_shared<restbed::Request>(uri);
+    req->set_method(deviceKey_.empty() ? "LISTEN" : "SUBSCRIBE");
+
    std::lock_guard<std::mutex> lock(searchLock_);
    auto search = searches_.find(key);
    if (search == searches_.end()) {
@@ -786,7 +672,6 @@ DhtProxyClient::doListen(const InfoHash& key, ValueCallback cb, Value::Filter fi
    }
    DHT_LOG.d(key, "[search %s] sending %s", key.to_c_str(), deviceKey_.empty() ? "listen" : "subscribe");

-    auto req = std::make_shared<restbed::Request>(uri);
    auto token = ++listenerToken_;
    auto l = search->second.listeners.find(token);
    if (l == search->second.listeners.end()) {
@@ -806,12 +691,9 @@ DhtProxyClient::doListen(const InfoHash& key, ValueCallback cb, Value::Filter fi
                scheduler.edit(l->second.cacheExpirationJob, next);
            }), req, std::move(f)
        }).first;
-    } else {
-        if (l->second.state)
-            l->second.state->cancel = true;
    }

-    auto state = std::make_shared<ListenState>();
+    auto state = std::make_shared<Listener::State>();
    l->second.state = state;
    l->second.cb = [this,key,token,state](const std::vector<Sp<Value>>& values, bool expired) {
        if (state->cancel)
@@ -837,34 +719,82 @@ DhtProxyClient::doListen(const InfoHash& key, ValueCallback cb, Value::Filter fi
        loopSignal_();
        return true;
    };
-
    auto pushNotifToken = std::make_shared<proxy::ListenToken>(0);
    auto vcb = l->second.cb;
    l->second.pushNotifToken = pushNotifToken;
-    l->second.req = req;
-
-    if (not deviceKey_.empty()) {
-        // Relaunch push listeners even if a timeout is not received (if the proxy crash for any reason)
-        l->second.refreshJob = scheduler.add(scheduler.time() + proxy::OP_TIMEOUT - proxy::OP_MARGIN, [this, key, token, state] {
-            if (state->cancel)
-                return;
-            std::lock_guard<std::mutex> lock(searchLock_);
-            auto s = searches_.find(key);
-            if (s != searches_.end()) {
-                auto l = s->second.listeners.find(token);
-                if (l != s->second.listeners.end()) {
-                    resubscribe(key, l->second);
-                }
+    l->second.thread = std::thread([this,req,filter,vcb,pushNotifToken,state]()
+        {
+            auto settings = std::make_shared<restbed::Settings>();
+            if (deviceKey_.empty()) {
+                std::chrono::milliseconds timeout(std::numeric_limits<int>::max());
+                settings->set_connection_timeout(timeout); // Avoid the client to close the socket after 5 seconds.
            }
-        });
-        l->second.thread = std::thread([this, req,pushNotifToken, state](){
-            sendSubscribe(req,pushNotifToken,state);
-        });
-    } else {
-        l->second.thread = std::thread([this, req, vcb, filter, state]{
-            sendListen(req,vcb,filter,state);
-        });
-    }
+#if OPENDHT_PUSH_NOTIFICATIONS
+            else
+                fillBodyToGetToken(req);
+#endif
+
+            restbed::Http::async(req,
+                [this, filter, vcb, pushNotifToken, state](const std::shared_ptr<restbed::Request>& req,
+                                                           const std::shared_ptr<restbed::Response>& reply) {
+                auto code = reply->get_status_code();
+                if (code == 200) {
+                    try {
+                        std::string err;
+                        Json::Value json;
+                        Json::CharReaderBuilder rbuilder;
+                        auto reader = std::unique_ptr<Json::CharReader>(rbuilder.newCharReader());
+                        if (!deviceKey_.empty()) {
+                            restbed::Http::fetch("\n", reply);
+                            if (state->cancel)
+                                return;
+                            std::string body;
+                            reply->get_body(body);
+
+                            auto* char_data = reinterpret_cast<const char*>(&body[0]);
+                            if (reader->parse(char_data, char_data + body.size(), &json, &err)) {
+                                *pushNotifToken = unpackId(json, "token");
+                            } else {
+                                state->ok = false;
+                            }
+                        } else {
+                            while (restbed::Http::is_open(req) and not state->cancel) {
+                                restbed::Http::fetch("\n", reply);
+                                if (state->cancel)
+                                    break;
+                                std::string body;
+                                reply->get_body(body);
+                                reply->set_body(""); // Reset the body for the next fetch
+
+                                auto* char_data = reinterpret_cast<const char*>(&body[0]);
+                                if (reader->parse(char_data, char_data + body.size(), &json, &err)) {
+                                    auto value = std::make_shared<Value>(json);
+                                    if ((not filter or filter(*value)) and vcb)  {
+                                        std::lock_guard<std::mutex> lock(lockCallbacks);
+                                        callbacks_.emplace_back([vcb, value, state]() {
+                                            if (not state->cancel and not vcb({value}, false))
+                                                state->cancel = true;
+                                        });
+                                        loopSignal_();
+                                    }
+                                } else {
+                                    state->ok = false;
+                                }
+                            }
+                        }
+                    } catch (std::runtime_error&) {
+                        // NOTE: Http::close() can occurs here. Ignore this.
+                    }
+                } else {
+                    state->ok = false;
+                }
+            }, settings).get();
+            auto& s = *state;
+            if (not s.ok and not s.cancel) {
+                opFailed();
+            }
+        }
+    );
    return token;
 }

@@ -948,53 +878,81 @@ DhtProxyClient::getConnectivityStatus()
 void
 DhtProxyClient::restartListeners()
 {
-    DHT_LOG.d("Refresh permanent puts");
-    for (auto& search : searches_) {
-        for (auto& put : search.second.puts) {
-            if (!*put.second.ok) {
-                auto ok = put.second.ok;
-                doPut(search.first, put.second.value,
-                [ok](bool result, const std::vector<std::shared_ptr<dht::Node> >&){
-                    *ok = result;
-                }, time_point::max(), true);
-                scheduler.edit(put.second.refreshJob, scheduler.time() + proxy::OP_TIMEOUT - proxy::OP_MARGIN);
-            }
-        }
-    }
    if (not deviceKey_.empty()) {
-        DHT_LOG.d("resubscribe due to a connectivity change");
-        // Connectivity changed, refresh all subscribe
-        for (auto& search : searches_)
-            for (auto& listener : search.second.listeners)
-                if (!listener.second.state->ok)
-                    resubscribe(search.first, listener.second);
        return;
    }
    DHT_LOG.d("Restarting listeners");
    std::lock_guard<std::mutex> lock(searchLock_);
    for (auto& search: searches_) {
-        for (auto& l: search.second.listeners) {
-            auto& listener = l.second;
-            auto state = listener.state;
-            if (listener.thread.joinable()) {
-                state->cancel = true;
-                if (listener.req)
-                    restbed::Http::close(listener.req);
-                listener.thread.join();
-            }
-            // Redo listen
-            state->cancel = false;
-            state->ok = true;
-            auto filter = listener.filter;
-            auto cb = listener.cb;
-            restbed::Uri uri(proxy::HTTP_PROTO + serverHost_ + "/" + search.first.toString());
-            auto req = std::make_shared<restbed::Request>(uri);
-            req->set_method("LISTEN");
-            listener.req = req;
-            listener.thread = std::thread([this, req, cb, filter, state]() {
-                sendListen(req, cb, filter, state);
-            });
+    for (auto& l: search.second.listeners) {
+        auto& listener = l.second;
+        auto state = listener.state;
+        if (listener.thread.joinable()) {
+            state->cancel = true;
+            if (listener.req)
+                restbed::Http::close(listener.req);
+            listener.thread.join();
        }
+        // Redo listen
+        state->cancel = false;
+        state->ok = true;
+        auto filter = listener.filter;
+        auto cb = listener.cb;
+        restbed::Uri uri(proxy::HTTP_PROTO + serverHost_ + "/" + search.first.toString());
+        auto req = std::make_shared<restbed::Request>(uri);
+        req->set_method("LISTEN");
+        listener.req = req;
+        listener.thread = std::thread([this, filter, cb, req, state]()
+            {
+                auto settings = std::make_shared<restbed::Settings>();
+                std::chrono::milliseconds timeout(std::numeric_limits<int>::max());
+                settings->set_connection_timeout(timeout); // Avoid the client to close the socket after 5 seconds.
+
+                auto ok = std::make_shared<std::atomic_bool>(true);
+                restbed::Http::async(req,
+                    [this, filter, cb, state](const std::shared_ptr<restbed::Request>& req,
+                                           const std::shared_ptr<restbed::Response>& reply) {
+                    auto code = reply->get_status_code();
+
+                    if (code == 200) {
+                        try {
+                            while (restbed::Http::is_open(req) and not state->cancel) {
+                                restbed::Http::fetch("\n", reply);
+                                if (state->cancel)
+                                    break;
+                                std::string body;
+                                reply->get_body(body);
+                                reply->set_body(""); // Reset the body for the next fetch
+
+                                Json::Value json;
+                                std::string err;
+                                Json::CharReaderBuilder rbuilder;
+                                auto reader = std::unique_ptr<Json::CharReader>(rbuilder.newCharReader());
+                                if (reader->parse(body.data(), body.data() + body.size(), &json, &err)) {
+                                    auto value = std::make_shared<Value>(json);
+                                    if ((not filter or filter(*value)) and cb)  {
+                                        std::lock_guard<std::mutex> lock(lockCallbacks);
+                                        callbacks_.emplace_back([cb, value, state]() {
+                                            if (not state->cancel and not cb({value}, false))
+                                                state->cancel = true;
+                                        });
+                                        loopSignal_();
+                                    }
+                                }
+                            }
+                        } catch (std::runtime_error&) {
+                            // NOTE: Http::close() can occurs here. Ignore this.
+                        }
+                    } else {
+                        state->ok = false;
+                    }
+                }, settings).get();
+                auto& s = *state;
+                if (not s.ok and not s.cancel)
+                    opFailed();
+            }
+        );
+    }
    }
 }

@@ -1003,11 +961,9 @@ DhtProxyClient::pushNotificationReceived(const std::map<std::string, std::string
 {
 #if OPENDHT_PUSH_NOTIFICATIONS
    scheduler.syncTime();
-    {
-        // If a push notification is received, the proxy is up and running
-        std::lock_guard<std::mutex> l(lockCurrentProxyInfos_);
-        statusIpv4_ = NodeStatus::Connected;
-        statusIpv6_ = NodeStatus::Connected;
+    DHT_LOG.d("Got push notification");
+    for (const auto& n : notification) {
+        DHT_LOG.d("   %s -> %s", n.first.c_str(), n.second.c_str());
    }
    try {
        std::lock_guard<std::mutex> lock(searchLock_);
@@ -1020,6 +976,7 @@ DhtProxyClient::pushNotificationReceived(const std::map<std::string, std::string
                // Refresh put
                auto vid = std::stoull(vidIt->second);
                auto& put = search.puts.at(vid);
+                DHT_LOG.d("refresh PermanentPut: %s %llx %p %s", key.to_c_str(), vid, put.refreshJob.get(), put.refreshJob ? (put.refreshJob->do_ ? "set":"not set") : "(null job)");
                scheduler.edit(put.refreshJob, scheduler.time());
                loopSignal_();
            } else {
@@ -1054,7 +1011,6 @@ DhtProxyClient::resubscribe(const InfoHash& key, Listener& listener)
 {
 #if OPENDHT_PUSH_NOTIFICATIONS
    if (deviceKey_.empty()) return;
-    scheduler.syncTime();
    DHT_LOG.d(key, "[search %s] resubscribe push listener", key.to_c_str());
    // Subscribe
    restbed::Uri uri(proxy::HTTP_PROTO + serverHost_ + "/" + key.toString());
@@ -1069,9 +1025,37 @@ DhtProxyClient::resubscribe(const InfoHash& key, Listener& listener)
    state->ok = true;
    listener.req = req;
    listener.pushNotifToken = pushNotifToken;
-    scheduler.edit(listener.refreshJob, scheduler.time() + proxy::OP_TIMEOUT - proxy::OP_MARGIN);
    listener.thread = std::thread([this, req, pushNotifToken, state]() {
-        sendSubscribe(req, pushNotifToken, state);
+        fillBodyToGetToken(req);
+        auto settings = std::make_shared<restbed::Settings>();
+        restbed::Http::async(req,
+            [pushNotifToken, state](const std::shared_ptr<restbed::Request>&,
+                                       const std::shared_ptr<restbed::Response>& reply) {
+            auto code = reply->get_status_code();
+            if (code == 200) {
+                try {
+                    restbed::Http::fetch("\n", reply);
+                    std::string body;
+                    reply->get_body(body);
+
+                    std::string err;
+                    Json::Value json;
+                    Json::CharReaderBuilder rbuilder;
+                    auto reader = std::unique_ptr<Json::CharReader>(rbuilder.newCharReader());
+                    if (reader->parse(body.data(), body.data() + body.size(), &json, &err)) {
+                        if (!json.isMember("token")) return;
+                        *pushNotifToken = unpackId(json, "token");
+                    }
+                } catch (std::runtime_error&) {
+                    // NOTE: Http::close() can occurs here. Ignore this.
+                }
+            } else {
+                state->ok = false;
+            }
+        }, settings).get();
+        auto& s = *state;
+        if (not s.ok and not s.cancel)
+            opFailed();
    });
 #endif
 }
@@ -29,7 +29,6 @@
 #include <chrono>
 #include <functional>
 #include <limits>
-#include <iostream>

 using namespace std::placeholders;

@@ -70,7 +69,6 @@ DhtProxyServer::DhtProxyServer(std::shared_ptr<DhtRunner> dht, in_port_t port ,
        auto resource = std::make_shared<restbed::Resource>();
        resource->set_path("/");
        resource->set_method_handler("GET", std::bind(&DhtProxyServer::getNodeInfo, this, _1));
-        resource->set_method_handler("STATS", std::bind(&DhtProxyServer::getStats, this, _1));
        service_->publish(resource);
        resource = std::make_shared<restbed::Resource>();
        resource->set_path("/{hash: .*}");
@@ -125,6 +123,12 @@ DhtProxyServer::DhtProxyServer(std::shared_ptr<DhtRunner> dht, in_port_t port ,
        while (not service_->is_up() and not stopListeners) {
            std::this_thread::sleep_for(std::chrono::seconds(1));
        }
+        printStatsJob_ = scheduler_.add(scheduler_.time() + PRINT_STATS_PERIOD, [this]{
+            if (service_->is_up() and not stopListeners) {
+                std::cout << getStats().toString() << std::endl;
+                scheduler_.edit(printStatsJob_, scheduler_.time() + PRINT_STATS_PERIOD);
+            }
+        });
        while (service_->is_up()  and not stopListeners) {
            std::unique_lock<std::mutex> lock(schedulerLock_);
            auto next = scheduler_.run();
@@ -134,19 +138,8 @@ DhtProxyServer::DhtProxyServer(std::shared_ptr<DhtRunner> dht, in_port_t port ,
                schedulerCv_.wait_until(lock, next);
        }
    });
+
    dht->forwardAllMessages(true);
-    printStatsJob_ = scheduler_.add(scheduler_.time() + PRINT_STATS_PERIOD, [this] {
-        if (stopListeners) return;
-        if (service_->is_up())
-            updateStats();
-        // Refresh stats cache
-        auto newInfo = dht_->getNodeInfo();
-        {
-            std::lock_guard<std::mutex> lck(statsMutex_);
-            nodeInfo_ = std::move(newInfo);
-        }
-        scheduler_.edit(printStatsJob_, scheduler_.time() + PRINT_STATS_PERIOD);
-    });
 }

 DhtProxyServer::~DhtProxyServer()
@@ -165,7 +158,7 @@ DhtProxyServer::stop()
        auto listener = currentListeners_.begin();
        while (listener != currentListeners_.end()) {
            listener->session->close();
-            ++listener;
+            ++ listener;
        }
    }
    stopListeners = true;
@@ -179,20 +172,21 @@ DhtProxyServer::stop()
        server_thread.join();
 }

-void
-DhtProxyServer::updateStats() const
+DhtProxyServer::ServerStats
+DhtProxyServer::getStats() const
 {
+    ServerStats ret {};
    auto now = clock::now();
    auto last = lastStatsReset_.exchange(now);
    auto count = requestNum_.exchange(0);
    auto dt = std::chrono::duration<double>(now - last);
-    stats_.requestRate = count / dt.count();
+    ret.requestRate = count / dt.count();
 #if OPENDHT_PUSH_NOTIFICATIONS
-    stats_.pushListenersCount = pushListeners_.size();
+    ret.pushListenersCount = pushListeners_.size();
 #endif
-    stats_.putCount = puts_.size();
-    stats_.listenCount = currentListeners_.size();
-    stats_.nodeInfo = nodeInfo_;
+    ret.putCount = puts_.size();
+    ret.listenCount = currentListeners_.size();
+    return ret;
 }

 void
@@ -202,22 +196,17 @@ DhtProxyServer::getNodeInfo(const std::shared_ptr<restbed::Session>& session) co
    const auto request = session->get_request();
    int content_length = std::stoi(request->get_header("Content-Length", "0"));
    session->fetch(content_length,
-        [this](const std::shared_ptr<restbed::Session> s, const restbed::Bytes& /*b*/) mutable
+        [=](const std::shared_ptr<restbed::Session> s, const restbed::Bytes& /*b*/)
        {
            try {
                if (dht_) {
                    Json::Value result;
-                    {
-                        std::lock_guard<std::mutex> lck(statsMutex_);
-                        if (nodeInfo_.ipv4.good_nodes == 0 && nodeInfo_.ipv6.good_nodes == 0) {
-                            // NOTE: we want to avoid the disconnected state as much as possible
-                            // So, if the node is disconnected, we should force the update of the cache
-                            // and reconnect as soon as possible
-                            // This should not happen much
-                            nodeInfo_ = dht_->getNodeInfo();
-                        }
-                        result = nodeInfo_.toJson();
-                    }
+                    auto id = dht_->getId();
+                    if (id)
+                        result["id"] = id.toString();
+                    result["node_id"] = dht_->getNodeId().toString();
+                    result["ipv4"] = dht_->getNodesStats(AF_INET).toJson();
+                    result["ipv6"] = dht_->getNodesStats(AF_INET6).toJson();
                    result["public_ip"] = s->get_origin(); // [ipv6:ipv4]:port or ipv4:port
                    Json::StreamWriterBuilder wbuilder;
                    wbuilder["commentStyle"] = "None";
@@ -234,36 +223,6 @@ DhtProxyServer::getNodeInfo(const std::shared_ptr<restbed::Session>& session) co
    );
 }

-void
-DhtProxyServer::getStats(const std::shared_ptr<restbed::Session>& session) const
-{
-    requestNum_++;
-    const auto request = session->get_request();
-    int content_length = std::stoi(request->get_header("Content-Length", "0"));
-    session->fetch(content_length,
-        [this](const std::shared_ptr<restbed::Session> s, const restbed::Bytes& /*b*/) mutable
-        {
-            try {
-                if (dht_) {
-#ifdef OPENDHT_JSONCPP
-                    Json::StreamWriterBuilder wbuilder;
-                    wbuilder["commentStyle"] = "None";
-                    wbuilder["indentation"] = "";
-                    auto output = Json::writeString(wbuilder, stats_.toJson()) + "\n";
-                    s->close(restbed::OK, output);
-#else
-                    s->close(restbed::NotFound, "{\"err\":\"JSON not enabled on this instance\"}");
-#endif
-                }
-                else
-                    s->close(restbed::SERVICE_UNAVAILABLE, "{\"err\":\"Incorrect DhtRunner\"}");
-            } catch (...) {
-                s->close(restbed::INTERNAL_SERVER_ERROR, "{\"err\":\"Internal server error\"}");
-            }
-        }
-    );
-}
-
 void
 DhtProxyServer::get(const std::shared_ptr<restbed::Session>& session) const
 {
@@ -402,10 +361,7 @@ DhtProxyServer::subscribe(const std::shared_ptr<restbed::Session>& session)
                    return;
                }
                auto pushToken = root["key"].asString();
-                if (pushToken.empty()) {
-                    s->close(restbed::BAD_REQUEST, "{\"err\":\"No token\"}");
-                    return;
-                }
+                if (pushToken.empty()) return;
                auto tokenFromReq = unpackId(root, "token");
                auto platform = root["platform"].asString();
                auto isAndroid = platform == "android";
@@ -643,6 +599,7 @@ DhtProxyServer::put(const std::shared_ptr<restbed::Session>& session)
                                    clientId = pVal["client_id"].asString();
                                    platform = pVal["platform"].asString();
                                }
+                                std::cout << "permanent: pushToken " << pushToken << " clientId " << clientId << " platform " << platform << std::endl;
                                bool isAndroid = platform == "android";
                                std::unique_lock<std::mutex> lock(schedulerLock_);
                                scheduler_.syncTime();
@@ -652,16 +609,18 @@ DhtProxyServer::put(const std::shared_ptr<restbed::Session>& session)
                                auto r = sPuts->second.puts.emplace(vid, PermanentPut{});
                                auto& pput = r.first->second;
                                if (r.second) {
+                                    std::cout << "New ppush entry" << std::endl;
                                    pput.expireJob = scheduler_.add(timeout, [this, infoHash, vid]{
                                        std::cout << "Permanent put expired: " << infoHash << " " << vid << std::endl;
                                        cancelPut(infoHash, vid);
                                    });
 #if OPENDHT_PUSH_NOTIFICATIONS
                                    if (not pushToken.empty()) {
+                                        std::cout << "Adding refresh job" << std::endl;
                                        pput.expireNotifyJob = scheduler_.add(timeout - proxy::OP_MARGIN,
                                            [this, infoHash, vid, pushToken, clientId, isAndroid]
                                        {
-                                            std::cout << "Permanent put refresh: " << infoHash << " " << vid << std::endl;
+                                            std::cout << "Permanent put refresh: " << infoHash << " " << std::to_string(vid) << " " << clientId << std::endl;
                                            Json::Value json;
                                            json["timeout"] = infoHash.toString();
                                            json["to"] = clientId;
@@ -671,6 +630,7 @@ DhtProxyServer::put(const std::shared_ptr<restbed::Session>& session)
                                    }
 #endif
                                } else {
+                                    std::cout << "Refreshing put jobs" << std::endl;
                                    scheduler_.edit(pput.expireJob, timeout);
                                    if (pput.expireNotifyJob)
                                        scheduler_.edit(pput.expireNotifyJob, timeout - proxy::OP_MARGIN);
@@ -42,7 +42,6 @@
 namespace dht {

 constexpr std::chrono::seconds DhtRunner::BOOTSTRAP_PERIOD;
-static constexpr size_t RX_QUEUE_MAX_SIZE = 1024 * 16;

 struct DhtRunner::Listener {
    size_t tokenClassicDht;
@@ -164,7 +163,7 @@ DhtRunner::shutdown(ShutdownCallback cb) {
 void
 DhtRunner::join()
 {
-    stopNetwork();
+    running_network = false;
    running = false;
    cv.notify_all();
    bootstrap_cv.notify_all();
@@ -248,10 +247,10 @@ void
 DhtRunner::setLoggers(LogMethod error, LogMethod warn, LogMethod debug) {
    std::lock_guard<std::mutex> lck(dht_mtx);
    if (dht_)
-        dht_->setLoggers(error, warn, debug);
+        dht_->setLoggers(std::forward<LogMethod>(error), std::forward<LogMethod>(warn), std::forward<LogMethod>(debug));
 #if OPENDHT_PROXY_CLIENT
    if (dht_via_proxy_)
-        dht_via_proxy_->setLoggers(error, warn, debug);
+        dht_via_proxy_->setLoggers(std::forward<LogMethod>(error), std::forward<LogMethod>(warn), std::forward<LogMethod>(debug));
 #endif
 }

@@ -302,17 +301,6 @@ DhtRunner::getNodesStats(sa_family_t af) const
    return activeDht()->getNodesStats(af);
 }

-NodeInfo
-DhtRunner::getNodeInfo() const {
-    std::lock_guard<std::mutex> lck(dht_mtx);
-    NodeInfo info;
-    info.id = getId();
-    info.node_id = getNodeId();
-    info.ipv4 = dht_->getNodesStats(AF_INET);
-    info.ipv6 = dht_->getNodesStats(AF_INET6);
-    return info;
-}
-
 std::vector<unsigned>
 DhtRunner::getNodeMessageStats(bool in) const
 {
@@ -354,9 +342,7 @@ std::vector<SockAddr>
 DhtRunner::getPublicAddress(sa_family_t af)
 {
    std::lock_guard<std::mutex> lck(dht_mtx);
-    if (auto dht = activeDht())
-        return dht->getPublicAddress(af);
-    return {};
+    return activeDht()->getPublicAddress(af);
 }
 std::vector<std::string>
 DhtRunner::getPublicAddressStr(sa_family_t af)
@@ -410,15 +396,12 @@ DhtRunner::loop_()
        received = std::move(rcv);
    }
    if (not received.empty()) {
-        while (not received.empty()) {
-            auto& pck = received.front();
-            auto delay = clock::now() - pck.received;
-            if (delay > std::chrono::milliseconds(500))
-                std::cerr << "Dropping packet with high delay: " << print_dt(delay) << std::endl;
-            else
-                wakeup = dht->periodic(pck.data.data(), pck.data.size()-1, pck.from);
-            received.pop();
+        for (const auto& pck : received) {
+            auto& buf = pck.first;
+            auto& from = pck.second;
+            wakeup = dht->periodic(buf.data(), buf.size()-1, from);
        }
+        received.clear();
    } else {
        wakeup = dht->periodic(nullptr, 0, nullptr, 0);
    }
@@ -458,10 +441,8 @@ int bindSocket(const SockAddr& addr, SockAddr& bound)
    if (is_ipv6)
        setsockopt(sock, IPPROTO_IPV6, IPV6_V6ONLY, (const char*)&set, sizeof(set));
    int rc = bind(sock, addr.get(), addr.getLength());
-    if(rc < 0) {
-        close(sock);
+    if(rc < 0)
        throw DhtException("Can't bind socket on " + addr.toString() + " " + strerror(rc));
-    }
    sockaddr_storage ss;
    socklen_t ss_len = sizeof(ss);
    getsockname(sock, (sockaddr*)&ss, &ss_len);
@@ -469,64 +450,12 @@ int bindSocket(const SockAddr& addr, SockAddr& bound)
    return sock;
 }

-#ifdef _WIN32
-inline void udpPipe(int fds[2])
-{
-    int lst = socket(AF_INET, SOCK_DGRAM, 0);
-    if (lst < 0)
-        throw DhtException(std::string("Can't open socket: ") + strerror(lst));
-    sockaddr_in inaddr;
-    sockaddr addr;
-    memset(&inaddr, 0, sizeof(inaddr));
-    memset(&addr, 0, sizeof(addr));
-    inaddr.sin_family = AF_INET;
-    inaddr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
-    inaddr.sin_port = 0;
-    int yes=1;
-    setsockopt(lst, SOL_SOCKET, SO_REUSEADDR, (char*)&yes, sizeof(yes));
-    int rc = bind(lst, (sockaddr*)&inaddr, sizeof(inaddr));
-    if (rc < 0) {
-        close(lst);
-        throw DhtException("Can't bind socket on " + print_addr((sockaddr*)&addr, sizeof(inaddr)) + " " + strerror(rc));
-    }
-    socklen_t len = sizeof(inaddr);
-    getsockname(lst, &addr, &len);
-    fds[0] = lst;
-    fds[1] = socket(AF_INET, SOCK_DGRAM, 0);
-    connect(fds[1], &addr, len);
-}
-#endif
-
-void
-DhtRunner::stopNetwork()
-{
-    running_network = false;
-    if (stop_writefd != -1) {
-        if (write(stop_writefd, "\0", 1) == -1) {
-            perror("write");
-        }
-    }
-}
-
 void
 DhtRunner::startNetwork(const SockAddr sin4, const SockAddr sin6)
 {
-    stopNetwork();
+    running_network = false;
    if (rcv_thread.joinable())
        rcv_thread.join();
-
-    int stopfds[2];
-#ifndef _WIN32
-    auto status = pipe(stopfds);
-    if (status == -1) {
-        throw DhtException("Can't open pipe");
-    }
-#else
-    udpPipe(stopfds);
-#endif
-    int stop_readfd = stopfds[0];
-    stop_writefd = stopfds[1];
-
    s4 = -1;
    s6 = -1;

@@ -541,19 +470,19 @@ DhtRunner::startNetwork(const SockAddr sin4, const SockAddr sin6)
 #endif

    running_network = true;
-    rcv_thread = std::thread([this, stop_readfd]() {
+    rcv_thread = std::thread([this]() {
        try {
            while (running_network) {
+                struct timeval tv {/*.tv_sec = */0, /*.tv_usec = */250000};
                fd_set readfds;

                FD_ZERO(&readfds);
-                FD_SET(stop_readfd, &readfds);
                if(s4 >= 0)
                    FD_SET(s4, &readfds);
                if(s6 >= 0)
                    FD_SET(s6, &readfds);

-                int rc = select(s4 > s6 ? s4 + 1 : s6 + 1, &readfds, nullptr, nullptr, nullptr);
+                int rc = select(s4 > s6 ? s4 + 1 : s6 + 1, &readfds, nullptr, nullptr, &tv);
                if(rc < 0) {
                    if(errno != EINTR) {
                        perror("select");
@@ -574,15 +503,11 @@ DhtRunner::startNetwork(const SockAddr sin4, const SockAddr sin6)
                    else if(s6 >= 0 && FD_ISSET(s6, &readfds))
                        rc = recvfrom(s6, (char*)buf.data(), buf.size(), 0, (sockaddr*)&from, &from_len);
                    else
-                        continue;
+                        break;
                    if (rc > 0) {
                        {
                            std::lock_guard<std::mutex> lck(sock_mtx);
-                            if (rcv.size() >= RX_QUEUE_MAX_SIZE) {
-                                std::cerr << "Dropping packet: queue is full!" << std::endl;
-                                rcv.pop();
-                            }
-                            rcv.emplace(ReceivedPacket {Blob {buf.begin(), buf.begin()+rc+1}, SockAddr(from, from_len), clock::now()});
+                            rcv.emplace_back(Blob {buf.begin(), buf.begin()+rc+1}, SockAddr(from, from_len));
                        }
                        cv.notify_all();
                    }
@@ -599,11 +524,6 @@ DhtRunner::startNetwork(const SockAddr sin4, const SockAddr sin6)
        s6 = -1;
        bound4 = {};
        bound6 = {};
-        if (stop_readfd != -1)
-            close(stop_readfd);
-        if (stop_writefd != -1)
-            close(stop_writefd);
-        stop_writefd = -1;
    });
 }

@@ -642,7 +562,6 @@ DhtRunner::listen(InfoHash hash, ValueCallback vcb, Value::Filter f, Where w)
    {
        std::lock_guard<std::mutex> lck(storage_mtx);
        pending_ops.emplace([=](SecureDht& dht) mutable {
-#if OPENDHT_PROXY_CLIENT
            auto tokenbGlobal = listener_token_++;
            Listener listener {};
            listener.hash = hash;
@@ -650,20 +569,21 @@ DhtRunner::listen(InfoHash hash, ValueCallback vcb, Value::Filter f, Where w)
            listener.w = std::move(w);
            listener.gcb = [hash,vcb,tokenbGlobal,this](const std::vector<Sp<Value>>& vals, bool expired){
                if (not vcb(vals, expired)) {
+#if OPENDHT_PROXY_CLIENT
                    cancelListen(hash, tokenbGlobal);
+#endif
                    return false;
                }
                return true;
            };
+#if OPENDHT_PROXY_CLIENT
            if (use_proxy)
                listener.tokenProxyDht = dht.listen(hash, listener.gcb, listener.f, listener.w);
            else
+#endif
                listener.tokenClassicDht = dht.listen(hash, listener.gcb, listener.f, listener.w);
            listeners_.emplace(tokenbGlobal, std::move(listener));
            ret_token->set_value(tokenbGlobal);
-#else
-            ret_token->set_value(dht.listen(hash, vcb, f, w));
-#endif
        });
    }
    cv.notify_all();
@@ -679,24 +599,20 @@ DhtRunner::listen(const std::string& key, GetCallback vcb, Value::Filter f, Wher
 void
 DhtRunner::cancelListen(InfoHash h, size_t token)
 {
-    {
-        std::lock_guard<std::mutex> lck(storage_mtx);
+    std::lock_guard<std::mutex> lck(storage_mtx);
+    pending_ops.emplace([=](SecureDht& dht) {
 #if OPENDHT_PROXY_CLIENT
-        pending_ops.emplace([=](SecureDht&) {
-            auto it = listeners_.find(token);
-            if (it == listeners_.end()) return;
-            if (it->second.tokenClassicDht)
-                dht_->cancelListen(h, it->second.tokenClassicDht);
-            if (it->second.tokenProxyDht and dht_via_proxy_)
-                dht_via_proxy_->cancelListen(h, it->second.tokenProxyDht);
-            listeners_.erase(it);
-        });
+        auto it = listeners_.find(token);
+        if (it == listeners_.end()) return;
+        if (it->second.tokenClassicDht)
+            dht_->cancelListen(h, it->second.tokenClassicDht);
+        if (it->second.tokenProxyDht and dht_via_proxy_)
+            dht_via_proxy_->cancelListen(h, it->second.tokenProxyDht);
+        listeners_.erase(it);
 #else
-        pending_ops.emplace([=](SecureDht& dht) {
-            dht.cancelListen(h, token);
-        });
+        dht.cancelListen(h, token);
 #endif // OPENDHT_PROXY_CLIENT
-    }
+    });
    cv.notify_all();
 }

@@ -705,21 +621,20 @@ DhtRunner::cancelListen(InfoHash h, std::shared_future<size_t> ftoken)
 {
    {
        std::lock_guard<std::mutex> lck(storage_mtx);
+        pending_ops.emplace([=](SecureDht& dht) {
+            auto token = ftoken.get();
 #if OPENDHT_PROXY_CLIENT
-        pending_ops.emplace([=](SecureDht&) {
-            auto it = listeners_.find(ftoken.get());
+            auto it = listeners_.find(token);
            if (it == listeners_.end()) return;
            if (it->second.tokenClassicDht)
                dht_->cancelListen(h, it->second.tokenClassicDht);
            if (it->second.tokenProxyDht and dht_via_proxy_)
                dht_via_proxy_->cancelListen(h, it->second.tokenProxyDht);
            listeners_.erase(it);
-        });
 #else
-        pending_ops.emplace([=](SecureDht& dht) {
-            dht.cancelListen(h, ftoken.get());
-        });
+            dht.cancelListen(h, token);
 #endif // OPENDHT_PROXY_CLIENT
+        });
    }
    cv.notify_all();
 }
@@ -936,7 +851,7 @@ DhtRunner::connectivityChanged()
 {
    {
        std::lock_guard<std::mutex> lck(storage_mtx);
-        pending_ops_prio.emplace([=](SecureDht& dht) {
+        pending_ops.emplace([=](SecureDht& dht) {
            dht.connectivityChanged();
        });
    }
@@ -957,8 +872,8 @@ DhtRunner::findCertificate(InfoHash hash, std::function<void(const std::shared_p
 void
 DhtRunner::resetDht()
 {
-#if OPENDHT_PROXY_CLIENT
    listeners_.clear();
+#if OPENDHT_PROXY_CLIENT
    dht_via_proxy_.reset();
 #endif // OPENDHT_PROXY_CLIENT
    dht_.reset();
@@ -974,21 +889,6 @@ DhtRunner::activeDht() const
 #endif // OPENDHT_PROXY_CLIENT
 }

-void
-DhtRunner::setProxyServer(const std::string& proxy, const std::string& pushNodeId)
-{
-#if OPENDHT_PROXY_CLIENT
-    if (config_.proxy_server == proxy and config_.push_node_id == pushNodeId)
-        return;
-    config_.proxy_server = proxy;
-    config_.push_node_id = pushNodeId;
-    enableProxy(use_proxy and not config_.proxy_server.empty());
-#else
-    if (not proxy.empty())
-        std::cerr << "DHT proxy requested but OpenDHT built without proxy support." << std::endl;
-#endif
-}
-
 void
 DhtRunner::enableProxy(bool proxify)
 {
@@ -1071,7 +971,7 @@ DhtRunner::pushNotificationReceived(const std::map<std::string, std::string>& da
 #if OPENDHT_PROXY_CLIENT && OPENDHT_PUSH_NOTIFICATIONS
    {
        std::lock_guard<std::mutex> lck(storage_mtx);
-        pending_ops_prio.emplace([=](SecureDht&) {
+        pending_ops_prio.emplace([=](SecureDht& dht) {
            if (dht_via_proxy_)
                dht_via_proxy_->pushNotificationReceived(data);
        });
@@ -262,13 +262,7 @@ NetworkEngine::isRunning(sa_family_t af) const

 void
 NetworkEngine::clear()
-{
-    for (auto& request : requests) {
-        request.second->cancel();
-        request.second->node->setExpired();
-    }
-    requests.clear();
-}
+{}

 void
 NetworkEngine::connectivityChanged(sa_family_t af)
@@ -286,7 +280,7 @@ NetworkEngine::requestStep(Sp<Request> sreq)
    auto now = scheduler.time();
    auto& node = *req.node;
    if (req.isExpired(now)) {
-        DHT_LOG.d(node.id, "[node %s] expired !", node.toString().c_str());
+        DHT_LOG.e(node.id, "[node %s] expired !", node.toString().c_str());
        node.setExpired();
        if (not node.id)
            requests.erase(req.tid);
@@ -298,21 +292,24 @@ NetworkEngine::requestStep(Sp<Request> sreq)
    auto err = send((char*)req.msg.data(), req.msg.size(),
            (node.getReplyTime() >= now - UDP_REPLY_TIME) ? 0 : MSG_CONFIRM,
            node.getAddr());
+    std::weak_ptr<Request> wreq = sreq;
    if (err == ENETUNREACH  ||
        err == EHOSTUNREACH ||
        err == EAFNOSUPPORT ||
-        err == EPIPE        ||
-        err == EPERM)
+        err == EPIPE)
    {
-        node.setExpired();
-        if (not node.id)
-            requests.erase(req.tid);
+        scheduler.add(now, [this,wreq]() mutable {
+            if (auto req = wreq.lock()) {
+                req->node->setExpired();
+                if (not req->node->id)
+                    requests.erase(req->tid);
+            }
+        });
    } else {
        if (err != EAGAIN) {
            ++req.attempt_count;
        }
        req.last_try = now;
-        std::weak_ptr<Request> wreq = sreq;
        scheduler.add(req.last_try + Node::MAX_RESPONSE_TIME, [this,wreq] {
            if (auto req = wreq.lock())
                requestStep(req);
@@ -661,7 +658,7 @@ NetworkEngine::send(const char *buf, size_t len, int flags, const SockAddr& addr
    if (s < 0)
        return EAFNOSUPPORT;
 #ifdef MSG_NOSIGNAL
-    flags |= MSG_NOSIGNAL;
+    flags &= MSG_NOSIGNAL;
 #endif
    if (sendto(s, buf, len, flags, addr.get(), addr.getLength()) == -1) {
        int err = errno;
@@ -57,15 +57,6 @@ struct Request {
    bool pending() const { return state_ == State::PENDING; }
    bool over() const { return not pending(); }
    State getState() const { return state_; }
-    char getStateChar() const {
-        switch (state_) {
-            case State::PENDING:   return 'f';
-            case State::CANCELLED: return 'c';
-            case State::EXPIRED:   return 'e';
-            case State::COMPLETED: return 'a';
-            default:               return '?';
-        }
-    }

    Request(State state = State::PENDING) : state_(state) {}
    Request(MessageType type, Tid tid,
@@ -136,7 +136,7 @@ struct Dht::SearchNode {
     *
     * @return true if we can send get, else false.
     */
-    bool canGet(time_point now, time_point update, const Sp<Query>& q) const {
+    bool canGet(time_point now, time_point update, Sp<Query> q = {}) const {
        if (node->isExpired())
            return false;

@@ -147,9 +147,9 @@ struct Dht::SearchNode {
            if (s.second and s.second->pending())
                pending = true;
            if (s.first and q and q->isSatisfiedBy(*s.first) and s.second) {
-                if (s.second->pending())
+                if (s.second->pending() and not pending_sq_status)
                    pending_sq_status = true;
-                else if (s.second->completed() and not (update > s.second->reply_time))
+                if (s.second->completed() and not (update > s.second->reply_time) and not completed_sq_status)
                    completed_sq_status = true;
                if (completed_sq_status and pending_sq_status)
                    break;
@@ -157,7 +157,7 @@ struct Dht::SearchNode {
        }

        return (not pending and now > last_get_reply + Node::NODE_EXPIRE_TIME) or
-                not (completed_sq_status or pending_sq_status or hasStartedPagination(q));
+                not (hasStartedPagination(q) or completed_sq_status or pending_sq_status);
    }

    /**
@@ -302,7 +302,7 @@ struct Dht::SearchNode {
            return isListening(now, ls);
    }
    bool isListening(time_point now, NodeListenerStatus::const_iterator listen_status) const {
-        if (listen_status == listenStatus.end() or not listen_status->second.req)
+        if (listen_status == listenStatus.end())
            return false;
        return listen_status->second.req->reply_time + LISTEN_EXPIRE_TIME > now;
    }
@@ -320,7 +320,7 @@ struct Dht::SearchNode {
    }

    /**
-     * Assuming the node is synced, should a "put" request be sent to this node now ?
+     * Assumng the node is synced, should a "put" request be sent to this node now ?
     */
    time_point getAnnounceTime(Value::Id vid) const {
        const auto& ack = acked.find(vid);
@@ -331,11 +331,11 @@ struct Dht::SearchNode {
        return ((gs != getStatus.cend() and gs->second and gs->second->pending())
                or ack == acked.cend() or not ack->second.first or ack->second.first->pending())
                ? time_point::max()
-                : (ack->second.first->completed() ? ack->second.second - REANNOUNCE_MARGIN : time_point::min());
+                : ack->second.second - REANNOUNCE_MARGIN;
    }

    /**
-     * Assuming the node is synced, should the "listen" request with Query q be
+     * Assumng the node is synced, should the "listen" request with Query q be
     * sent to this node now ?
     */
    time_point getListenTime(const Sp<Query>& q) const {
@@ -430,8 +430,7 @@ struct Dht::Search {
     *
     * @param query  The query identifying a 'get' request.
     */
-    time_point getLastGetTime(const Query&) const;
-    time_point getLastGetTime() const;
+    time_point getLastGetTime(Sp<Query> query = {}) const;

    /**
     * Is this get operation done ?
@@ -747,20 +746,11 @@ Dht::Search::isSynced(time_point now) const
 }

 time_point
-Dht::Search::getLastGetTime(const Query& q) const
+Dht::Search::getLastGetTime(Sp<Query> q) const
 {
    time_point last = time_point::min();
    for (const auto& g : callbacks)
-        last = std::max(last, (q.isSatisfiedBy(*g.second.query) ? g.second.start : time_point::min()));
-    return last;
-}
-
-time_point
-Dht::Search::getLastGetTime() const
-{
-    time_point last = time_point::min();
-    for (const auto& g : callbacks)
-        last = std::max(last, g.second.start);
+        last = std::max(last, (not q or q->isSatisfiedBy(*g.second.query) ? g.second.start : time_point::min()));
    return last;
 }

@@ -154,16 +154,15 @@ struct Storage {
     *
     * @param now  The reference to now
     * @param vid  The value id
-     * @return time of the next expiration, time_point::max() if no expiration
+     * @return true if a value storage was updated, false otherwise
     */
-    time_point refresh(const time_point& now, const Value::Id& vid, const TypeStore& types) {
+    bool refresh(const time_point& now, const Value::Id& vid) {
        for (auto& vs : values)
            if (vs.data->id == vid) {
                vs.created = now;
-                vs.expiration = std::max(vs.expiration, now + types.getType(vs.data->type).expiration);
-                return vs.expiration;
+                return true;
            }
-        return time_point::max();
+        return false;
    }

    StoreDiff remove(const InfoHash& id, Value::Id);
@@ -22,9 +22,9 @@
 #include "default_types.h"
 #include "securedht.h" // print certificate ID

-#ifdef OPENDHT_JSONCPP
+#if OPENDHT_PROXY_SERVER || OPENDHT_PROXY_CLIENT
 #include "base64.h"
-#endif
+#endif //OPENDHT_PROXY_SERVER


 namespace dht {
@@ -172,7 +172,7 @@ Value::msgpack_unpack_body(const msgpack::object& o)
    }
 }

-#ifdef OPENDHT_JSONCPP
+#if OPENDHT_PROXY_SERVER || OPENDHT_PROXY_CLIENT
 Value::Value(Json::Value& json)
 {
    id = Value::Id(unpackId(json, "id"));
@@ -248,7 +248,7 @@ unpackId(const Json::Value& json, const std::string& key) {
    } catch (...) {}
    return ret;
 }
-#endif
+#endif //OPENDHT_PROXY_SERVER

 bool
 FieldValue::operator==(const FieldValue& vfd) const
@@ -66,7 +66,7 @@ public:
    CallbackQueue expireValues(const time_point& now, time_point& next) {
        std::vector<Sp<Value>> expired_values;
        for (auto it = values.begin(); it != values.end();) {
-            if (it->second.expiration <= now) {
+            if (it->second.expiration < now) {
                expired_values.emplace_back(std::move(it->second.data));
                it = values.erase(it);
            } else {
@@ -74,7 +74,7 @@ public:
                ++it;
            }
        }
-        while (values.size() > MAX_VALUES) {
+        while (values.size() >= MAX_VALUES) {
            // too many values, remove oldest values
            time_point oldest_creation = time_point::max();
            auto oldest_value = values.end();
@@ -127,7 +127,7 @@ private:
    ValueCache& operator=(ValueCache&&) = delete;

    /* The maximum number of values we store in the cache. */
-    static constexpr unsigned MAX_VALUES {4096};
+    static constexpr unsigned MAX_VALUES {1024};

    struct CacheValueStorage {
        Sp<Value> data {};
@@ -3,7 +3,7 @@ bin_PROGRAMS = opendht_unit_tests

 AM_CPPFLAGS = -I../include

-nobase_include_HEADERS = infohashtester.h cryptotester.h dhtrunnertester.h dhtproxytester.h
-opendht_unit_tests_SOURCES = tests_runner.cpp cryptotester.cpp infohashtester.cpp dhtrunnertester.cpp dhtproxytester.cpp
+nobase_include_HEADERS = infohashtester.h dhtrunnertester.h
+opendht_unit_tests_SOURCES = tests_runner.cpp infohashtester.cpp dhtrunnertester.cpp
 opendht_unit_tests_LDFLAGS = -lopendht -lcppunit -L@top_builddir@/src/.libs @GnuTLS_LIBS@
 endif
@@ -1,90 +0,0 @@
-/*
- *  Copyright (C) 2018 Savoir-faire Linux Inc.
- *
- *  Author: Adrien Béraud <adrien.beraud@savoirfairelinux.com>
- *
- *  This program is free software; you can redistribute it and/or modify
- *  it under the terms of the GNU General Public License as published by
- *  the Free Software Foundation; either version 3 of the License, or
- *  (at your option) any later version.
- *
- *  This program is distributed in the hope that it will be useful,
- *  but WITHOUT ANY WARRANTY; without even the implied warranty of
- *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- *  GNU General Public License for more details.
- *
- *  You should have received a copy of the GNU General Public License
- *  along with this program. If not, see <https://www.gnu.org/licenses/>.
- */
-
-#include "cryptotester.h"
-
-#include "opendht/crypto.h"
-
-namespace test {
-CPPUNIT_TEST_SUITE_REGISTRATION(CryptoTester);
-
-void
-CryptoTester::setUp() {
-
-}
-
-void
-CryptoTester::testSignatureEncryption() {
-    auto key = dht::crypto::PrivateKey::generate();
-    auto public_key = key.getPublicKey();
-
-    std::vector<uint8_t> data {5, 10};
-    std::vector<uint8_t> signature = key.sign(data);
-
-    // check signature
-    CPPUNIT_ASSERT(public_key.checkSignature(data, signature));
-
-    // encrypt data
-    std::vector<uint8_t> encrypted = public_key.encrypt(data);
-    std::vector<uint8_t> decrypted = key.decrypt(encrypted);
-    CPPUNIT_ASSERT(data == decrypted);
-}
-
-void
-CryptoTester::testCertificateRevocation()
-{
-    auto ca1 = dht::crypto::generateIdentity("ca1");
-    auto account1 = dht::crypto::generateIdentity("acc1", ca1, 4096, true);
-    auto device11 = dht::crypto::generateIdentity("dev11", account1);
-    auto device12 = dht::crypto::generateIdentity("dev12", account1);
-
-
-    dht::crypto::TrustList list;
-    list.add(*ca1.second);
-    auto v = list.verify(*account1.second);
-    CPPUNIT_ASSERT_MESSAGE(v.toString(), v);
-
-    list.add(*account1.second);
-    v = list.verify(*device11.second);
-    CPPUNIT_ASSERT_MESSAGE(v.toString(), v);
-    v = list.verify(*device12.second);
-    CPPUNIT_ASSERT_MESSAGE(v.toString(), v);
-
-    auto ca2 = dht::crypto::generateIdentity("ca2");
-    auto account2 = dht::crypto::generateIdentity("acc2", ca2, 4096, true);
-    auto device2 = dht::crypto::generateIdentity("dev2", account2);
-
-    v = list.verify(*device2.second);
-    CPPUNIT_ASSERT_MESSAGE(v.toString(), !v);
-    
-    account1.second->revoke(*account1.first, *device11.second);
-    dht::crypto::TrustList list2;
-    list2.add(*account1.second);
-
-    v = list2.verify(*device11.second);
-    CPPUNIT_ASSERT_MESSAGE(v.toString(), !v);
-    v = list2.verify(*device12.second);
-    CPPUNIT_ASSERT_MESSAGE(v.toString(), v);
-}
-
-void
-CryptoTester::tearDown() {
-
-}
-}  // namespace test
@@ -1,53 +0,0 @@
-/*
- *  Copyright (C) 2018 Savoir-faire Linux Inc.
- *
- *  Author: Adrien Béraud <adrien.beraud@savoirfairelinux.com>
- *
- *  This program is free software; you can redistribute it and/or modify
- *  it under the terms of the GNU General Public License as published by
- *  the Free Software Foundation; either version 3 of the License, or
- *  (at your option) any later version.
- *
- *  This program is distributed in the hope that it will be useful,
- *  but WITHOUT ANY WARRANTY; without even the implied warranty of
- *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- *  GNU General Public License for more details.
- *
- *  You should have received a copy of the GNU General Public License
- *  along with this program. If not, see <https://www.gnu.org/licenses/>.
- */
-
-#pragma once
-
-// cppunit
-#include <cppunit/TestFixture.h>
-#include <cppunit/extensions/HelperMacros.h>
-
-namespace test {
-
-class CryptoTester : public CppUnit::TestFixture {
-    CPPUNIT_TEST_SUITE(CryptoTester);
-    CPPUNIT_TEST(testSignatureEncryption);
-    CPPUNIT_TEST(testCertificateRevocation);
-    CPPUNIT_TEST_SUITE_END();
-
- public:
-    /**
-     * Method automatically called before each test by CppUnit
-     */
-    void setUp();
-    /**
-     * Method automatically called after each test CppUnit
-     */
-    void tearDown();
-    /**
-     * Test data signature, encryption and decryption
-     */
-    void testSignatureEncryption();
-    /**
-     * Test certificate generation, validation and revocation
-     */
-    void testCertificateRevocation();
-};
-
-}  // namespace test
@@ -1,126 +0,0 @@
-/*
- *  Copyright (C) 2018 Savoir-faire Linux Inc.
- *
- *  Author: Sébastien Blin <sebastien.blin@savoirfairelinux.com>
- *
- *  This program is free software; you can redistribute it and/or modify
- *  it under the terms of the GNU General Public License as published by
- *  the Free Software Foundation; either version 3 of the License, or
- *  (at your option) any later version.
- *
- *  This program is distributed in the hope that it will be useful,
- *  but WITHOUT ANY WARRANTY; without even the implied warranty of
- *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- *  GNU General Public License for more details.
- *
- *  You should have received a copy of the GNU General Public License
- *  along with this program. If not, see <https://www.gnu.org/licenses/>.
- */
-
-#include "dhtproxytester.h"
-
-// std
-#include <iostream>
-#include <string>
-
-#include <chrono>
-#include <condition_variable>
-
-
-namespace test {
-CPPUNIT_TEST_SUITE_REGISTRATION(DhtProxyTester);
-
-void
-DhtProxyTester::setUp() {
-    nodePeer.run(42222, {}, true);
-    nodeProxy = std::make_shared<dht::DhtRunner>();
-    nodeClient = std::make_shared<dht::DhtRunner>();
-
-    nodeProxy->run(42232, {}, true);
-    nodeProxy->bootstrap(nodePeer.getBound());
-    server = std::unique_ptr<dht::DhtProxyServer>(new dht::DhtProxyServer(nodeProxy, 8080));
-
-    nodeClient->run(42242, {}, true);
-    nodeClient->bootstrap(nodePeer.getBound());
-    nodeClient->setProxyServer("127.0.0.1:8080");
-    nodeClient->enableProxy(true);
-}
-
-void
-DhtProxyTester::tearDown() {
-    nodePeer.join();
-    nodeClient->join();
-    server->stop();
-    server = nullptr;
-    nodeProxy->join();
-}
-
-void
-DhtProxyTester::testGetPut() {
-    bool done = false;
-    std::condition_variable cv;
-    std::mutex cv_m;
-
-    auto key = dht::InfoHash::get("GLaDOs");
-    dht::Value val {"Hey! It's been a long time. How have you been?"};
-    auto val_data = val.data;
-
-    nodePeer.put(key, std::move(val), [&](bool) {
-        done = true;
-        cv.notify_all();
-    });
-    std::unique_lock<std::mutex> lk(cv_m);
-    cv.wait_for(lk, std::chrono::seconds(10), [&]{ return done; });
-
-    auto vals = nodeClient->get(key).get();
-    CPPUNIT_ASSERT(not vals.empty());
-    CPPUNIT_ASSERT(vals.front()->data == val_data);
-}
-
-
-void
-DhtProxyTester::testListen() {
-    bool done = false;
-    std::condition_variable cv;
-    std::mutex cv_m;
-    std::unique_lock<std::mutex> lk(cv_m);
-    auto key = dht::InfoHash::get("GLaDOs");
-
-    // If a peer send a value, the listen operation from the client
-    // should retrieve this value
-    dht::Value firstVal {"Hey! It's been a long time. How have you been?"};
-    auto firstVal_data = firstVal.data;
-    nodePeer.put(key, std::move(firstVal), [&](bool) {
-        done = true;
-        cv.notify_all();
-    });
-    cv.wait_for(lk, std::chrono::seconds(10), [&]{ return done; });
-    done = false;
-
-    auto values = std::vector<dht::Blob>();
-    nodeClient->listen(key, [&](const std::vector<std::shared_ptr<dht::Value>>& v, bool) {
-        for (const auto& value : v)
-            values.emplace_back(value->data);
-        done = true;
-        cv.notify_all();
-        return true;
-    });
-
-    cv.wait_for(lk, std::chrono::seconds(10), [&]{ return done; });
-    done = false;
-    // Here values should contains 2 values
-    CPPUNIT_ASSERT_EQUAL(static_cast<int>(values.size()), 1);
-    CPPUNIT_ASSERT(values.front() == firstVal_data);
-
-    // And the listen should retrieve futures values
-    // All values
-    dht::Value secondVal {"You're a monster"};
-    auto secondVal_data = secondVal.data;
-    nodePeer.put(key, std::move(secondVal));
-    cv.wait_for(lk, std::chrono::seconds(10), [&]{ return done; });
-    // Here values should contains 3 values
-    CPPUNIT_ASSERT_EQUAL(static_cast<int>(values.size()), 2);
-    CPPUNIT_ASSERT(values.back() == secondVal_data);
-}
-
-}  // namespace test
@@ -1,64 +0,0 @@
-/*
- *  Copyright (C) 2018 Savoir-faire Linux Inc.
- *
- *  Author: Sébastien Blin <sebastien.blin@savoirfairelinux.com>
- *
- *  This program is free software; you can redistribute it and/or modify
- *  it under the terms of the GNU General Public License as published by
- *  the Free Software Foundation; either version 3 of the License, or
- *  (at your option) any later version.
- *
- *  This program is distributed in the hope that it will be useful,
- *  but WITHOUT ANY WARRANTY; without even the implied warranty of
- *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- *  GNU General Public License for more details.
- *
- *  You should have received a copy of the GNU General Public License
- *  along with this program. If not, see <https://www.gnu.org/licenses/>.
- */
-
-#pragma once
-
-// cppunit
-#include <cppunit/TestFixture.h>
-#include <cppunit/extensions/HelperMacros.h>
-
-#include <opendht/dhtrunner.h>
-#include <opendht/dht_proxy_server.h>
-
-namespace test {
-
-class DhtProxyTester : public CppUnit::TestFixture {
-    CPPUNIT_TEST_SUITE(DhtProxyTester);
-    CPPUNIT_TEST(testGetPut);
-    CPPUNIT_TEST(testListen);
-    CPPUNIT_TEST_SUITE_END();
-
- public:
-    /**
-     * Method automatically called before each test by CppUnit
-     * Init nodes
-     */
-    void setUp();
-    /**
-     * Method automatically called after each test CppUnit
-     */
-    void tearDown();
-    /**
-     * Test get and put methods
-     */
-    void testGetPut();
-    /**
-     * Test listen
-     */
-    void testListen();
-
- private:
-    dht::DhtRunner nodePeer {};
-
-    std::shared_ptr<dht::DhtRunner> nodeClient;
-    std::shared_ptr<dht::DhtRunner> nodeProxy;
-    std::unique_ptr<dht::DhtProxyServer> server;
-};
-
-}  // namespace test
@@ -144,7 +144,7 @@ void cmd_loop(std::shared_ptr<DhtRunner>& dht, dht_params& params
 #if OPENDHT_PROXY_SERVER
            for (const auto& proxy : proxies) {
                std::cout << "Stats for proxy on port " << proxy.first << std::endl;
-                std::cout << "  " << proxy.second->stats().toString() << std::endl;
+                std::cout << "  " << proxy.second->getStats().toString() << std::endl;
            }
 #endif
            continue;
@@ -199,7 +199,6 @@ void cmd_loop(std::shared_ptr<DhtRunner>& dht, dht_params& params
            continue;
        } else if (op == "cc") {
            dht->connectivityChanged();
-            continue;
        }
 #if OPENDHT_PROXY_SERVER
        else if (op == "pst") {
@@ -497,7 +496,7 @@ main(int argc, char **argv)
            crt = dht::crypto::generateIdentity("DHT Node", ca_tmp);
        }

-        dht::DhtRunner::Config config {};
+        dht::DhtRunner::Config config;
        config.dht_config.node_config.network = params.network;
        config.dht_config.node_config.maintain_storage = false;
        config.dht_config.id = crt;