update to version 1.7.4

dhtproxyclient: fix ok when a permanent PUT operations fails

.travis.yml

@@ -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='';
+      options='-DOPENDHT_SANITIZE=On ';
       if [[ "$OPENDHT_TEST_JOB" == *"opendht.proxyserver"* ]] || [[ "$OPENDHT_TEST_JOB" == *"opendht.push"* ]]; then
         options+='-DOPENDHT_PROXY_SERVER=ON ';
       else

CMakeLists.txt

@@ -1,7 +1,7 @@
 cmake_minimum_required (VERSION 3.1)
 project (opendht)
 set (opendht_VERSION_MAJOR 1)
-set (opendht_VERSION_MINOR 7.1)
+set (opendht_VERSION_MINOR 7.4)
 set (opendht_VERSION ${opendht_VERSION_MAJOR}.${opendht_VERSION_MINOR})
 set (PACKAGE_VERSION ${opendht_VERSION})
 set (VERSION "${opendht_VERSION}")
@@ -47,9 +47,21 @@ 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)
-    pkg_search_module(Jsoncpp jsoncpp)
+    if (!Jsoncpp_FOUND)
+        message(SEND_ERROR "Jsoncpp is required for DHT proxy support")
+    endif()
 endif()
 
 # Build flags
@@ -182,11 +194,6 @@ 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 ()
@@ -204,16 +211,17 @@ 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")
@@ -304,13 +312,24 @@ install (FILES ${CMAKE_CURRENT_BINARY_DIR}/opendhtConfigVersion.cmake DESTINATIO
 IF(OPENDHT_TESTS)
   FIND_PACKAGE(Cppunit REQUIRED)
   # unit testing
-    add_executable(opendht_unit_tests
-      tests/tests_runner.cpp
+    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}
+    )
     if (OPENDHT_SHARED)
       TARGET_LINK_LIBRARIES(opendht_unit_tests opendht)
     else ()

MSVC/opendht.vcxproj

@@ -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_NATIVE;WIN32_LEAN_AND_MEAN;_MBCS;%(PreprocessorDefinitions)</PreprocessorDefinitions>
+      <PreprocessorDefinitions>_CRT_SECURE_NO_WARNINGS;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_NATIVE;WIN32_LEAN_AND_MEAN;_MBCS;%(PreprocessorDefinitions)</PreprocessorDefinitions>
+      <PreprocessorDefinitions>_CRT_SECURE_NO_WARNINGS;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_NATIVE;WIN32_LEAN_AND_MEAN;_MBCS;%(PreprocessorDefinitions)</PreprocessorDefinitions>
+      <PreprocessorDefinitions>_CRT_SECURE_NO_WARNINGS;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_NATIVE;WIN32_LEAN_AND_MEAN;_MBCS;%(PreprocessorDefinitions)</PreprocessorDefinitions>
+      <PreprocessorDefinitions>_CRT_SECURE_NO_WARNINGS;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>

configure.ac

@@ -1,7 +1,7 @@
 dnl define macros
 m4_define([opendht_major_version], 1)
 m4_define([opendht_minor_version], 7)
-m4_define([opendht_patch_version], 1)
+m4_define([opendht_patch_version], 4)
 m4_define([opendht_version],
 		  [opendht_major_version.opendht_minor_version.opendht_patch_version])
 
@@ -101,9 +101,51 @@ 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
@@ -135,58 +177,21 @@ AM_COND_IF([ENABLE_TOOLS], [
   ])
 ])
 
-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_SERVER,
+    [CPPFLAGS+=" -DOPENDHT_PROXY_SERVER=true"],
+    [CPPFLAGS+=" -DOPENDHT_PROXY_SERVER=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_PROXY_CLIENT,
+    [CPPFLAGS+=" -DOPENDHT_PROXY_CLIENT=true"],
+    [CPPFLAGS+=" -DOPENDHT_PROXY_CLIENT=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"])
 
-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])
-])
+AM_COND_IF(ENABLE_PROXY_SERVER_IDENTITY,
+    [CPPFLAGS+=" -DOPENDHT_PROXY_SERVER_IDENTITY=true"],
+    [CPPFLAGS+=" -DOPENDHT_PROXY_SERVER_IDENTITY=false"])
 
 AC_CONFIG_FILES([doc/Doxyfile doc/Makefile])
 
