wip

2026-11-09 04:49:14 +00:00 · 2018-04-21 17:16:37 -04:00
29 changed files with 410 additions and 911 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.2)
+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)
@@ -1,7 +1,7 @@
 dnl define macros
 m4_define([opendht_major_version], 1)
 m4_define([opendht_minor_version], 7)
-m4_define([opendht_patch_version], 2)
+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 {

@@ -52,45 +52,14 @@ struct OPENDHT_PUBLIC NodeStats {
    unsigned table_depth {0};
    unsigned getKnownNodes() const { return good_nodes + dubious_nodes; }
    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)
-};
-
-struct OPENDHT_PUBLIC MessageStats {
-    unsigned packets {0};
-    unsigned ping    {0};
-    unsigned find    {0};
-    unsigned get     {0};
-    unsigned put     {0};
-    unsigned listen  {0};
-    unsigned refresh {0};
+#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>
@@ -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,13 +21,12 @@

 #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>
@@ -238,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
@@ -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
@@ -457,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};
@@ -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;
@@ -27,7 +27,6 @@
 #include "utils.h"
 #include "rng.h"
 #include "rate_limiter.h"
-#include "callbacks.h"

 #include <vector>
 #include <string>
@@ -409,10 +408,6 @@ public:
        return stats;
    }

-    MessageStats getMessageStats(bool in) {
-        return in ? in_stats : out_stats;
-    }
-
    void blacklistNode(const Sp<Node>& n);

    std::vector<Sp<Node>> getCachedNodes(const InfoHash& id, sa_family_t sa_f, size_t count) {
@@ -462,6 +457,16 @@ private:
     */
    void sendRequest(const Sp<Request>& request);

+    struct MessageStats {
+        unsigned ping    {0};
+        unsigned find    {0};
+        unsigned get     {0};
+        unsigned put     {0};
+        unsigned listen  {0};
+        unsigned refresh {0};
+    };
+
+
    // basic wrapper for socket sendto function
    int send(const char *buf, size_t len, int flags, const SockAddr& addr);

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

@@ -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 \
@@ -68,7 +68,7 @@ NodeStats::toString() const
    return ss.str();
 }

-#ifdef OPENDHT_JSONCPP
+#if OPENDHT_PROXY_SERVER || OPENDHT_PROXY_CLIENT
 /**
 * Build a json object from a NodeStats
 */
@@ -98,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

 }
@@ -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)) {
@@ -411,11 +410,10 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr) {
            { /* 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;

+                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);
@@ -518,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()) {
@@ -541,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;
-        }
    }
 }

@@ -1543,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 << "] ";
@@ -2364,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;

@@ -124,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();
@@ -133,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())
-            std::cout << getStats().toString() << std::endl;
-        // 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()
@@ -164,7 +158,7 @@ DhtProxyServer::stop()
        auto listener = currentListeners_.begin();
        while (listener != currentListeners_.end()) {
            listener->session->close();
-            ++listener;
+            ++ listener;
        }
    }
    stopListeners = true;
@@ -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";
@@ -372,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";
@@ -613,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();
@@ -622,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;
@@ -641,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;
@@ -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,25 +396,12 @@ DhtRunner::loop_()
        received = std::move(rcv);
    }
    if (not received.empty()) {
-        size_t packet_num = received.size();
-        auto start = clock::now();
-        while (not received.empty()) {
-            auto& pck = received.front();
-            auto now = clock::now();
-            auto delay = now - pck.received;
-            if (delay > std::chrono::seconds(1))
-                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);
-                auto end = clock::now();
-                std::cerr << "periodic took " << print_dt(end - now)*1000 << std::endl;
-            }
-            received.pop();
-        }
-        if (packet_num > 1) {
-            auto time_all = print_dt(clock::now() - start);
-            std::cerr << "Treating " << packet_num << " packets took " << time_all << " - " << (packet_num/time_all) << " packets/sec" << std::endl;
+        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);
    }
@@ -534,11 +507,7 @@ DhtRunner::startNetwork(const SockAddr sin4, const SockAddr sin6)
                    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;
-                                continue;
-                            }
-                            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();
                    }
@@ -630,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();
 }

@@ -656,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();
 }
@@ -887,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();
        });
    }
@@ -925,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)
 {
@@ -1022,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);
@@ -403,7 +400,6 @@ NetworkEngine::isNodeBlacklisted(const SockAddr& addr) const
 void
 NetworkEngine::processMessage(const uint8_t *buf, size_t buflen, const SockAddr& from)
 {
-    in_stats.packets++;
    if (isMartian(from)) {
        DHT_LOG.w("Received packet from martian node %s", from.toString().c_str());
        return;
@@ -662,10 +658,8 @@ 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
-
-    out_stats.packets++;
    if (sendto(s, buf, len, flags, addr.get(), addr.getLength()) == -1) {
        int err = errno;
        DHT_LOG.e("Can't send message to %s: %s", addr.toString().c_str(), strerror(err));
@@ -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;
    }
@@ -331,7 +331,7 @@ 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;
    }

    /**
@@ -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;
 }

@@ -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
@@ -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,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/>.
- */
-
-#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::tearDown() {
-
-}
-}  // namespace test
@@ -1,48 +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_SUITE_END();
-
- public:
-    /**
-     * Method automatically called before each test by CppUnit
-     */
-    void setUp();
-    /**
-     * Method automatically called after each test CppUnit
-     */
-    void tearDown();
-    /**
-     * Test the differents behaviors of constructors
-     */
-    void testSignatureEncryption();
-};
-
-}  // namespace test
@@ -1,129 +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), [&](bool) {
-        done = true;
-        cv.notify_all();
-    });
-    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
@@ -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;