@@ -202,6 +207,6 @@ AM_COND_IF([USE_CYTHON], [
 AC_CONFIG_FILES([Makefile
                  src/Makefile
                  tools/Makefile
-								 tests/Makefile
+                 tests/Makefile
                  opendht.pc])
 AC_OUTPUT

include/opendht/callbacks.h

@@ -27,9 +27,9 @@
 #include <memory>
 #include <functional>
 
-#if OPENDHT_PROXY_SERVER
+#ifdef OPENDHT_JSONCPP
 #include <json/json.h>
-#endif //OPENDHT_PROXY_SERVER
+#endif
 
 namespace dht {
 
@@ -52,14 +52,35 @@ struct OPENDHT_PUBLIC NodeStats {
     unsigned table_depth {0};
     unsigned getKnownNodes() const { return good_nodes + dubious_nodes; }
     std::string toString() const;
-#if OPENDHT_PROXY_SERVER || OPENDHT_PROXY_CLIENT
+
+#ifdef OPENDHT_JSONCPP
     /**
      * Build a json object from a NodeStats
      */
     Json::Value toJson() const;
     NodeStats() {};
     explicit NodeStats(const Json::Value& v);
-#endif //OPENDHT_PROXY_SERVER
+#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)
 };
 
 /**

include/opendht/default_types.h

@@ -114,6 +114,8 @@ 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;
@@ -123,10 +125,11 @@ 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)
+    MSGPACK_DEFINE_MAP(id, msg, date, status, datatype)
 };
 
 class OPENDHT_PUBLIC TrustRequest : public EncryptedValue<TrustRequest>

include/opendht/dht_proxy_client.h

@@ -269,6 +269,7 @@ 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);
@@ -278,6 +279,10 @@ 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);
 
     /**
@@ -296,7 +301,7 @@ private:
     std::string serverHost_;
     std::string pushClientId_;
 
-    std::atomic_flag ongoingStatusUpdate_ = ATOMIC_FLAG_INIT;
+    mutable std::mutex lockCurrentProxyInfos_;
     NodeStatus statusIpv4_ {NodeStatus::Disconnected};
     NodeStatus statusIpv6_ {NodeStatus::Disconnected};
     NodeStats stats4_ {};
@@ -336,8 +341,9 @@ private:
     std::vector<std::function<void()>> callbacks_;
     std::mutex lockCallbacks;
 
+    Sp<InfoState> infoState_;
     std::thread statusThread_;
-    mutable std::mutex lockCurrentProxyInfos_;
+    mutable std::mutex statusLock_;
 
     Scheduler scheduler;
     /**

include/opendht/dht_proxy_server.h

@@ -21,12 +21,13 @@
 
 #pragma once
 
+#include "callbacks.h"
 #include "def.h"
-#include "sockaddr.h"
 #include "infohash.h"
-#include "scheduler.h"
-#include "value.h"
 #include "proxy.h"
+#include "scheduler.h"
+#include "sockaddr.h"
+#include "value.h"
 
 #include <thread>
 #include <memory>
@@ -237,6 +238,8 @@ 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

include/opendht/dhtrunner.h

@@ -293,6 +293,7 @@ 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;
@@ -380,15 +381,7 @@ public:
      */
     void join();
 
-    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
-    }
+    void setProxyServer(const std::string& proxy, const std::string& pushNodeId = "");
 
     /**
      * Start or stop the proxy
@@ -464,7 +457,13 @@ private:
 
     std::thread rcv_thread {};
     std::mutex sock_mtx {};
-    std::vector<std::pair<Blob, SockAddr>> rcv {};
+
+    struct ReceivedPacket {
+        Blob data;
+        SockAddr from;
+        time_point received;
+    };
+    std::queue<ReceivedPacket> rcv {};
 
     /** true if currently actively boostraping */
     std::atomic_bool bootstraping {false};

include/opendht/log_enable.h

@@ -43,8 +43,23 @@ 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>
-    LogMethod(T&& t) : func(std::forward<T>(t)) {}
+    explicit LogMethod(T&& t) : func(std::forward<T>(t)) {}
+
+    template<typename T>
+    LogMethod(const T& t) : func(t) {}
 
     void operator()(char const* format, ...) const {
         va_list args;

include/opendht/proxy.h

@@ -22,7 +22,7 @@
 namespace dht {
 namespace proxy {
 
-constexpr const std::chrono::seconds OP_TIMEOUT {1 * 60 * 60}; // onw hour
+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 char* const HTTP_PROTO {"http://"};
 using ListenToken = uint64_t;

include/opendht/rng.h

@@ -25,7 +25,7 @@
 namespace dht {
 namespace crypto {
 
-#ifndef WIN32_NATIVE
+#ifndef _MSC_VER
 #ifdef _WIN32
 
 /**

include/opendht/securedht.h

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

include/opendht/value.h

@@ -37,9 +37,9 @@
 #include <chrono>
 #include <set>
 
-#if OPENDHT_PROXY_SERVER || OPENDHT_PROXY_CLIENT
+#ifdef OPENDHT_JSONCPP
 #include <json/json.h>
-#endif //OPENDHT_PROXY_SERVER
+#endif
 
 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) {}
 
-#if OPENDHT_PROXY_SERVER || OPENDHT_PROXY_CLIENT
+#ifdef OPENDHT_JSONCPP
     /**
      * Build a value from a json object
      * @param json
      */
     Value(Json::Value& json);
-#endif //OPENDHT_PROXY_SERVER
+#endif
 
     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();
     }
 
-#if OPENDHT_PROXY_SERVER || OPENDHT_PROXY_CLIENT
+#ifdef OPENDHT_JSONCPP
     /**
      * Build a json object from a value
      * Example:
@@ -461,7 +461,7 @@ struct OPENDHT_PUBLIC Value
      * }
      */
     Json::Value toJson() const;
-#endif //OPENDHT_PROXY_SERVER
+#endif
 
     /** 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;
 }
 
-#if OPENDHT_PROXY_SERVER || OPENDHT_PROXY_CLIENT
+#ifdef OPENDHT_JSONCPP
 uint64_t unpackId(const Json::Value& json, const std::string& key);
 #endif
 

src/Makefile.am

@@ -1,7 +1,7 @@
 lib_LTLIBRARIES = libopendht.la
 
-libopendht_la_CPPFLAGS = @CPPFLAGS@ -I$(top_srcdir)/include/opendht @Argon2_CFLAGS@
-libopendht_la_LIBADD   = @Argon2_LIBS@ @GnuTLS_LIBS@ @Nettle_LIBS@
+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_LDFLAGS  = @LDFLAGS@ @Argon2_LDFLAGS@
 libopendht_la_SOURCES  = \
         dht.cpp \

src/callbacks.cpp

@@ -68,7 +68,7 @@ NodeStats::toString() const
     return ss.str();
 }
 
-#if OPENDHT_PROXY_SERVER || OPENDHT_PROXY_CLIENT
+#ifdef OPENDHT_JSONCPP
 /**
  * Build a json object from a NodeStats
  */
@@ -98,6 +98,32 @@ NodeStats::NodeStats(const Json::Value& val)
     if (val.isMember("table_depth"))
         table_depth = static_cast<unsigned>(val["table_depth"].asLargestUInt());
 }
-#endif //OPENDHT_PROXY_SERVER
+
+/**
+ * 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
+
 
 }

src/dht.cpp

@@ -319,11 +319,12 @@ 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 = 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)
+        auto& query = sr->callbacks.empty() ? ANY_QUERY : cb->second.query;
+        const time_point up = (not sr->callbacks.empty() and update)
+                                ? sr->getLastGetTime(*query)
                                 : time_point::min();
 
         if (pn and pn->canGet(now, up, query)) {
@@ -410,10 +411,11 @@ 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);
@@ -516,7 +518,7 @@ Dht::searchSynchedNodeListen(const Sp<Search>& sr, SearchNode& n)
             });
         }
         auto prev_req = r != n.listenStatus.end() ? r->second.req : nullptr;
-        r->second.req = network_engine.sendListen(n.node, sr->id, *query, n.token, prev_req,
+        auto new_req = network_engine.sendListen(n.node, sr->id, *query, n.token, prev_req,
             [this,ws,query](const net::Request& req, net::RequestAnswer&& answer) mutable
             { /* on done */
                 if (auto sr = ws.lock()) {
@@ -539,12 +541,18 @@ 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;
+        }
     }
 }
 
@@ -1535,10 +1543,7 @@ Dht::dumpSearch(const Search& sr, std::ostream& out) const
                     if (ack == n.acked.end() or not ack->second.first) {
                         out << ' ';
                     } else {
-                        if (ack->second.first->completed())
-                            out << 'a';
-                        else if (ack->second.first->pending())
-                            out << 'f';
+                        out << ack->second.first->getStateChar();
                     }
                 }
                 out << "] ";
@@ -1580,11 +1585,6 @@ 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)
@@ -1593,6 +1593,12 @@ 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();
 }
 
@@ -2359,9 +2365,6 @@ 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);
 }
 

src/dht_proxy_client.cpp

@@ -33,12 +33,18 @@
 
 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;
@@ -46,7 +52,7 @@ struct DhtProxyClient::Listener
     Sp<restbed::Request> req;
     std::thread thread;
     unsigned callbackId;
-    Sp<State> state;
+    Sp<ListenState> state;
     Sp<proxy::ListenToken> pushNotifToken;
     Sp<Scheduler::Job> refreshJob;
     Listener(ValueCache&& c, Sp<Scheduler::Job>&& j, const Sp<restbed::Request>& r, Value::Filter&& f)
@@ -60,6 +66,7 @@ struct DhtProxyClient::Listener
 struct PermanentPut {
     Sp<Value> value;
     Sp<Scheduler::Job> refreshJob;
+    Sp<std::atomic_bool> ok;
 };
 
 struct DhtProxyClient::ProxySearch {
@@ -99,6 +106,8 @@ DhtProxyClient::~DhtProxyClient()
     isDestroying_ = true;
     cancelAllOperations();
     cancelAllListeners();
+    if (infoState_)
+        infoState_->cancel = true;
     if (statusThread_.joinable())
         statusThread_.join();
 }
@@ -322,24 +331,29 @@ 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;
+        auto nextRefresh = scheduler.time() + proxy::OP_TIMEOUT - proxy::OP_MARGIN;
+        auto ok = std::make_shared<std::atomic_bool>(false);
         search->second.puts.erase(id);
-        search->second.puts.emplace(id, PermanentPut {val, scheduler.add(nextRefresh, [this, key, id]{
+        search->second.puts.emplace(id, PermanentPut {val, scheduler.add(nextRefresh, [this, key, id, ok]{
+            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, {}, time_point::max(), true);
-                    scheduler.edit(p->second.refreshJob, scheduler.time() + proxy::OP_TIMEOUT);
+                    doPut(key, p->second.value,
+                    [ok](bool result, const std::vector<std::shared_ptr<dht::Node> >&){
+                        *ok = result;
+                    }, time_point::max(), true);
+                    scheduler.edit(p->second.refreshJob, scheduler.time() + proxy::OP_TIMEOUT - proxy::OP_MARGIN);
                 }
             }
-        })});
+        }), 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());
@@ -471,9 +485,12 @@ void
 DhtProxyClient::getProxyInfos()
 {
     DHT_LOG.d("Requesting proxy server node information");
+    std::lock_guard<std::mutex> l(statusLock_);
 
-    if (ongoingStatusUpdate_.test_and_set())
-        return;
+    auto infoState = std::make_shared<InfoState>();
+    if (infoState_)
+        infoState_->cancel = true;
+    infoState_ = infoState;
 
     {
         std::lock_guard<std::mutex> l(lockCurrentProxyInfos_);
@@ -488,11 +505,15 @@ DhtProxyClient::getProxyInfos()
 
     // Try to contact the proxy and set the status to connected when done.
     // will change the connectivity status
-    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>();
+    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]{
         try {
             auto hostAndService = splitPort(serverHost);
             auto resolved_proxies = SockAddr::resolve(hostAndService.first, hostAndService.second);
@@ -507,8 +528,10 @@ 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, successCount](
+                    [this, resolved_proxy, infoState](
                                 const std::shared_ptr<restbed::Request>&,
                                 const std::shared_ptr<restbed::Response>& reply)
                 {
@@ -516,6 +539,8 @@ 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);
 
@@ -524,13 +549,14 @@ DhtProxyClient::getProxyInfos()
                         auto reader = std::unique_ptr<Json::CharReader>(rbuilder.newCharReader());
                         try {
                             reader->parse(body.data(), body.data() + body.size(), &proxyInfos, &err);
-                        } catch (const std::exception& e) {
-                            DHT_LOG.w("Error parsing proxy infos: %s", e.what());
+                        } catch (...) {
+                            return;
                         }
                         auto family = resolved_proxy.getFamily();
-                        if      (family == AF_INET)  successCount->ipv4++;
-                        else if (family == AF_INET6) successCount->ipv6++;
-                        onProxyInfos(proxyInfos, family);
+                        if      (family == AF_INET)  state.ipv4++;
+                        else if (family == AF_INET6) state.ipv6++;
+                        if (not state.cancel)
+                            onProxyInfos(proxyInfos, family);
                     }
                 }));
             }
@@ -540,22 +566,24 @@ DhtProxyClient::getProxyInfos()
         } catch (const std::exception& e) {
             DHT_LOG.e("Error sending proxy info request: %s", e.what());
         }
-        if (successCount->ipv4 == 0) onProxyInfos(Json::Value{}, AF_INET);
-        if (successCount->ipv6 == 0) onProxyInfos(Json::Value{}, AF_INET6);
-        ongoingStatusUpdate_.clear();
+        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);
     });
 }
 
 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"]);
@@ -619,7 +647,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;
@@ -652,14 +680,104 @@ 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()) {
@@ -668,6 +786,7 @@ 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()) {
@@ -687,9 +806,12 @@ 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<Listener::State>();
+    auto state = std::make_shared<ListenState>();
     l->second.state = state;
     l->second.cb = [this,key,token,state](const std::vector<Sp<Value>>& values, bool expired) {
         if (state->cancel)
@@ -715,82 +837,34 @@ 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.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.
-            }
-#if OPENDHT_PUSH_NOTIFICATIONS
-            else
-                fillBodyToGetToken(req);
-#endif
+    l->second.req = req;
 
-            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;
+    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);
                 }
-            }, settings).get();
-            auto& s = *state;
-            if (not s.ok and not s.cancel) {
-                opFailed();
             }
-        }
-    );
+        });
+        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);
+        });
+    }
     return token;
 }
 
@@ -874,81 +948,53 @@ 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, 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();
+        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);
+            });
+        }
     }
 }
 
@@ -957,6 +1003,12 @@ 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;
+    }
     try {
         std::lock_guard<std::mutex> lock(searchLock_);
         auto timeout = notification.find("timeout");
@@ -967,7 +1019,7 @@ DhtProxyClient::pushNotificationReceived(const std::map<std::string, std::string
             if (vidIt != notification.end()) {
                 // Refresh put
                 auto vid = std::stoull(vidIt->second);
-                auto put = search.puts.at(vid);
+                auto& put = search.puts.at(vid);
                 scheduler.edit(put.refreshJob, scheduler.time());
                 loopSignal_();
             } else {
@@ -1002,6 +1054,7 @@ 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());
@@ -1016,37 +1069,9 @@ 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]() {
-        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();
+        sendSubscribe(req, pushNotifToken, state);
     });
 #endif
 }

src/dht_proxy_server.cpp

@@ -29,6 +29,7 @@
 #include <chrono>
 #include <functional>
 #include <limits>
+#include <iostream>
 
 using namespace std::placeholders;
 
@@ -123,12 +124,6 @@ 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();
@@ -138,8 +133,19 @@ 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()
@@ -158,7 +164,7 @@ DhtProxyServer::stop()
         auto listener = currentListeners_.begin();
         while (listener != currentListeners_.end()) {
             listener->session->close();
-            ++ listener;
+            ++listener;
         }
     }
     stopListeners = true;
@@ -196,17 +202,22 @@ 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,
-        [=](const std::shared_ptr<restbed::Session> s, const restbed::Bytes& /*b*/)
+        [this](const std::shared_ptr<restbed::Session> s, const restbed::Bytes& /*b*/) mutable
         {
             try {
                 if (dht_) {
                     Json::Value result;
-                    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();
+                    {
+                        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();
+                    }
                     result["public_ip"] = s->get_origin(); // [ipv6:ipv4]:port or ipv4:port
                     Json::StreamWriterBuilder wbuilder;
                     wbuilder["commentStyle"] = "None";
@@ -361,7 +372,10 @@ DhtProxyServer::subscribe(const std::shared_ptr<restbed::Session>& session)
                     return;
                 }
                 auto pushToken = root["key"].asString();
-                if (pushToken.empty()) return;
+                if (pushToken.empty()) {
+                    s->close(restbed::BAD_REQUEST, "{\"err\":\"No token\"}");
+                    return;
+                }
                 auto tokenFromReq = unpackId(root, "token");
                 auto platform = root["platform"].asString();
                 auto isAndroid = platform == "android";
@@ -588,8 +602,8 @@ DhtProxyServer::put(const std::shared_ptr<restbed::Session>& session)
                         if (reader->parse(char_data, char_data + b.size(), &root, &err)) {
                             // Build the Value from json
                             auto value = std::make_shared<Value>(root);
-                            auto permanent = root.isMember("permanent");
-                            std::cout << "Got put " << infoHash << " " << value << " " << (permanent ? "permanent" : "") << std::endl;
+                            bool permanent = root.isMember("permanent");
+                            std::cout << "Got put " << infoHash << " " << *value << " " << (permanent ? "permanent" : "") << std::endl;
 
                             if (permanent) {
                                 std::string pushToken, clientId, platform;
@@ -606,28 +620,30 @@ DhtProxyServer::put(const std::shared_ptr<restbed::Session>& session)
                                 auto vid = value->id;
                                 auto sPuts = puts_.emplace(infoHash, SearchPuts{}).first;
                                 auto r = sPuts->second.puts.emplace(vid, PermanentPut{});
+                                auto& pput = r.first->second;
                                 if (r.second) {
-                                    r.first->second.expireJob = scheduler_.add(timeout, [this, infoHash, vid]{
-                                        std::cout << "Permanent put expired: " << infoHash << 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())
-                                        r.first->second.expireNotifyJob = scheduler_.add(timeout - proxy::OP_MARGIN,
+                                    if (not pushToken.empty()) {
+                                        pput.expireNotifyJob = scheduler_.add(timeout - proxy::OP_MARGIN,
                                             [this, infoHash, vid, pushToken, clientId, isAndroid]
                                         {
-                                            std::cout << "Permanent put refresh: " << infoHash << std::endl;
+                                            std::cout << "Permanent put refresh: " << infoHash << " " << vid << std::endl;
                                             Json::Value json;
                                             json["timeout"] = infoHash.toString();
                                             json["to"] = clientId;
                                             json["vid"] = std::to_string(vid);
                                             sendPushNotification(pushToken, json, isAndroid);
                                         });
+                                    }
 #endif
                                 } else {
-                                    scheduler_.edit(r.first->second.expireJob, timeout);
-                                    if (r.first->second.expireNotifyJob)
-                                        scheduler_.edit(r.first->second.expireNotifyJob, timeout - proxy::OP_MARGIN);
+                                    scheduler_.edit(pput.expireJob, timeout);
+                                    if (pput.expireNotifyJob)
+                                        scheduler_.edit(pput.expireNotifyJob, timeout - proxy::OP_MARGIN);
                                 }
                                 lock.unlock();
                                 schedulerCv_.notify_one();
@@ -652,7 +668,8 @@ DhtProxyServer::put(const std::shared_ptr<restbed::Session>& session)
                 } else {
                     s->close(restbed::SERVICE_UNAVAILABLE, "{\"err\":\"Incorrect DhtRunner\"}");
                 }
-            } catch (...) {
+            } catch (const std::exception& e) {
+                std::cout << "Error performing put: " << e.what() << std::endl;
                 s->close(restbed::INTERNAL_SERVER_ERROR, "{\"err\":\"Internal server error\"}");
             }
         }

src/dhtrunner.cpp

@@ -42,6 +42,7 @@
 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;
@@ -247,10 +248,10 @@ void
 DhtRunner::setLoggers(LogMethod error, LogMethod warn, LogMethod debug) {
     std::lock_guard<std::mutex> lck(dht_mtx);
     if (dht_)
-        dht_->setLoggers(std::forward<LogMethod>(error), std::forward<LogMethod>(warn), std::forward<LogMethod>(debug));
+        dht_->setLoggers(error, warn, debug);
 #if OPENDHT_PROXY_CLIENT
     if (dht_via_proxy_)
-        dht_via_proxy_->setLoggers(std::forward<LogMethod>(error), std::forward<LogMethod>(warn), std::forward<LogMethod>(debug));
+        dht_via_proxy_->setLoggers(error, warn, debug);
 #endif
 }
 
@@ -301,6 +302,17 @@ 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
 {
@@ -342,7 +354,9 @@ std::vector<SockAddr>
 DhtRunner::getPublicAddress(sa_family_t af)
 {
     std::lock_guard<std::mutex> lck(dht_mtx);
-    return activeDht()->getPublicAddress(af);
+    if (auto dht = activeDht())
+        return dht->getPublicAddress(af);
+    return {};
 }
 std::vector<std::string>
 DhtRunner::getPublicAddressStr(sa_family_t af)
@@ -396,12 +410,15 @@ DhtRunner::loop_()
         received = std::move(rcv);
     }
     if (not received.empty()) {
-        for (const auto& pck : received) {
-            auto& buf = pck.first;
-            auto& from = pck.second;
-            wakeup = dht->periodic(buf.data(), buf.size()-1, from);
+        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();
         }
-        received.clear();
     } else {
         wakeup = dht->periodic(nullptr, 0, nullptr, 0);
     }
@@ -507,7 +524,11 @@ DhtRunner::startNetwork(const SockAddr sin4, const SockAddr sin6)
                     if (rc > 0) {
                         {
                             std::lock_guard<std::mutex> lck(sock_mtx);
-                            rcv.emplace_back(Blob {buf.begin(), buf.begin()+rc+1}, SockAddr(from, from_len));
+                            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()});
                         }
                         cv.notify_all();
                     }
@@ -562,6 +583,7 @@ 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;
@@ -569,21 +591,20 @@ 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();
@@ -599,20 +620,24 @@ 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);
-    pending_ops.emplace([=](SecureDht& dht) {
+    {
+        std::lock_guard<std::mutex> lck(storage_mtx);
 #if OPENDHT_PROXY_CLIENT
-        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);
+        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);
+        });
 #else
-        dht.cancelListen(h, token);
+        pending_ops.emplace([=](SecureDht& dht) {
+            dht.cancelListen(h, token);
+        });
 #endif // OPENDHT_PROXY_CLIENT
-    });
+    }
     cv.notify_all();
 }
 
@@ -621,20 +646,21 @@ 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
-            auto it = listeners_.find(token);
+        pending_ops.emplace([=](SecureDht&) {
+            auto it = listeners_.find(ftoken.get());
             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
-            dht.cancelListen(h, token);
-#endif // OPENDHT_PROXY_CLIENT
         });
+#else
+        pending_ops.emplace([=](SecureDht& dht) {
+            dht.cancelListen(h, ftoken.get());
+        });
+#endif // OPENDHT_PROXY_CLIENT
     }
     cv.notify_all();
 }
@@ -851,7 +877,7 @@ DhtRunner::connectivityChanged()
 {
     {
         std::lock_guard<std::mutex> lck(storage_mtx);
-        pending_ops.emplace([=](SecureDht& dht) {
+        pending_ops_prio.emplace([=](SecureDht& dht) {
             dht.connectivityChanged();
         });
     }
@@ -872,8 +898,8 @@ DhtRunner::findCertificate(InfoHash hash, std::function<void(const std::shared_p
 void
 DhtRunner::resetDht()
 {
-    listeners_.clear();
 #if OPENDHT_PROXY_CLIENT
+    listeners_.clear();
     dht_via_proxy_.reset();
 #endif // OPENDHT_PROXY_CLIENT
     dht_.reset();
@@ -889,6 +915,21 @@ 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)
 {
@@ -971,7 +1012,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& dht) {
+        pending_ops_prio.emplace([=](SecureDht&) {
             if (dht_via_proxy_)
                 dht_via_proxy_->pushNotificationReceived(data);
         });

src/network_engine.cpp

@@ -262,7 +262,13 @@ 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)
@@ -280,7 +286,7 @@ NetworkEngine::requestStep(Sp<Request> sreq)
     auto now = scheduler.time();
     auto& node = *req.node;
     if (req.isExpired(now)) {
-        DHT_LOG.e(node.id, "[node %s] expired !", node.toString().c_str());
+        DHT_LOG.d(node.id, "[node %s] expired !", node.toString().c_str());
         node.setExpired();
         if (not node.id)
             requests.erase(req.tid);
@@ -292,24 +298,21 @@ 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 == EPIPE        ||
+        err == EPERM)
     {
-        scheduler.add(now, [this,wreq]() mutable {
-            if (auto req = wreq.lock()) {
-                req->node->setExpired();
-                if (not req->node->id)
-                    requests.erase(req->tid);
-            }
-        });
+        node.setExpired();
+        if (not 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);
@@ -658,7 +661,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;

src/request.h

@@ -57,6 +57,15 @@ 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,

src/search.h

@@ -136,7 +136,7 @@ struct Dht::SearchNode {
      *
      * @return true if we can send get, else false.
      */
-    bool canGet(time_point now, time_point update, Sp<Query> q = {}) const {
+    bool canGet(time_point now, time_point update, const 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() and not pending_sq_status)
+                if (s.second->pending())
                     pending_sq_status = true;
-                if (s.second->completed() and not (update > s.second->reply_time) and not completed_sq_status)
+                else if (s.second->completed() and not (update > s.second->reply_time))
                     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 (hasStartedPagination(q) or completed_sq_status or pending_sq_status);
+                not (completed_sq_status or pending_sq_status or hasStartedPagination(q));
     }
 
     /**
@@ -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())
+        if (listen_status == listenStatus.end() or not listen_status->second.req)
             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.second - REANNOUNCE_MARGIN;
+                : (ack->second.first->completed() ? ack->second.second - REANNOUNCE_MARGIN : time_point::min());
     }
 
     /**
@@ -430,7 +430,8 @@ struct Dht::Search {
      *
      * @param query  The query identifying a 'get' request.
      */
-    time_point getLastGetTime(Sp<Query> query = {}) const;
+    time_point getLastGetTime(const Query&) const;
+    time_point getLastGetTime() const;
 
     /**
      * Is this get operation done ?
@@ -746,11 +747,20 @@ Dht::Search::isSynced(time_point now) const
 }
 
 time_point
-Dht::Search::getLastGetTime(Sp<Query> q) const
+Dht::Search::getLastGetTime(const Query& q) const
 {
     time_point last = time_point::min();
     for (const auto& g : callbacks)
-        last = std::max(last, (not q or q->isSatisfiedBy(*g.second.query) ? g.second.start : time_point::min()));
+        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);
     return last;
 }
 

src/value.cpp

@@ -22,9 +22,9 @@
 #include "default_types.h"
 #include "securedht.h" // print certificate ID
 
-#if OPENDHT_PROXY_SERVER || OPENDHT_PROXY_CLIENT
+#ifdef OPENDHT_JSONCPP
 #include "base64.h"
-#endif //OPENDHT_PROXY_SERVER
+#endif
 
 
 namespace dht {
@@ -172,7 +172,7 @@ Value::msgpack_unpack_body(const msgpack::object& o)
     }
 }
 
-#if OPENDHT_PROXY_SERVER || OPENDHT_PROXY_CLIENT
+#ifdef OPENDHT_JSONCPP
 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 //OPENDHT_PROXY_SERVER
+#endif
 
 bool
 FieldValue::operator==(const FieldValue& vfd) const

tests/Makefile.am

@@ -3,7 +3,7 @@ bin_PROGRAMS = opendht_unit_tests
 
 AM_CPPFLAGS = -I../include
 
-nobase_include_HEADERS = infohashtester.h dhtrunnertester.h
-opendht_unit_tests_SOURCES = tests_runner.cpp infohashtester.cpp dhtrunnertester.cpp
+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
 opendht_unit_tests_LDFLAGS = -lopendht -lcppunit -L@top_builddir@/src/.libs @GnuTLS_LIBS@
 endif

tests/cryptotester.cpp

@@ -0,0 +1,53 @@
+/*
+ *  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

tests/cryptotester.h

@@ -0,0 +1,48 @@
+/*
+ *  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

tests/dhtproxytester.cpp

@@ -0,0 +1,126 @@
+/*
+ *  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

tests/dhtproxytester.h

@@ -0,0 +1,64 @@
+/*
+ *  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

tools/dhtnode.cpp

@@ -199,6 +199,7 @@ 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") {
@@ -496,7 +497,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;