tools/cnode: show value ID when put completes

Fix calls to DeserializeNodes. They must
catch possible exceptions

Signed-off-by: Lucas Dias <lucas_dias@redes.ufsm.br>

.github/workflows/ccpp.yml

@@ -13,7 +13,7 @@ jobs:
         sudo apt update && \
         sudo apt install libncurses5-dev libreadline-dev nettle-dev libasio-dev \
         libgnutls28-dev libuv1-dev python3-pip python3-dev python3-setuptools libcppunit-dev libjsoncpp-dev \
-        autotools-dev autoconf libfmt-dev libmsgpack-dev libargon2-0-dev
+        autotools-dev autoconf libfmt-dev libmsgpack-dev libargon2-dev
     - name: Cython
       run: sudo pip3 install Cython
     - name: llhttp
@@ -56,7 +56,7 @@ jobs:
         sudo apt update && \
         sudo apt install ninja-build libncurses5-dev libreadline-dev nettle-dev libasio-dev \
         libgnutls28-dev libuv1-dev python3-dev python3-setuptools python3-pip \
-        libcppunit-dev libjsoncpp-dev libfmt-dev libmsgpack-dev libargon2-0-dev
+        libcppunit-dev libjsoncpp-dev libfmt-dev libmsgpack-dev libargon2-dev
     - name: Python dependencies
       run: sudo pip3 install meson Cython
     - name: Configure
@@ -73,7 +73,7 @@ jobs:
       run: |
         sudo apt update && \
         sudo apt install libncurses5-dev libreadline-dev nettle-dev libfmt-dev \
-        libgnutls28-dev libcppunit-dev libmsgpack-dev libargon2-0-dev
+        libgnutls28-dev libcppunit-dev libmsgpack-dev libargon2-dev
     - name: Configure
       run: |
         mkdir build && cd build
@@ -87,7 +87,7 @@ jobs:
 
   build-macos:
     name: macOS/Clang build
-    runs-on: macos-12
+    runs-on: macOS-15
     steps:
     - uses: actions/checkout@v4
     - name: Dependencies
@@ -96,7 +96,7 @@ jobs:
 
     - name: expected-lite
       run: |
-        sudo mkdir /usr/local/include/nonstd && \
+        sudo mkdir -p /usr/local/include/nonstd && \
         sudo wget https://raw.githubusercontent.com/martinmoene/expected-lite/master/include/nonstd/expected.hpp \
           -O /usr/local/include/nonstd/expected.hpp
 

.github/workflows/clang-analyzer.yml

@@ -12,7 +12,7 @@ jobs:
           sudo apt update && \
           sudo apt install libncurses5-dev libreadline-dev nettle-dev \
           libgnutls28-dev libuv1-dev python3-dev python3-setuptools libcppunit-dev libjsoncpp-dev \
-          autotools-dev autoconf libfmt-dev libmsgpack-dev libargon2-0-dev libasio-dev \
+          autotools-dev autoconf libfmt-dev libmsgpack-dev libargon2-dev libasio-dev \
           llvm llvm-dev clang clang-tools && \
           sudo apt remove gcc g++
 

CMakeLists.txt

@@ -33,6 +33,7 @@ option (OPENDHT_PROXY_CLIENT "Enable DHT proxy client, use Restinio and jsoncpp"
 option (OPENDHT_PROXY_OPENSSL "Build DHT proxy with OpenSSL" ON)
 CMAKE_DEPENDENT_OPTION(OPENDHT_HTTP "Build embedded http(s) client" OFF "NOT OPENDHT_PROXY_SERVER;NOT OPENDHT_PROXY_CLIENT" ON)
 option (OPENDHT_PEER_DISCOVERY "Enable multicast peer discovery" ON)
+option (OPENDHT_IO_URING "Use io_uring if available on the system (Linux only)" OFF)
 option (OPENDHT_INDEX "Build DHT indexation feature" OFF)
 option (OPENDHT_TESTS_NETWORK "Enable unit tests that require network access" ON)
 option (OPENDHT_C "Build C bindings" OFF)
@@ -107,7 +108,7 @@ if (NOT MSVC)
             )
             set(llhttp_target llhttp_static)
         else()
-            FetchContent_Declare(llhttp-local URL "https://github.com/nodejs/llhttp/archive/refs/tags/release/v9.2.0.tar.gz")
+            FetchContent_Declare(llhttp-local URL "https://github.com/nodejs/llhttp/archive/refs/tags/release/v9.2.1.tar.gz")
             if (BUILD_SHARED_LIBS)
                 set(BUILD_SHARED_LIBS ON CACHE INTERNAL "")
             else()
@@ -122,7 +123,7 @@ if (NOT MSVC)
             endif()
         endif()
         set(http_lib "-lllhttp")
-        
+
         if (NOT Jsoncpp_FOUND)
             message(SEND_ERROR "Jsoncpp is required for DHT proxy support")
         endif()
@@ -156,14 +157,14 @@ else ()
             ${WIN32_DEP_DIR}/openssl/include
             ${WIN32_DEP_DIR}/restinio/dev
             ${WIN32_DEP_DIR}/fmt/include
-            ${WIN32_DEP_DIR}/http_parser
+            ${WIN32_DEP_DIR}/llhttp/include
         )
     endif ()
 endif ()
 
 if (OPENDHT_HTTP OR OPENDHT_PEER_DISCOVERY)
     add_definitions(-DASIO_STANDALONE)
-    if (UNIX AND NOT APPLE)
+    if (OPENDHT_IO_URING AND UNIX AND NOT APPLE)
         pkg_search_module(liburing IMPORTED_TARGET liburing)
     endif ()
 endif()
@@ -343,12 +344,12 @@ if (MSVC)
             )
             if (OPENDHT_HTTP)
                 add_obj_lib (win32_fmt ${WIN32_DEP_DIR}/fmt/msvc/Release/fmt.lib)
-                add_obj_lib (win32_http_parser ${WIN32_DEP_DIR}/http_parser/x64/Release/http-parser.lib)
+                add_obj_lib (win32_llhttp ${WIN32_DEP_DIR}/llhttp/build/Release/llhttp.lib)
                 add_obj_lib (win32_ssl ${WIN32_DEP_DIR}/openssl/libssl_static.lib)
                 add_obj_lib (win32_crypto ${WIN32_DEP_DIR}/openssl/libcrypto_static.lib)
                 list (APPEND obj_libs
                     $<TARGET_OBJECTS:win32_fmt>
-                    $<TARGET_OBJECTS:win32_http_parser>
+                    $<TARGET_OBJECTS:win32_llhttp>
                     $<TARGET_OBJECTS:win32_ssl>
                     $<TARGET_OBJECTS:win32_crypto>
                 )
@@ -361,7 +362,7 @@ if (MSVC)
             )
             list (APPEND win32_Libs
                 ${PROJECT_SOURCE_DIR}/../fmt/msvc/Release/fmt.lib
-                ${PROJECT_SOURCE_DIR}/../http_parser/x64/Release/http-parser.lib
+                ${PROJECT_SOURCE_DIR}/../llhttp/build/Release/llhttp.lib
                 ${PROJECT_SOURCE_DIR}/../openssl/libssl.lib
                 ${PROJECT_SOURCE_DIR}/../openssl/libcrypto.lib
             )
@@ -407,7 +408,7 @@ else()
     if (OPENDHT_PROXY_OPENSSL)
         target_link_libraries(opendht PUBLIC PkgConfig::OPENSSL)
     endif()
-    if (liburing_FOUND)
+    if (OPENDHT_IO_URING AND liburing_FOUND)
         set(iouring_lib ", liburing")
         target_link_libraries(opendht PUBLIC PkgConfig::liburing)
         target_compile_definitions(opendht PUBLIC ASIO_HAS_IO_URING ASIO_DISABLE_EPOLL)

c/opendht.cpp

@@ -344,6 +344,7 @@ int dht_runner_run_config(dht_runner* r, in_port_t port, const dht_runner_config
         config.push_token = conf->push_token ? std::string(conf->push_token) : std::string{};
         config.push_topic = conf->push_topic ? std::string(conf->push_topic) : std::string{};
         config.push_platform = conf->push_platform ? std::string(conf->push_platform) : std::string{};
+        config.proxy_user_agent = conf->proxy_user_agent ? std::string(conf->proxy_user_agent) : std::string{};
         config.peer_discovery = conf->peer_discovery;
         config.peer_publish = conf->peer_publish;
 

c/opendht_c.h

@@ -148,6 +148,7 @@ struct OPENDHT_PUBLIC dht_runner_config {
     const char* push_token;
     const char* push_topic;
     const char* push_platform;
+    const char* proxy_user_agent;
     bool peer_discovery;
     bool peer_publish;
     dht_certificate* server_ca;

include/opendht/dht_proxy_client.h

@@ -53,19 +53,21 @@ public:
     DhtProxyClient();
 
     explicit DhtProxyClient(
-        std::shared_ptr<crypto::Certificate> serverCA, crypto::Identity clientIdentity,
-        std::function<void()> loopSignal, const std::string& serverHost,
-        const std::string& pushClientId = "", std::shared_ptr<Logger> logger = {});
+        std::shared_ptr<crypto::Certificate> serverCA,
+        crypto::Identity clientIdentity,
+        std::function<void()> loopSignal,
+        const std::string& serverHost,
+        const std::string& userAgent = "",
+        const std::string& pushClientId = "",
+        const std::string& pushToken = "",
+        const std::string& pushTopic = "",
+        const std::string& pushPlatform = "",
+        std::shared_ptr<Logger> logger = {}
+    );
 
     void setHeaderFields(http::Request& request);
 
-    virtual void setPushNotificationToken(const std::string& token) override {
-#ifdef OPENDHT_PUSH_NOTIFICATIONS
-        deviceKey_ = token;
-#else
-        (void) token;
-#endif
-    }
+    virtual void setPushNotificationToken(const std::string& token) override;
 
     virtual void setPushNotificationTopic(const std::string& topic) override {
 #ifdef OPENDHT_PUSH_NOTIFICATIONS
@@ -351,7 +353,7 @@ private:
     std::string proxyUrl_;
     dht::crypto::Identity clientIdentity_;
     std::shared_ptr<dht::crypto::Certificate> serverCertificate_;
-    //std::pair<std::string, std::string> serverHostService_;
+    std::string userAgent_ {"OpenDHT"};
     std::string pushClientId_;
     std::string pushSessionId_;
 

include/opendht/dht_proxy_server.h

@@ -98,6 +98,32 @@ public:
 
     asio::io_context& io_context() const;
 
+    using clock = std::chrono::steady_clock;
+    using time_point = clock::time_point;
+
+    struct PushStats {
+        uint64_t highPriorityCount {0};
+        uint64_t normalPriorityCount {0};
+
+        void increment(bool highPriority) {
+            if (highPriority)
+                highPriorityCount++;
+            else
+                normalPriorityCount++;
+        }
+
+        Json::Value toJson() const {
+            Json::Value val;
+            val["highPriorityCount"] = static_cast<Json::UInt64>(highPriorityCount);
+            val["normalPriorityCount"] = static_cast<Json::UInt64>(normalPriorityCount);
+            return val;
+        }
+
+        std::string toString() const {
+            return fmt::format("{} high priority, {} normal priority", highPriorityCount, normalPriorityCount);
+        }
+    };
+
     struct ServerStats {
         /** Current number of listen operations */
         size_t listenCount {0};
@@ -107,24 +133,39 @@ public:
         size_t totalPermanentPuts {0};
         /** Current number of push tokens with at least one listen operation */
         size_t pushListenersCount {0};
+
+        /** Time at which the server was started */
+        time_point serverStartTime;
+        /** Last time at which the stats were updated */
+        time_point lastUpdated;
+        /** Total number of push notification requests that the server attempted to
+          * send since being started, broken down by type and priority level */
+        PushStats androidPush;
+        PushStats iosPush;
+        PushStats unifiedPush;
+
         /** Average requests per second */
         double requestRate {0};
         /** Node Info **/
         std::shared_ptr<NodeInfo> nodeInfo {};
 
         std::string toString() const {
-            std::ostringstream ss;
-            ss << "Listens: " << listenCount << " Puts: " << putCount << " PushListeners: " << pushListenersCount << std::endl;
-            ss << "Requests: " << requestRate << " per second." << std::endl;
+            auto ret = fmt::format("Listens: {}, Puts: {}, PushListeners: {}\n"
+                        "Push requests in the last {}: [Android: {}], [iOS: {}], [Unified: {}]\n"
+                        "Requests: {} per second.",
+                        listenCount, putCount, pushListenersCount,
+                        print_duration(lastUpdated - serverStartTime),
+                        androidPush.toString(), iosPush.toString(), unifiedPush.toString(),
+                        requestRate);
             if (nodeInfo) {
                 auto& ipv4 = nodeInfo->ipv4;
                 if (ipv4.table_depth > 1)
-                    ss << "IPv4 Network estimation: " << ipv4.getNetworkSizeEstimation() << std::endl;;
+                    ret += fmt::format("IPv4 Network estimation: {}\n", ipv4.getNetworkSizeEstimation());
                 auto& ipv6 = nodeInfo->ipv6;
                 if (ipv6.table_depth > 1)
-                    ss << "IPv6 Network estimation: " << ipv6.getNetworkSizeEstimation() << std::endl;;
+                    ret += fmt::format("IPv6 Network estimation: {}\n", ipv6.getNetworkSizeEstimation());
             }
-            return ss.str();
+            return ret;
         }
 
         /**
@@ -136,6 +177,11 @@ public:
             result["putCount"] = static_cast<Json::UInt64>(putCount);
             result["totalPermanentPuts"] = static_cast<Json::UInt64>(totalPermanentPuts);
             result["pushListenersCount"] = static_cast<Json::UInt64>(pushListenersCount);
+            result["serverStartTime"] = static_cast<Json::LargestInt>(to_time_t(serverStartTime));
+            result["lastUpdated"] = static_cast<Json::LargestInt>(to_time_t(lastUpdated));
+            result["androidPush"] = androidPush.toJson();
+            result["iosPush"] = iosPush.toJson();
+            result["unifiedPush"] = unifiedPush.toJson();
             result["requestRate"] = requestRate;
             if (nodeInfo)
                 result["nodeInfo"] = nodeInfo->toJson();
@@ -366,9 +412,6 @@ private:
     template <typename Is>
     void loadState(Is& is, size_t size);
 
-    using clock = std::chrono::steady_clock;
-    using time_point = clock::time_point;
-
     std::shared_ptr<asio::io_context> ioContext_;
     std::shared_ptr<DhtRunner> dht_;
     Json::StreamWriterBuilder jsonBuilder_;
@@ -393,6 +436,11 @@ private:
     std::shared_ptr<ServerStats> stats_;
     std::shared_ptr<NodeInfo> nodeInfo_ {};
     std::unique_ptr<asio::steady_timer> printStatsTimer_;
+    const time_point serverStartTime_;
+    mutable std::mutex pushStatsMutex_;
+    PushStats androidPush_;
+    PushStats iosPush_;
+    PushStats unifiedPush_;
 
     // Thread-safe access to listeners map.
     std::mutex lockListener_;

include/opendht/dhtrunner.h

@@ -64,6 +64,7 @@ public:
         std::string push_token {};
         std::string push_topic {};
         std::string push_platform {};
+        std::string proxy_user_agent {};
         bool peer_discovery {false};
         bool peer_publish {false};
         std::shared_ptr<dht::crypto::Certificate> server_ca;

include/opendht/http.h

@@ -72,7 +72,7 @@ class OPENDHT_PUBLIC Url
 {
 public:
     Url() = default;
-    Url(const std::string& url);
+    Url(std::string_view url);
     std::string url;
     std::string protocol {"http"};
     std::string host;

include/opendht/indexation/pht.h

@@ -409,7 +409,7 @@ private:
 
         /**
          * This mutlimap contains all prefix insert in the tree in time order
-         * We could then delete the last one if there is too much node
+         * We could then delete the last one if there are too many nodes
          * The tree will self destroy is branch ( thanks to share_ptr )
          */
         std::multimap<time_point, std::shared_ptr<Node>> leaves_;

include/opendht/proxy.h

@@ -24,7 +24,7 @@ namespace proxy {
 
 constexpr const std::chrono::hours OP_TIMEOUT {24}; // one day
 constexpr const std::chrono::hours OP_MARGIN {2}; // two hours
-constexpr const char* const HTTP_PROTO {"http://"};
+
 using ListenToken = uint64_t;
 
 }

include/opendht/utils.h

@@ -21,6 +21,7 @@
 #include "def.h"
 
 #include <msgpack.hpp>
+#include <fmt/format.h>
 
 #include <chrono>
 #include <random>
@@ -60,7 +61,7 @@ void erase_if(std::map<Key, Item>& map, const Condition& condition)
  * Split "[host]:port" or "host:port" to pair<"host", "port">.
  */
 OPENDHT_PUBLIC std::pair<std::string, std::string>
-splitPort(const std::string& s);
+splitPort(std::string_view s);
 
 class OPENDHT_PUBLIC DhtException : public std::runtime_error {
 public:
@@ -84,37 +85,23 @@ using duration = clock::duration;
 time_point from_time_t(std::time_t t);
 std::time_t to_time_t(time_point t);
 
-inline std::string
-to_str(double d) {
-    char buf[16];
-    auto ret = snprintf(buf, sizeof(buf), "%.3g", d);
-    return (ret < 0) ? std::to_string(d) : std::string(buf, ret);
-}
-
-/**
- * Converts std::chrono::duration to floating-point seconds.
- */
-template <class DT>
-static double
-print_dt(DT d) {
-    return std::chrono::duration_cast<std::chrono::duration<double>>(d).count();
-}
-
 template <class DT>
 static std::string
 print_duration(DT d) {
     if (d < std::chrono::seconds(0)) {
         return "-" + print_duration(-d);
     } else if (d < std::chrono::milliseconds(1)) {
-        return to_str(std::chrono::duration_cast<std::chrono::duration<double, std::micro>>(d).count()) +  " us";
+        return fmt::format("{:.3g} us", std::chrono::duration_cast<std::chrono::duration<double, std::micro>>(d).count());
     } else if (d < std::chrono::seconds(1)) {
-        return to_str(std::chrono::duration_cast<std::chrono::duration<double, std::milli>>(d).count()) +  " ms";
+        return fmt::format("{:.3g} ms", std::chrono::duration_cast<std::chrono::duration<double, std::milli>>(d).count());
     } else if (d < std::chrono::minutes(1)) {
-        return to_str(std::chrono::duration_cast<std::chrono::duration<double>>(d).count()) +  " s";
+        return fmt::format("{:.3g} s", std::chrono::duration_cast<std::chrono::duration<double>>(d).count());
     } else if (d < std::chrono::hours(1)) {
-        return to_str(std::chrono::duration_cast<std::chrono::duration<double, std::ratio<60>>>(d).count()) +  " min";
+        return fmt::format("{:.3g} min", std::chrono::duration_cast<std::chrono::duration<double, std::ratio<60>>>(d).count());
+    } else if (d < std::chrono::hours(72)) {
+        return fmt::format("{:.3g} h", std::chrono::duration_cast<std::chrono::duration<double, std::ratio<3600>>>(d).count());
     } else {
-        return to_str(std::chrono::duration_cast<std::chrono::duration<double, std::ratio<3600>>>(d).count()) +  " h";
+        return fmt::format("{:.3g} days", std::chrono::duration_cast<std::chrono::duration<double, std::ratio<86400>>>(d).count());
     }
 }
 

meson.build

@@ -16,6 +16,7 @@ llhttp = dependency('llhttp', 'libllhttp', required: get_option('proxy_client'))
 io_uring = dependency('liburing', required: false)
 
 deps = [fmt, gnutls, nettle, msgpack, argon2, openssl, jsoncpp, llhttp, io_uring]
+conf_data = configuration_data()
 
 add_project_arguments('-DMSGPACK_NO_BOOST', '-DASIO_STANDALONE', language : 'cpp')
 if io_uring.found()
@@ -45,8 +46,6 @@ opendht_src = [
     'src/thread_pool.cpp',
 ]
 
-conf_data = configuration_data()
-
 if get_option('indexation').enabled()
     opendht_src += 'src/indexation/pht.cpp'
     add_project_arguments('-DOPENDHT_INDEXATION', language : 'cpp')

python/opendht.pyx

@@ -500,12 +500,13 @@ cdef class DhtConfig(object):
     def setRateLimit(self, ssize_t max_req_per_sec, ssize_t max_peer_req_per_sec):
         self._config.dht_config.node_config.max_req_per_sec = max_req_per_sec
         self._config.dht_config.node_config.max_peer_req_per_sec = max_peer_req_per_sec
-    def setProxyInfo(self, str proxy_server, str push_node_id="", str push_token="", str push_topic="", str push_platform="", Certificate server_ca = Certificate()):
+    def setProxyInfo(self, str proxy_server, str push_node_id="", str push_token="", str push_topic="", str push_platform="", str user_agent="", Certificate server_ca = Certificate()):
         self._config.proxy_server = proxy_server.encode()
         self._config.push_node_id = push_node_id.encode()
         self._config.push_token = push_token.encode()
         self._config.push_topic = push_topic.encode()
         self._config.push_platform = push_platform.encode()
+        self._config.proxy_user_agent = user_agent.encode()
         self._config.server_ca = server_ca._cert
     def setPeerDiscovery(self, bool peer_discovery, bool peer_publish):
         self._config.peer_discovery = peer_discovery

python/opendht_cpp.pxd

@@ -246,6 +246,7 @@ cdef extern from "opendht/dhtrunner.h" namespace "dht":
             string push_token
             string push_topic
             string push_platform
+            string proxy_user_agent
             bool peer_discovery
             bool peer_publish
             shared_ptr[Certificate] server_ca

rust/src/dhtrunner.rs

@@ -49,6 +49,7 @@ impl DhtRunnerConfig {
             push_token: ptr::null(),
             push_topic: ptr::null(),
             push_platform: ptr::null(),
+            proxy_user_agent: ptr::null(),
             peer_discovery: false,
             peer_publish: false,
             server_ca: ptr::null_mut(),
@@ -84,6 +85,10 @@ impl DhtRunnerConfig {
         self.push_platform = CString::new(push_platform).unwrap().into_raw();
     }
 
+    pub fn set_proxy_user_agent(&mut self, proxy_user_agent: &str) {
+        self.proxy_user_agent = CString::new(proxy_user_agent).unwrap().into_raw();
+    }
+
     pub fn set_identity(&mut self, certificate: Box<DhtCertificate>, privatekey: Box<PrivateKey>) {
         self.dht_config.id.privatekey = Box::into_raw(privatekey);
         self.dht_config.id.certificate = Box::into_raw(certificate);

rust/src/ffi.rs

@@ -120,6 +120,7 @@ pub struct DhtRunnerConfig
     pub push_token: *const c_char,
     pub push_topic: *const c_char,
     pub push_platform: *const c_char,
+    pub proxy_user_agent: *const c_char,
     pub peer_discovery: bool,
     pub peer_publish: bool,
     pub server_ca: *mut DhtCertificate,

src/crypto.cpp

@@ -1164,11 +1164,17 @@ saveIdentity(const Identity& id, const std::string& path, const std::string& pri
         auto ca_key_data = id.first->serialize(privkey_password);
         std::ofstream key_file(path + ".pem");
         key_file.write((char*)ca_key_data.data(), ca_key_data.size());
+        // Throw error if the file is not written
+        if (!key_file)
+            throw CryptoException("Could not write private key file");
     }
     {
         auto ca_key_data = id.second->getPacked();
         std::ofstream crt_file(path + ".crt");
         crt_file.write((char*)ca_key_data.data(), ca_key_data.size());
+        // Throw error if the file is not written
+        if (!crt_file)
+            throw CryptoException("Could not write certificate file");
     }
 }
 
@@ -1276,12 +1282,14 @@ Certificate::generate(const PrivateKey& key, const std::string& name, const Iden
         if (not ca.second->isCA()) {
             throw CryptoException("Signing certificate must be CA");
         }
-        if (err = gnutls_x509_crt_privkey_sign(cert, ca.second->cert, ca.first->key, pk.getPreferredDigest(), 0)) {
+        err = gnutls_x509_crt_privkey_sign(cert, ca.second->cert, ca.first->key, pk.getPreferredDigest(), 0);
+        if (err) {
             throw CryptoException(std::string("Error when signing certificate ") + gnutls_strerror(err));
         }
         ret.issuer = ca.second;
     } else {
-        if (err = gnutls_x509_crt_privkey_sign(cert, cert, key.key, pk.getPreferredDigest(), 0)) {
+        err = gnutls_x509_crt_privkey_sign(cert, cert, key.key, pk.getPreferredDigest(), 0);
+        if (err) {
             throw CryptoException(std::string("Error when signing certificate ") + gnutls_strerror(err));
         }
     }

src/dht.cpp

@@ -93,7 +93,7 @@ Dht::shutdown(ShutdownCallback cb, bool stop)
                     r.second.done_cb(false, {});
                 sr.second->callbacks.clear();
                 for (const auto& a : sr.second->announce) {
-                    if (a.callback) a.callback(false, {});
+                    for (auto& cb : a.callbacks) cb(false, {});
                 }
                 sr.second->announce.clear();
                 sr.second->listeners.clear();
@@ -113,7 +113,7 @@ Dht::shutdown(ShutdownCallback cb, bool stop)
     auto str_donecb = [=](bool, const std::vector<Sp<Node>>&) {
         --*remaining;
         if (logger_)
-            logger_->w("shuting down node: %u ops remaining", *remaining);
+            logger_->w("Shutting down node: %u ops remaining", *remaining);
         if (!*remaining && cb) { cb(); }
     };
 
@@ -121,7 +121,7 @@ Dht::shutdown(ShutdownCallback cb, bool stop)
         *remaining += maintainStorage(str, true, str_donecb);
 
     if (logger_)
-        logger_->w("shuting down node: after storage, %u ops", *remaining);
+        logger_->w("Shutting down node: after storage, %u ops", *remaining);
 
     if (!*remaining) {
         if (cb) cb();
@@ -147,7 +147,7 @@ Dht::sendCachedPing(Bucket& b)
 {
     if (b.cached)
         if (logger_)
-            logger_->d(b.cached->id, "[node %s] sending ping to cached node", b.cached->toString().c_str());
+            logger_->d(b.cached->id, "[node %s] Sending ping to cached node", b.cached->toString().c_str());
     b.sendCachedPing(network_engine);
 }
 
@@ -280,7 +280,7 @@ Dht::expireSearches()
         auto b = sr.callbacks.empty() && sr.announce.empty() && sr.listeners.empty() && sr.step_time < t;
         if (b) {
             if (logger_)
-                logger_->d(srp.first, "[search %s] removing search", srp.first.toString().c_str());
+                logger_->d(srp.first, "[search %s] Removing search", srp.first.toString().c_str());
             sr.clear();
             return b;
         } else { return false; }
@@ -360,7 +360,7 @@ void Dht::paginate(std::weak_ptr<Search> ws, Sp<Query> query, SearchNode* n) {
                 auto query_for_vid = std::make_shared<Query>(Select {}, Where {}.id(vid));
                 sn->pagination_queries[query].push_back(query_for_vid);
                 if (logger_)
-                    logger_->d(id, sn->node->id, "[search %s] [node %s] sending %s",
+                    logger_->d(id, sn->node->id, "[search %s] [node %s] Sending %s",
                         id.toString().c_str(), sn->node->toString().c_str(), query_for_vid->toString().c_str());
                 sn->getStatus[query_for_vid] = network_engine.sendGetValues(status.node,
                         id,
@@ -371,8 +371,8 @@ void Dht::paginate(std::weak_ptr<Search> ws, Sp<Query> query, SearchNode* n) {
                         );
             } catch (const std::out_of_range&) {
                 if (logger_)
-                    logger_->e(id, sn->node->id, "[search %s] [node %s] received non-id field in response to "\
-                        "'SELECT id' request...",
+                    logger_->e(id, sn->node->id, "[search %s] [node %s] Received non-id field in response to "\
+                        "'SELECT id' request…",
                         id.toString().c_str(), sn->node->toString().c_str());
             }
         }
@@ -381,7 +381,7 @@ void Dht::paginate(std::weak_ptr<Search> ws, Sp<Query> query, SearchNode* n) {
     n->pagination_queries[query].push_back(select_q);
 
     if (logger_)
-        logger_->d(sr->id, n->node->id, "[search %s] [node %s] sending %s",
+        logger_->d(sr->id, n->node->id, "[search %s] [node %s] Sending %s",
             sr->id.toString().c_str(), n->node->toString().c_str(), select_q->toString().c_str());
     n->getStatus[select_q] = network_engine.sendGetValues(n->node,
             sr->id,
@@ -426,7 +426,7 @@ Dht::searchSendGetValues(Sp<Search> sr, SearchNode* pn, bool update)
                 return nullptr;
 
             /* if (logger_)
-                   logger_->d(sr->id, n->node->id, "[search %s] [node %s] sending 'find_node'",
+                   logger_->d(sr->id, n->node->id, "[search %s] [node %s] Sending 'find_node'",
                         sr->id.toString().c_str(), n->node->toString().c_str());*/
             n->getStatus[query] = network_engine.sendFindNode(n->node,
                     sr->id,
@@ -439,9 +439,9 @@ Dht::searchSendGetValues(Sp<Search> sr, SearchNode* pn, bool update)
                 continue;
 
             if (query and not query->select.empty()) {
-                /* The request contains a select. No need to paginate... */
+                /* The request contains a select. No need to paginate… */
                 /* if (logger_)
-                       logger_->d(sr->id, n->node->id, "[search %s] [node %s] sending 'get'",
+                       logger_->d(sr->id, n->node->id, "[search %s] [node %s] Sending 'get'",
                             sr->id.toString().c_str(), n->node->toString().c_str());*/
                 n->getStatus[query] = network_engine.sendGetValues(n->node,
                         sr->id,
@@ -453,7 +453,7 @@ Dht::searchSendGetValues(Sp<Search> sr, SearchNode* pn, bool update)
                 paginate(ws, query, n);
         }
 
-        /* We only try to send one request. return. */
+        /* We only attempt to send one request. Return. */
         return n;
 
     } while (++cb != sr->callbacks.end());
@@ -519,7 +519,7 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr, unsigned syncLevel) {
             /* only put the value if the node doesn't already have it */
             if (not hasValue or seq_no < a.value->seq) {
                 if (logger_)
-                    logger_->d(sr->id, sn->node->id, "[search %s] [node %s] sending 'put' (vid: %016" PRIx64 ")",
+                    logger_->d(sr->id, sn->node->id, "[search %s] [node %s] Sending 'put' (vid: %016" PRIx64 ")",
                         sr->id.toString().c_str(), sn->node->toString().c_str(), a.value->id);
                 auto created = a.permanent ? time_point::max() : a.created;
                 acked = {
@@ -528,7 +528,7 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr, unsigned syncLevel) {
                 };
             } else if (hasValue and a.permanent) {
                 if (logger_)
-                    logger_->w(sr->id, sn->node->id, "[search %s] [node %s] sending 'refresh' (vid: %016" PRIx64 ")",
+                    logger_->w(sr->id, sn->node->id, "[search %s] [node %s] Sending 'refresh' (vid: %016" PRIx64 ")",
                         sr->id.toString().c_str(), sn->node->toString().c_str(), a.value->id);
                 acked = {
                     network_engine.sendRefreshValue(sn->node, sr->id, a.value->id, sn->token, onDone,
@@ -540,7 +540,7 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr, unsigned syncLevel) {
                     ](const net::Request& /*req*/, net::DhtProtocolException&& e){
                         if (e.getCode() == net::DhtProtocolException::NOT_FOUND) {
                             if (logger_)
-                                logger_->e(node->id, "[node %s] returned error 404: storage not found", node->toString().c_str());
+                                logger_->e(node->id, "[node %s] Returned error 404: storage not found", node->toString().c_str());
                             if (auto sr = ws.lock()) {
                                 if (auto sn = sr->getNode(node)) {
                                     sn->acked[v->id] = {
@@ -558,7 +558,7 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr, unsigned syncLevel) {
                 };
             } else {
                 if (logger_)
-                    logger_->w(sr->id, sn->node->id, "[search %s] [node %s] already has value (vid: %016" PRIx64 "). Aborting.",
+                    logger_->w(sr->id, sn->node->id, "[search %s] [node %s] Already has value (vid: %016" PRIx64 "). Aborting.",
                         sr->id.toString().c_str(), sn->node->toString().c_str(), a.value->id);
                 auto ack_req = std::make_shared<net::Request>(net::Request::State::COMPLETED);
                 ack_req->reply_time = now;
@@ -594,7 +594,7 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr, unsigned syncLevel) {
                     sendQuery = true;
                 } else {
                     if (logger_)
-                        logger_->w(sr->id, n.node->id, "[search %s] [node %s] sending 'put' (vid: %016" PRIx64 ")",
+                        logger_->w(sr->id, n.node->id, "[search %s] [node %s] Sending 'put' (vid: %016" PRIx64 ")",
                             sr->id.toString().c_str(), n.node->toString().c_str(), a.value->id);
                     n.acked[a.value->id] = {
                         network_engine.sendAnnounceValue(n.node, sr->id, a.value, a.created, n.token, onDone, onExpired),
@@ -607,7 +607,7 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr, unsigned syncLevel) {
         if (sendQuery) {
             n.probe_query = PROBE_QUERY;
             if (logger_)
-                logger_->d(sr->id, n.node->id, "[search %s] [node %s] sending %s",
+                logger_->d(sr->id, n.node->id, "[search %s] [node %s] Sending %s",
                     sr->id.toString().c_str(), n.node->toString().c_str(), n.probe_query->toString().c_str());
             auto req = network_engine.sendGetValues(n.node,
                     sr->id,
@@ -634,7 +634,7 @@ Dht::searchSynchedNodeListen(const Sp<Search>& sr, SearchNode& n)
         if (n.getListenTime(r, listenExp) > scheduler.time())
             continue;
         // if (logger_)
-        //     logger_->d(sr->id, n.node->id, "[search %s] [node %s] sending 'listen'",
+        //     logger_->d(sr->id, n.node->id, "[search %s] [node %s] Sending 'listen'",
         //        sr->id.toString().c_str(), n.node->toString().c_str());
 
         if (r == n.listenStatus.end()) {
@@ -706,7 +706,7 @@ Dht::searchStep(std::weak_ptr<Search> ws)
     auto syncLevel = preSynced ? level - MARGIN : 0;
     /*if (auto req_count = sr->currentlySolicitedNodeCount())
         if (logger_)
-            logger_->d(sr->id, "[search %s IPv%c] step (%d requests)",
+            logger_->d(sr->id, "[search %s IPv%c] Step (%d requests)",
                 sr->id.toString().c_str(), sr->af == AF_INET ? '4' : '6', req_count);*/
     sr->step_time = now;
 
@@ -787,7 +787,7 @@ Dht::searchStep(std::weak_ptr<Search> ws)
     if (sr->getNumberOfConsecutiveBadNodes() >= std::min<size_t>(sr->nodes.size(), SEARCH_MAX_BAD_NODES))
     {
         if (logger_)
-            logger_->w(sr->id, "[search %s IPv%c] expired", sr->id.toString().c_str(), sr->af == AF_INET ? '4' : '6');
+            logger_->w(sr->id, "[search %s IPv%c] Expired", sr->id.toString().c_str(), sr->af == AF_INET ? '4' : '6');
         sr->expire();
         if (not public_stable)
             connectivityChanged(sr->af);
@@ -804,7 +804,7 @@ unsigned Dht::refill(Dht::Search& sr) {
 
     if (cached_nodes.empty()) {
         if (logger_)
-            logger_->e(sr.id, "[search %s IPv%c] no nodes from cache while refilling search",
+            logger_->e(sr.id, "[search %s IPv%c] No nodes from cache while refilling search",
                 sr.id.toString().c_str(), (sr.af == AF_INET) ? '4' : '6');
         return 0;
     }
@@ -816,7 +816,7 @@ unsigned Dht::refill(Dht::Search& sr) {
             ++inserted;
     }
     if (logger_)
-        logger_->d(sr.id, "[search %s IPv%c] refilled search with %u nodes from node cache",
+        logger_->d(sr.id, "[search %s IPv%c] Refilled search with %u nodes from node cache",
             sr.id.toString().c_str(), (sr.af == AF_INET) ? '4' : '6', inserted);
     return inserted;
 }
@@ -828,7 +828,7 @@ Dht::search(const InfoHash& id, sa_family_t af, GetCallback gcb, QueryCallback q
 {
     if (!isRunning(af)) {
         if (logger_)
-            logger_->e(id, "[search %s IPv%c] unsupported protocol", id.toString().c_str(), (af == AF_INET) ? '4' : '6');
+            logger_->e(id, "[search %s IPv%c] Unsupported protocol", id.toString().c_str(), (af == AF_INET) ? '4' : '6');
         if (dcb)
             dcb(false, {});
         return {};
@@ -856,7 +856,7 @@ Dht::search(const InfoHash& id, sa_family_t af, GetCallback gcb, QueryCallback q
             }
             if (not sr) {
                 if (logger_)
-                    logger_->e(id, "[search %s IPv%c] maximum number of searches reached !", id.toString().c_str(), (af == AF_INET) ? '4' : '6');
+                    logger_->e(id, "[search %s IPv%c] Maximum number of searches reached!", id.toString().c_str(), (af == AF_INET) ? '4' : '6');
                 return {};
             }
         }
@@ -870,7 +870,7 @@ Dht::search(const InfoHash& id, sa_family_t af, GetCallback gcb, QueryCallback q
         sr->nodes.reserve(SEARCH_NODES+1);
         sr->nextSearchStep = scheduler.add(time_point::max(), std::bind(&Dht::searchStep, this, std::weak_ptr<Search>(sr)));
         if (logger_)
-            logger_->w(id, "[search %s IPv%c] new search", id.toString().c_str(), (af == AF_INET) ? '4' : '6');
+            logger_->w(id, "[search %s IPv%c] New search", id.toString().c_str(), (af == AF_INET) ? '4' : '6');
         if (search_id == 0)
             search_id++;
     }
@@ -904,16 +904,16 @@ Dht::listenTo(const InfoHash& id, sa_family_t af, ValueCallback cb, Value::Filte
 {
     if (!isRunning(af))
         return 0;
-       // logger__ERR("[search %s IPv%c] search_time is now in %lfs", sr->id.toString().c_str(), (sr->af == AF_INET) ? '4' : '6', print_dt(tm-clock::now()));
+       // logger__ERR("[search %s IPv%c] Search_time is now in %lfs", sr->id.toString().c_str(), (sr->af == AF_INET) ? '4' : '6', print_duration(tm-clock::now()));
 
     //logger__WARN("listenTo %s", id.toString().c_str());
     auto& srs = searches(af);
     auto srp = srs.find(id);
     Sp<Search> sr = (srp == srs.end()) ? search(id, af) : srp->second;
     if (!sr)
-        throw DhtException("Can't create search");
+        throw DhtException("Unable to create search");
     if (logger_)
-        logger_->w(id, "[search %s IPv%c] listen", id.to_c_str(), (af == AF_INET) ? '4' : '6');
+        logger_->w(id, "[search %s IPv%c] Listen", id.to_c_str(), (af == AF_INET) ? '4' : '6');
     return sr->listen(cb, std::move(f), q, scheduler);
 }
 
@@ -1022,7 +1022,7 @@ Dht::put(const InfoHash& id, Sp<Value> val, DoneCallback callback, time_point cr
     storageStore(id, val, created, {}, permanent);
 
     if (logger_)
-        logger_->d(id, "put: adding %s -> %s", id.toString().c_str(), val->toString().c_str());
+        logger_->d(id, "put: adding %s → %s", id.toString().c_str(), val->toString().c_str());
 
     auto op = std::make_shared<OpStatus>();
     auto donecb = [callback](const std::vector<Sp<Node>>& nodes, OpStatus& op) {
@@ -1265,7 +1265,7 @@ Dht::storageChanged(const InfoHash& id, Storage& st, const Sp<Value>& v, bool ne
                     vals.push_back(v);
                 if (not vals.empty()) {
                     if (logger_)
-                        logger_->d(id, "[store %s] sending update local listener with token %lu",
+                        logger_->d(id, "[store %s] Sending update local listener with token %lu",
                             id.toString().c_str(),
                             l.first);
                     cbs.emplace_back(l.second.get_cb, std::move(vals));
@@ -1286,7 +1286,7 @@ Dht::storageChanged(const InfoHash& id, Storage& st, const Sp<Value>& v, bool ne
                 if (f and not f(*v))
                     continue;
                 if (logger_)
-                    logger_->w(id, node_listeners.first->id, "[store %s] [node %s] sending update",
+                    logger_->w(id, node_listeners.first->id, "[store %s] [node %s] Sending update",
                         id.toString().c_str(),
                         node_listeners.first->toString().c_str());
                 std::vector<Sp<Value>> vals {};
@@ -1392,7 +1392,7 @@ void
 Dht::storageRemoved(const InfoHash& id, Storage& st, const std::vector<Sp<Value>>& values, size_t totalSize)
 {
     if (logger_)
-        logger_->d(id, "[store %s] discarded %ld values (%ld bytes)",
+        logger_->d(id, "[store %s] Discarded %ld values (%ld bytes)",
         id.toString().c_str(), values.size(), totalSize);
 
     total_store_size -= totalSize;
@@ -1410,7 +1410,7 @@ Dht::storageRemoved(const InfoHash& id, Storage& st, const std::vector<Sp<Value>
         for (const auto& node_listeners : st.listeners) {
             for (const auto& l : node_listeners.second) {
                 if (logger_)
-                    logger_->w(id, node_listeners.first->id, "[store %s] [node %s] sending expired",
+                    logger_->w(id, node_listeners.first->id, "[store %s] [node %s] Sending expired",
                         id.toString().c_str(),
                         node_listeners.first->toString().c_str());
                 Blob ntoken = makeToken(node_listeners.first->getAddr(), false);
@@ -1432,7 +1432,7 @@ Dht::expireStore()
 
         if (i->second.empty() && i->second.listeners.empty() && i->second.local_listeners.empty()) {
             if (logger_)
-                logger_->d(i->first, "[store %s] discarding empty storage", i->first.toString().c_str());
+                logger_->d(i->first, "[store %s] Discarding empty storage", i->first.toString().c_str());
             i = store.erase(i);
         }
         else
@@ -1900,7 +1900,7 @@ Dht::neighbourhoodMaintenance(RoutingTable& list)
     auto n = q->randomNode(rd);
     if (n) {
         if (logger_)
-            logger_->d(id, n->id, "[node %s] sending [find %s] for neighborhood maintenance",
+            logger_->d(id, n->id, "[node %s] Sending [find %s] for neighborhood maintenance",
                 n->toString().c_str(), id.toString().c_str());
         /* Since our node-id is the same in both DHTs, it's probably
            profitable to query both families. */
@@ -1954,14 +1954,14 @@ Dht::bucketMaintenance(RoutingTable& list)
                 }
 
                 if (logger_)
-                    logger_->d(id, n->id, "[node %s] sending find %s for bucket maintenance", n->toString().c_str(), id.toString().c_str());
+                    logger_->d(id, n->id, "[node %s] Sending find %s for bucket maintenance", n->toString().c_str(), id.toString().c_str());
                 //auto start = scheduler.time();
                 network_engine.sendFindNode(n, id, want, nullptr, [this,n](const net::Request&, bool over) {
                     if (over) {
                         const auto& end = scheduler.time();
                         // using namespace std::chrono;
                         // if (logger_)
-                        //     logger_->d(n->id, "[node %s] bucket maintenance op expired after %s", n->toString().c_str(), print_duration(end-start).c_str());
+                        //     logger_->d(n->id, "[node %s] Bucket maintenance op expired after %s", n->toString().c_str(), print_duration(end-start).c_str());
                         scheduler.edit(nextNodesConfirmation, end + Node::MAX_RESPONSE_TIME);
                     }
                 });
@@ -1979,7 +1979,7 @@ Dht::dataPersistence(InfoHash id)
     auto str = store.find(id);
     if (str != store.end() and now > str->second.maintenance_time) {
         if (logger_)
-            logger_->d(id, "[storage %s] maintenance (%u values, %u bytes)",
+            logger_->d(id, "[storage %s] Maintenance (%u values, %u bytes)",
                 id.toString().c_str(), str->second.valueCount(), str->second.totalSize());
         maintainStorage(*str);
         str->second.maintenance_time = now + MAX_STORAGE_MAINTENANCE_EXPIRE_TIME;
@@ -2033,7 +2033,7 @@ Dht::periodic(const uint8_t *buf, size_t buflen, SockAddr from, const time_point
             network_engine.processMessage(buf, buflen, std::move(from));
         } catch (const std::exception& e) {
             if (logger_)
-                logger_->w("Can't process message: %s", e.what());
+                logger_->w("Unable to process message: %s", e.what());
         }
     }
     return scheduler.run();
@@ -2087,7 +2087,7 @@ Dht::bootstrap()
             }
         } catch (const std::exception& e) {
             if (logger_)
-                logger_->e(myid, "Can't resolve %s:%s: %s", boootstrap.first.c_str(), boootstrap.second.c_str(), e.what());
+                logger_->e(myid, "Unable to resolve %s:%s: %s", boootstrap.first.c_str(), boootstrap.second.c_str(), e.what());
         }
     }
     scheduler.cancel(bootstrapJob);
@@ -2118,12 +2118,12 @@ Dht::confirmNodes()
 
     if (dht4.searches.empty() and dht4.status == NodeStatus::Connected) {
         if (logger_)
-            logger_->d(myid, "[confirm nodes] initial IPv4 'get' for my id (%s)", myid.toString().c_str());
+            logger_->d(myid, "[confirm nodes] Initial IPv4 'get' for my id (%s)", myid.toString().c_str());
         search(myid, AF_INET);
     }
     if (dht6.searches.empty() and dht6.status == NodeStatus::Connected) {
         if (logger_)
-            logger_->d(myid, "[confirm nodes] initial IPv6 'get' for my id (%s)", myid.toString().c_str());
+            logger_->d(myid, "[confirm nodes] Initial IPv6 'get' for my id (%s)", myid.toString().c_str());
         search(myid, AF_INET6);
     }
 
@@ -2281,7 +2281,7 @@ Dht::onError(Sp<net::Request> req, net::DhtProtocolException e) {
     const auto& node = req->node;
     if (e.getCode() == net::DhtProtocolException::UNAUTHORIZED) {
         if (logger_)
-            logger_->e(node->id, "[node %s] token flush", node->toString().c_str());
+            logger_->e(node->id, "[node %s] Token flush", node->toString().c_str());
         node->authError();
         for (auto& srp : searches(node->getFamily())) {
             auto& sr = srp.second;
@@ -2296,7 +2296,7 @@ Dht::onError(Sp<net::Request> req, net::DhtProtocolException e) {
         }
     } else if (e.getCode() == net::DhtProtocolException::NOT_FOUND) {
         if (logger_)
-            logger_->e(node->id, "[node %s] returned error 404: storage not found", node->toString().c_str());
+            logger_->e(node->id, "[node %s] Returned error 404: storage not found", node->toString().c_str());
         node->cancelRequest(req);
     }
 }
@@ -2347,7 +2347,7 @@ Dht::onGetValues(Sp<Node> node, const InfoHash& hash, want_t, const Query& query
     if (st != store.end() && not st->second.empty()) {
         answer.values = st->second.get(query.where.getFilter());
         if (logger_)
-            logger_->d(hash, "[node %s] sending %u values", node->toString().c_str(), answer.values.size());
+            logger_->d(hash, "[node %s] Sending %u values", node->toString().c_str(), answer.values.size());
     }
     return answer;
 }
@@ -2359,18 +2359,18 @@ void Dht::onGetValuesDone(const Sp<Node>& node,
 {
     if (not sr) {
         if (logger_)
-            logger_->w("[search unknown] got reply to 'get'. Ignoring.");
+            logger_->w("[search unknown] Got reply to 'get'. Ignoring.");
         return;
     }
 
     /* if (logger_)
-           logger_->d(sr->id, "[search %s] [node %s] got reply to 'get' with %u nodes",
+           logger_->d(sr->id, "[search %s] [node %s] Got reply to 'get' with %u nodes",
             sr->id.toString().c_str(), node->toString().c_str(), a.nodes4.size()+a.nodes6.size());*/
 
     if (not a.ntoken.empty()) {
         if (not a.values.empty() or not a.fields.empty()) {
             if (logger_)
-                logger_->d(sr->id, node->id, "[search %s] [node %s] found %u values",
+                logger_->d(sr->id, node->id, "[search %s] [node %s] Found %u values",
                       sr->id.toString().c_str(), node->toString().c_str(), a.values.size());
             for (auto& getp : sr->callbacks) { /* call all callbacks for this search */
                 auto& get = getp.second;
@@ -2419,7 +2419,7 @@ void Dht::onGetValuesDone(const Sp<Node>& node,
         }
     } else {
         if (logger_)
-            logger_->w(sr->id, "[node %s] no token provided. Ignoring response content.", node->toString().c_str());
+            logger_->w(sr->id, "[node %s] No token provided. Ignoring response content.", node->toString().c_str());
         network_engine.blacklistNode(node);
     }
 
@@ -2436,7 +2436,7 @@ Dht::onListen(Sp<Node> node, const InfoHash& hash, const Blob& token, size_t soc
 {
     if (not hash) {
         if (logger_)
-            logger_->w(node->id, "[node %s] listen with no info_hash", node->toString().c_str());
+            logger_->w(node->id, "[node %s] Listen with no info_hash", node->toString().c_str());
         throw net::DhtProtocolException {
             net::DhtProtocolException::NON_AUTHORITATIVE_INFORMATION,
             net::DhtProtocolException::LISTEN_NO_INFOHASH
@@ -2444,7 +2444,7 @@ Dht::onListen(Sp<Node> node, const InfoHash& hash, const Blob& token, size_t soc
     }
     if (not tokenMatch(token, node->getAddr())) {
         if (logger_)
-            logger_->w(hash, node->id, "[node %s] incorrect token %s for 'listen'", node->toString().c_str(), hash.toString().c_str());
+            logger_->w(hash, node->id, "[node %s] Incorrect token %s for 'listen'", node->toString().c_str(), hash.toString().c_str());
         throw net::DhtProtocolException {net::DhtProtocolException::UNAUTHORIZED, net::DhtProtocolException::LISTEN_WRONG_TOKEN};
     }
     Query q = query;
@@ -2456,7 +2456,7 @@ void
 Dht::onListenDone(const Sp<Node>& /* node */, net::RequestAnswer& /* answer */, Sp<Search>& sr)
 {
     // if (logger_)
-    //     logger_->d(sr->id, node->id, "[search %s] [node %s] got listen confirmation",
+    //     logger_->d(sr->id, node->id, "[search %s] [node %s] Got listen confirmation",
     //            sr->id.toString().c_str(), node->toString().c_str(), answer.values.size());
 
     if (not sr->done) {
@@ -2476,7 +2476,7 @@ Dht::onAnnounce(Sp<Node> n,
     auto& node = *n;
     if (not hash) {
         if (logger_)
-            logger_->w(node.id, "put with no info_hash");
+            logger_->w(node.id, "Put with no info_hash");
         throw net::DhtProtocolException {
             net::DhtProtocolException::NON_AUTHORITATIVE_INFORMATION,
             net::DhtProtocolException::PUT_NO_INFOHASH
@@ -2484,7 +2484,7 @@ Dht::onAnnounce(Sp<Node> n,
     }
     if (!tokenMatch(token, node.getAddr())) {
         if (logger_)
-            logger_->w(hash, node.id, "[node %s] incorrect token %s for 'put'", node.toString().c_str(), hash.toString().c_str());
+            logger_->w(hash, node.id, "[node %s] Incorrect token %s for 'put'", node.toString().c_str(), hash.toString().c_str());
         throw net::DhtProtocolException {net::DhtProtocolException::UNAUTHORIZED, net::DhtProtocolException::PUT_WRONG_TOKEN};
     }
     {
@@ -2493,7 +2493,7 @@ Dht::onAnnounce(Sp<Node> n,
         auto closest_nodes = buckets(node.getFamily()).findClosestNodes(hash, scheduler.time(), SEARCH_NODES);
         if (closest_nodes.size() >= TARGET_NODES and hash.xorCmp(closest_nodes.back()->id, myid) < 0) {
             if (logger_)
-                logger_->w(hash, node.id, "[node %s] announce too far from the target. Dropping value.", node.toString().c_str());
+                logger_->w(hash, node.id, "[node %s] Announce too far from the target. Dropping value.", node.toString().c_str());
             return {};
         }
     }
@@ -2502,7 +2502,7 @@ Dht::onAnnounce(Sp<Node> n,
     for (const auto& v : values) {
         if (v->id == Value::INVALID_ID) {
             if (logger_)
-                logger_->w(hash, node.id, "[value %s] incorrect value id", hash.toString().c_str());
+                logger_->w(hash, node.id, "[value %s] Incorrect value id", hash.toString().c_str());
             throw net::DhtProtocolException {
                 net::DhtProtocolException::NON_AUTHORITATIVE_INFORMATION,
                 net::DhtProtocolException::PUT_INVALID_ID
@@ -2514,17 +2514,17 @@ Dht::onAnnounce(Sp<Node> n,
             if (*lv == *vc) {
                 storageRefresh(hash, v->id);
                 if (logger_)
-                    logger_->d(hash, node.id, "[store %s] [node %s] refreshed value %016" PRIx64, hash.toString().c_str(), node.toString().c_str(), v->id);
+                    logger_->d(hash, node.id, "[store %s] [node %s] Refreshed value %016" PRIx64, hash.toString().c_str(), node.toString().c_str(), v->id);
             } else {
                 const auto& type = getType(lv->type);
                 if (type.editPolicy(hash, lv, vc, node.id, node.getAddr())) {
                     if (logger_)
-                        logger_->d(hash, node.id, "[store %s] editing %s",
+                        logger_->d(hash, node.id, "[store %s] Editing %s",
                             hash.toString().c_str(), vc->toString().c_str());
                     storageStore(hash, vc, created, node.getAddr());
                 } else {
                     if (logger_)
-                        logger_->d(hash, node.id, "[store %s] rejecting edition of %s because of storage policy",
+                        logger_->d(hash, node.id, "[store %s] Rejecting edition of %s because of storage policy",
                             hash.toString().c_str(), vc->toString().c_str());
                 }
             }
@@ -2533,11 +2533,11 @@ Dht::onAnnounce(Sp<Node> n,
             const auto& type = getType(vc->type);
             if (type.storePolicy(hash, vc, node.id, node.getAddr())) {
                 // if (logger_)
-                //     logger_->d(hash, node.id, "[store %s] storing %s", hash.toString().c_str(), std::to_string(vc->id).c_str());
+                //     logger_->d(hash, node.id, "[store %s] Storing %s", hash.toString().c_str(), std::to_string(vc->id).c_str());
                 storageStore(hash, vc, created, node.getAddr());
             } else {
                 if (logger_)
-                    logger_->d(hash, node.id, "[store %s] rejecting storage of %s",
+                    logger_->d(hash, node.id, "[store %s] Rejecting storage of %s",
                         hash.toString().c_str(), vc->toString().c_str());
             }
         }
@@ -2552,15 +2552,15 @@ Dht::onRefresh(Sp<Node> node, const InfoHash& hash, const Blob& token, const Val
 
     if (not tokenMatch(token, node->getAddr())) {
         if (logger_)
-            logger_->w(hash, node->id, "[node %s] incorrect token %s for 'put'", node->toString().c_str(), hash.toString().c_str());
+            logger_->w(hash, node->id, "[node %s] Incorrect token %s for 'put'", node->toString().c_str(), hash.toString().c_str());
         throw DhtProtocolException {DhtProtocolException::UNAUTHORIZED, DhtProtocolException::PUT_WRONG_TOKEN};
     }
     if (storageRefresh(hash, vid)) {
         if (logger_)
-            logger_->d(hash, node->id, "[store %s] [node %s] refreshed value %016" PRIx64, hash.toString().c_str(), node->toString().c_str(), vid);
+            logger_->d(hash, node->id, "[store %s] [node %s] Refreshed value %016" PRIx64, hash.toString().c_str(), node->toString().c_str(), vid);
     } else {
         if (logger_)
-            logger_->d(hash, node->id, "[store %s] [node %s] got refresh for unknown value",
+            logger_->d(hash, node->id, "[store %s] [node %s] Got refresh for unknown value",
                 hash.toString().c_str(), node->toString().c_str());
         throw DhtProtocolException {DhtProtocolException::NOT_FOUND, DhtProtocolException::STORAGE_NOT_FOUND};
     }
@@ -2583,7 +2583,7 @@ Dht::storageRefresh(const InfoHash& id, Value::Id vid)
             for (const auto& node_listeners : st.listeners) {
                 for (const auto& l : node_listeners.second) {
                     if (logger_)
-                        logger_->w(id, node_listeners.first->id, "[store %s] [node %s] sending refresh",
+                        logger_->w(id, node_listeners.first->id, "[store %s] [node %s] Sending refresh",
                             id.toString().c_str(),
                             node_listeners.first->toString().c_str());
                     Blob ntoken = makeToken(node_listeners.first->getAddr(), false);
@@ -2608,7 +2608,7 @@ void
 Dht::onAnnounceDone(const Sp<Node>& node, net::RequestAnswer& answer, Sp<Search>& sr)
 {
     if (logger_)
-        logger_->d(sr->id, node->id, "[search %s] [node %s] got reply to put!",
+        logger_->d(sr->id, node->id, "[search %s] [node %s] Got reply to put!",
             sr->id.toString().c_str(), node->toString().c_str());
     searchSendGetValues(sr);
     sr->checkAnnounced(answer.vid);

src/dht_proxy_client.cpp

@@ -106,13 +106,26 @@ getRandomSessionId(size_t length = 8) {
 DhtProxyClient::DhtProxyClient() {}
 
 DhtProxyClient::DhtProxyClient(
-        std::shared_ptr<dht::crypto::Certificate> serverCA, dht::crypto::Identity clientIdentity,
-        std::function<void()> signal, const std::string& serverHost,
-        const std::string& pushClientId, std::shared_ptr<dht::Logger> logger)
+        std::shared_ptr<dht::crypto::Certificate> serverCA,
+        dht::crypto::Identity clientIdentity,
+        std::function<void()> signal,
+        const std::string& serverHost,
+        const std::string& userAgent,
+        const std::string& pushClientId,
+        const std::string& pushToken,
+        const std::string& pushTopic,
+        const std::string& pushPlatform,
+        std::shared_ptr<dht::Logger> logger
+)
     : DhtInterface(logger)
     , proxyUrl_(serverHost)
     , clientIdentity_(clientIdentity), serverCertificate_(serverCA)
-    , pushClientId_(pushClientId), pushSessionId_(getRandomSessionId())
+    , userAgent_(userAgent)
+    , pushClientId_(pushClientId)
+    , pushSessionId_(getRandomSessionId())
+    , deviceKey_(pushToken)
+    , notificationTopic_(pushTopic)
+    , platform_(pushPlatform)
     , loopSignal_(signal)
     , jsonReader_(Json::CharReaderBuilder{}.newCharReader())
 {
@@ -484,7 +497,7 @@ DhtProxyClient::buildRequest(const std::string& target)
         request->set_certificate_authority(serverCertificate_);
     if (clientIdentity_.first and clientIdentity_.second)
         request->set_identity(clientIdentity_);
-    request->set_header_field(restinio::http_field_t::user_agent, "RESTinio client");
+    request->set_header_field(restinio::http_field_t::user_agent, userAgent_);
     return request;
 }
 
@@ -800,8 +813,8 @@ SockAddr
 DhtProxyClient::parsePublicAddress(const Json::Value& val)
 {
     auto public_ip = val.asString();
-    auto hostAndService = splitPort(public_ip);
-    auto sa = SockAddr::resolve(hostAndService.first);
+    auto [host, service] = splitPort(public_ip);
+    auto sa = SockAddr::resolve(host);
     if (sa.empty()) return {};
     return sa.front().getMappedIPv4();
 }
@@ -1381,4 +1394,24 @@ DhtProxyClient::fillBody(bool resubscribe)
 }
 #endif // OPENDHT_PUSH_NOTIFICATIONS
 
+void
+DhtProxyClient::setPushNotificationToken([[maybe_unused]] const std::string& token) {
+#ifdef OPENDHT_PUSH_NOTIFICATIONS
+    std::unique_lock<std::mutex> l(lockCurrentProxyInfos_);
+    if (deviceKey_ != token) {
+        deviceKey_ = token;
+        if (statusIpv4_ == NodeStatus::Connected || statusIpv6_ == NodeStatus::Connected) {
+            if (logger_)
+                logger_->d("[proxy:client] [push] token changed, resubscribing");
+            for (auto& search : searches_) {
+                for (auto& listener : search.second.listeners) {
+                    resubscribe(search.first, listener.first, listener.second);
+                }
+            }
+        }
+    }
+#endif
+}
+
+
 } // namespace dht

src/dht_proxy_server.cpp

@@ -223,6 +223,7 @@ DhtProxyServer::DhtProxyServer(const std::shared_ptr<DhtRunner>& dht,
     :   ioContext_(std::make_shared<asio::io_context>()),
         dht_(dht), persistPath_(config.persistStatePath), logger_(logger),
         printStatsTimer_(std::make_unique<asio::steady_timer>(*ioContext_, 3s)),
+        serverStartTime_(clock::now()),
         connListener_(std::make_shared<ConnectionListener>(std::bind(&DhtProxyServer::onConnectionClosed, this, std::placeholders::_1))),
         pushServer_(config.pushServer),
         bundleId_(config.bundleId)
@@ -542,7 +543,15 @@ DhtProxyServer::updateStats(std::shared_ptr<NodeInfo> info) const
     stats.requestRate = count / dt.count();
 #ifdef OPENDHT_PUSH_NOTIFICATIONS
     stats.pushListenersCount = pushListeners_.size();
+    {
+        std::lock_guard lk(pushStatsMutex_);
+        stats.androidPush = androidPush_;
+        stats.iosPush = iosPush_;
+        stats.unifiedPush = unifiedPush_;
+    }
 #endif
+    stats.serverStartTime = serverStartTime_;
+    stats.lastUpdated = now;
     stats.totalPermanentPuts = 0;
     std::for_each(puts_.begin(), puts_.end(), [&stats](const auto& put) {
         stats.totalPermanentPuts += put.second.puts.size();
@@ -622,6 +631,8 @@ DhtProxyServer::createRestRouter()
     router->http_get("/key/:hash", std::bind(&DhtProxyServer::get, this, _1, _2));
     // key.post
     router->http_post("/key/:hash", std::bind(&DhtProxyServer::put, this, _1, _2));
+    router->add_handler(restinio::http_method_options(),
+                        "/key/:hash/listen", std::bind(&DhtProxyServer::options, this, _1, _2));
     // key.listen
     router->http_get("/key/:hash/listen", std::bind(&DhtProxyServer::listen, this, _1, _2));
 #ifdef OPENDHT_PUSH_NOTIFICATIONS
@@ -1107,10 +1118,14 @@ DhtProxyServer::sendPushNotification(const std::string& token, Json::Value&& jso
             notification["tokens"] = std::move(tokens);
             notification["platform"] = type == PushType::Android ? 2 : 1;
             notification["data"] = std::move(json);
-            notification["priority"] = highPriority ? "high" : "normal";
-            if (type == PushType::Android)
-                notification["time_to_live"] = 3600 * 24; // 24 hours
-            else {
+            auto priority = highPriority ? "high" : "normal";
+            if (type == PushType::Android) {
+                Json::Value androidConfig(Json::objectValue);
+                androidConfig["priority"] = priority;
+                androidConfig["ttl"] = "86400s"; // time to live = 24 hours
+                notification["android"] = std::move(androidConfig);
+            } else {
+                notification["priority"] = priority;
                 const auto expiration = std::chrono::system_clock::now() + std::chrono::hours(24);
                 uint32_t exp = std::chrono::duration_cast<std::chrono::seconds>(expiration.time_since_epoch()).count();
                 notification["expiration"] = exp;
@@ -1151,6 +1166,21 @@ DhtProxyServer::sendPushNotification(const std::string& token, Json::Value&& jso
             requests_[reqid] = request;
         }
         request->send();
+        // For monitoring purposes
+        std::lock_guard lk(pushStatsMutex_);
+        switch (type) {
+        case PushType::Android:
+            androidPush_.increment(highPriority);
+            break;
+        case PushType::iOS:
+            iosPush_.increment(highPriority);
+            break;
+        case PushType::UnifiedPush:
+            unifiedPush_.increment(highPriority);
+            break;
+        default:
+            break;
+        }
     }
     catch (const std::exception &e){
         if (logger_)

src/dhtrunner.cpp

@@ -1120,13 +1120,13 @@ DhtRunner::enableProxy(bool proxify)
                         cv.notify_all();
                     }
                 },
-                config_.proxy_server, config_.push_node_id, logger_);
-        if (not config_.push_token.empty())
-            dht_via_proxy->setPushNotificationToken(config_.push_token);
-        if (not config_.push_topic.empty())
-            dht_via_proxy->setPushNotificationTopic(config_.push_topic);
-        if (not config_.push_platform.empty())
-            dht_via_proxy->setPushNotificationPlatform(config_.push_platform);
+                config_.proxy_server,
+                config_.proxy_user_agent,
+                config_.push_node_id,
+                config_.push_token,
+                config_.push_topic,
+                config_.push_platform,
+                logger_);
         dht_ = std::make_unique<SecureDht>(std::move(dht_via_proxy), config_.dht_config, identityAnnouncedCb_, logger_);
         // and use it
         use_proxy = proxify;

src/http.cpp

@@ -54,7 +54,7 @@ constexpr const char HTTPS_PROTOCOL[] = "https://";
 constexpr const char ORIGIN_PROTOCOL[] = "//";
 constexpr unsigned MAX_REDIRECTS {5};
 
-Url::Url(const std::string& url): url(url)
+Url::Url(std::string_view url): url(url)
 {
     size_t addr_begin = 0;
     // protocol
@@ -69,10 +69,9 @@ Url::Url(const std::string& url): url(url)
     size_t addr_size = url.substr(addr_begin).find("/");
     if (addr_size == std::string::npos)
         addr_size = url.size() - addr_begin;
-    auto host_service = splitPort(url.substr(addr_begin, addr_size));
-    host = host_service.first;
-    if (!host_service.second.empty())
-        service = host_service.second;
+    auto [h, s] = splitPort(url.substr(addr_begin, addr_size));
+    host = std::move(h);
+    service = std::move(s);
     // target, query and fragment
     size_t query_begin = url.find("?");
     auto addr_end = addr_begin + addr_size;

src/network_engine.cpp

@@ -526,8 +526,13 @@ NetworkEngine::process(std::unique_ptr<ParsedMessage>&& msg, const SockAddr& fro
             throw DhtProtocolException {DhtProtocolException::UNKNOWN_TID, "Can't find socket", msg->id};
         node->received(now, {});
         onNewNode(node, 2);
-        deserializeNodes(*msg, from);
-        rsocket->on_receive(node, std::move(*msg));
+        try {
+            deserializeNodes(*msg, from);
+            rsocket->on_receive(node, std::move(*msg));
+        } catch (const DhtProtocolException& e) {
+            if (logger_)
+                logger_->w("Can't deserialize nodes %s", e.what());
+        }
     }
     else if (msg->type == MessageType::Error or msg->type == MessageType::Reply) {
         auto rsocket = node->getSocket(msg->tid);
@@ -590,13 +595,22 @@ NetworkEngine::process(std::unique_ptr<ParsedMessage>&& msg, const SockAddr& fro
                     r.node->authSuccess();
                 }
                 r.reply_time = scheduler.time();
-
-                deserializeNodes(*msg, from);
-                r.setDone(std::move(*msg));
+                try {
+                    deserializeNodes(*msg, from);
+                    r.setDone(std::move(*msg));
+                } catch (const DhtProtocolException& e) {
+                    if (logger_)
+                        logger_->w("Can't deserialize nodes %s", e.what());
+                }
                 break;
             } else { /* request socket data */
-                deserializeNodes(*msg, from);
-                rsocket->on_receive(node, std::move(*msg));
+                try {
+                    deserializeNodes(*msg, from);
+                    rsocket->on_receive(node, std::move(*msg));
+                } catch (const DhtProtocolException& e) {
+                    if (logger_)
+                        logger_->w("Can't deserialize nodes %s", e.what());
+                }
             }
             break;
         default:

src/search.h

@@ -45,7 +45,7 @@ struct Dht::Announce {
     bool permanent;
     Sp<Value> value;
     time_point created;
-    DoneCallback callback;
+    std::vector<DoneCallback> callbacks;
 };
 
 struct Dht::SearchNode {
@@ -442,8 +442,9 @@ struct Dht::Search {
             g.second.done_cb = {};
         }
         for (auto& a : announce) {
-            a.callback(false, {});
-            a.callback = {};
+            for (auto& cb : a.callbacks)
+                cb(false, {});
+            a.callbacks.clear();
         }
     }
 
@@ -627,7 +628,9 @@ struct Dht::Search {
             return a.value->id == value->id;
         });
         if (a_sr == announce.end()) {
-            announce.emplace_back(Announce {permanent, value, created, callback});
+            auto& a = announce.emplace_back(Announce {permanent, value, created, {}} );
+            if (callback)
+                a.callbacks.emplace_back(std::move(callback));
             for (auto& n : nodes) {
                 n->probe_query.reset();
                 n->acked[value->id].req.reset();
@@ -635,24 +638,26 @@ struct Dht::Search {
         } else {
             a_sr->permanent = permanent;
             a_sr->created = created;
-            if (a_sr->value != value) {
+            if (a_sr->value != value && *a_sr->value != *value) {
+                // Value is updated, previous ops are failed
                 a_sr->value = value;
                 for (auto& n : nodes) {
                     n->acked[value->id].req.reset();
                     n->probe_query.reset();
                 }
-            }
-            if (isAnnounced(value->id)) {
-                if (a_sr->callback)
-                    a_sr->callback(true, {});
-                a_sr->callback = {};
+                for (auto& cb: a_sr->callbacks)
+                    cb(false, {});
+                a_sr->callbacks.clear();
+                if (callback)
+                    a_sr->callbacks.emplace_back(std::move(callback));
+            } else if (isAnnounced(value->id)) {
+                // Same value, already announced
                 if (callback)
                     callback(true, {});
-                return;
             } else {
-                if (a_sr->callback)
-                    a_sr->callback(false, {});
-                a_sr->callback = callback;
+                // Same value, not announced yet
+                if (callback)
+                    a_sr->callbacks.emplace_back(std::move(callback));
             }
         }
     }
@@ -722,8 +727,8 @@ struct Dht::Search {
             std::vector<DoneCallback> a_cbs;
             a_cbs.reserve(announce.size());
             for (auto ait = announce.begin() ; ait != announce.end(); ) {
-                if (ait->callback)
-                    a_cbs.emplace_back(std::move(ait->callback));
+                a_cbs.insert(a_cbs.end(), std::make_move_iterator(ait->callbacks.begin()), std::make_move_iterator(ait->callbacks.end()));
+                ait->callbacks.clear();
                 if (not ait->permanent)
                     ait = announce.erase(ait);
                 else
@@ -747,9 +752,11 @@ struct Dht::Search {
                 if (vid != Value::INVALID_ID and (!a.value || a.value->id != vid))
                     return true;
                 if (isAnnounced(a.value->id)) {
-                    if (a.callback) {
-                        a.callback(true, getNodes());
-                        a.callback = nullptr;
+                    if (!a.callbacks.empty()) {
+                        const auto& nodes = getNodes();
+                        for (auto& cb : a.callbacks)
+                            cb(true, nodes);
+                        a.callbacks.clear();
                     }
                     if (not a.permanent)
                         return false;

src/utils.cpp

@@ -44,23 +44,23 @@ const HexMap hex_map = {};
 static constexpr std::array<uint8_t, 12> MAPPED_IPV4_PREFIX {{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xff, 0xff}};
 
 std::pair<std::string, std::string>
-splitPort(const std::string& s) {
+splitPort(std::string_view s) {
     if (s.empty())
         return {};
     if (s[0] == '[') {
         std::size_t closure = s.find_first_of(']');
         std::size_t found = s.find_last_of(':');
-        if (closure == std::string::npos)
-            return {s, ""};
-        if (found == std::string::npos or found < closure)
-            return {s.substr(1,closure-1), ""};
-        return {s.substr(1,closure-1), s.substr(found+1)};
+        if (closure == std::string_view::npos)
+            return {std::string(s), ""};
+        if (found == std::string_view::npos or found < closure)
+            return {std::string(s.substr(1,closure-1)), ""};
+        return {std::string(s.substr(1,closure-1)), std::string(s.substr(found+1))};
     }
     std::size_t found = s.find_last_of(':');
     std::size_t first = s.find_first_of(':');
-    if (found == std::string::npos or found != first)
-        return {s, ""};
-    return {s.substr(0,found), s.substr(found+1)};
+    if (found == std::string_view::npos or found != first)
+        return {std::string(s), ""};
+    return {std::string(s.substr(0,found)), std::string(s.substr(found+1))};
 }
 
 std::vector<SockAddr>

src/value.cpp

@@ -271,6 +271,7 @@ Value::encrypt(const crypto::PrivateKey& from, const crypto::PublicKey& to)
     setRecipient(to.getId());
     sign(from);
     Value nv {id};
+    nv.priority = priority;
     nv.setCypher(to.encrypt(getToEncrypt()));
     return nv;
 }

tests/dhtrunnertester.cpp

@@ -90,6 +90,59 @@ DhtRunnerTester::testGetPut() {
     CPPUNIT_ASSERT(vals.front()->data == val_data);
 }
 
+void
+DhtRunnerTester::testPutDuplicate() {
+    auto key = dht::InfoHash::get("123");
+    auto val = std::make_shared<dht::Value>("hey");
+    val->id = 42;
+    auto val_data = val->data;
+    std::promise<bool> p1;
+    std::promise<bool> p2;
+    node2.put(key, val, [&](bool ok){
+        p1.set_value(ok);
+    });
+    node2.put(key, val, [&](bool ok){
+        p2.set_value(ok);
+    });
+    auto p1ret = p1.get_future().get();
+    auto p2ret = p2.get_future().get();
+    CPPUNIT_ASSERT(p1ret);
+    CPPUNIT_ASSERT(p2ret);
+    auto vals = node1.get(key).get();
+    CPPUNIT_ASSERT(not vals.empty());
+    CPPUNIT_ASSERT(vals.size() == 1);
+    CPPUNIT_ASSERT(vals.front()->data == val_data);
+}
+
+
+void
+DhtRunnerTester::testPutOverride() {
+    auto key = dht::InfoHash::get("123");
+    auto val = std::make_shared<dht::Value>("meh");
+    val->id = 42;
+    auto val2 = std::make_shared<dht::Value>("hey");
+    val2->id = 42;
+    CPPUNIT_ASSERT_EQUAL(val->id, val2->id);
+    auto val_data = val2->data;
+    std::promise<bool> p1;
+    std::promise<bool> p2;
+    node2.put(key, val, [&](bool ok){
+        p1.set_value(ok);
+    });
+    node2.put(key, val2, [&](bool ok){
+        p2.set_value(ok);
+    });
+    auto p1ret = p1.get_future().get();
+    auto p2ret = p2.get_future().get();
+    CPPUNIT_ASSERT(!p1ret);
+    CPPUNIT_ASSERT(p2ret);
+    auto vals = node1.get(key).get();
+    CPPUNIT_ASSERT(not vals.empty());
+    CPPUNIT_ASSERT(vals.size() == 1);
+    CPPUNIT_ASSERT(vals.front()->data == val_data);
+}
+
+
 void
 DhtRunnerTester::testListen() {
     std::mutex mutex;

tests/dhtrunnertester.h

@@ -31,6 +31,8 @@ class DhtRunnerTester : public CppUnit::TestFixture {
     CPPUNIT_TEST_SUITE(DhtRunnerTester);
     CPPUNIT_TEST(testConstructors);
     CPPUNIT_TEST(testGetPut);
+    CPPUNIT_TEST(testPutDuplicate);
+    CPPUNIT_TEST(testPutOverride);
     CPPUNIT_TEST(testListen);
     CPPUNIT_TEST(testListenLotOfBytes);
     CPPUNIT_TEST(testIdOps);
@@ -55,6 +57,14 @@ class DhtRunnerTester : public CppUnit::TestFixture {
      * Test get and put methods
      */
     void testGetPut();
+    /**
+     * Test get and multiple put
+     */
+    void testPutDuplicate();
+    /**
+     * Test get and multiple put with changing value
+     */
+    void testPutOverride();
     /**
      * Test listen method
      */

tools/dhtchat.cpp

@@ -107,8 +107,8 @@ main(int argc, char **argv)
                     token = dht.listen<dht::ImMessage>(room, [&](dht::ImMessage&& msg) {
                         if (msg.from != myid)
                             std::cout << msg.from.toString() << " at " << printTime(msg.date)
-                                      << " (took " << print_dt(std::chrono::system_clock::now() - std::chrono::system_clock::from_time_t(msg.date))
-                                      << "s) " << (msg.to == myid ? "ENCRYPTED ":"") << ": " << msg.id << " - " << msg.msg << std::endl;
+                                      << " (took " << print_duration(std::chrono::system_clock::now() - std::chrono::system_clock::from_time_t(msg.date))
+                                      << ") " << (msg.to == myid ? "ENCRYPTED ":"") << ": " << msg.id << " - " << msg.msg << std::endl;
                         return true;
                     });
                     connected = true;

tools/dhtcnode.c

@@ -23,6 +23,7 @@
 #include <unistd.h>
 #include <stdlib.h>
 #include <stdatomic.h>
+#include <inttypes.h>
 
 #include <getopt.h>
 #include <readline/readline.h>
@@ -38,6 +39,10 @@ struct listen_context {
     dht_op_token* token;
     size_t count;
 };
+struct put_context {
+    dht_runner* runner;
+    dht_value* value;
+};
 
 bool dht_value_callback(const dht_value* value, bool expired, void* user_data)
 {
@@ -67,8 +72,10 @@ void dht_get_done_callback(bool ok, void* user_data)
 
 void dht_put_done_callback(bool ok, void* user_data)
 {
-    dht_runner* runner = (dht_runner*)user_data;
-    printf("Put completed: %s\n", ok ? "success !" : "failure :-(");
+    struct put_context* ctx = (struct put_context*)user_data;
+    printf("Put completed (id: %" PRIx64 "): %s\n", dht_value_get_id(ctx->value), ok ? "success !" : "failure :-(");
+    dht_value_unref(ctx->value);
+    free(ctx);
 }
 
 void dht_shutdown_callback(void* user_data)
@@ -261,11 +268,13 @@ int main(int argc, char **argv)
         else if (!strcmp(cmd, "p")) {
             key = parse_key(arg);
             dht_value* val = dht_value_new_from_string(value);
-            dht_runner_put(runner, &key, val, dht_put_done_callback, runner, true);
-            dht_value_unref(val);
+            struct put_context* ctx = malloc(sizeof(struct put_context));
+            ctx->runner = runner;
+            ctx->value = val;
+            dht_runner_put(runner, &key, val, dht_put_done_callback, ctx, true);
         }
         else {
-            printf("Unkown command: %s\n", cmd);
+            printf("Unknown command: %s\n", cmd);
         }
     }
 

tools/proxy_node.html

@@ -1,190 +1,78 @@
 <!DOCTYPE html>
 <html lang="en">
-	<head>
-		<meta charset="utf-8" />
-		<meta name="viewport" content="width=device-width, initial-scale=1.0, maximum-scale=1.0, user-scalable=no">
-		<title>OpenDHT tester</title>
-		<link rel="stylesheet" href="https://cdnjs.cloudflare.com/ajax/libs/twitter-bootstrap/3.3.7/css/bootstrap.min.css" />
-        <link rel="stylesheet" href="https://cdnjs.cloudflare.com/ajax/libs/font-awesome/4.7.0/css/font-awesome.css" />
-		<script src="https://ajax.googleapis.com/ajax/libs/jquery/3.2.1/jquery.min.js"></script>
-		<script type="text/javascript" src="https://cdnjs.cloudflare.com/ajax/libs/twitter-bootstrap/3.3.7/js/bootstrap.min.js"></script>
-		<script type="text/javascript">"use strict";
-let onGet;
-let onPut;
-let setServer;
-const valueGetElement = function(o) {
-    const d = window.atob(o.data);
-    return d;
-};
-$(function() {
-    let request = undefined;
-    let server;
-    const getTools = $("#getTools");
-    const getBtn = $("#getBtn");
-    const getDropdown = $("#getDropdown");
-    const listenBtn = $("#listenBtn").click(function(){onGet('LISTEN');});
-    const setGetRequest = function() {
-        getBtn.button('loading');
-        getStopBtn.appendTo(getTools);
-        getDropdown.hide();
-    }
-    const clearGetRequest = function() {
-        if (request === undefined)
-            return;
-        request.abort();
-        request = undefined;
-        getStopBtn.detach();
-        getDropdown.show();
-        getBtn.button('reset');
-    }
-    const getStopBtn = $("#getStopBtn").detach().click(clearGetRequest);
-    const putBtn = $("#putBtn");
-    const result = $("#dhtResult");
-    const group = $('<ul class="list-group"/>').appendTo(result);
-    onGet = function (method) {
-        if (request !== undefined)
-            return false;
-        const input = $("#getKey").val();
-        group.empty();
-        let lastAppended = 0;
-        let start = new Date().getTime();
-        request = new XMLHttpRequest();
-        request.onreadystatechange = function(event) {
-            if (this.readyState >= XMLHttpRequest.LOADING) {
-                if (this.readyState == XMLHttpRequest.DONE) {
-                    clearGetRequest();
-                }
-                if (this.status === 200) {
-                    const elements = this.responseText.split("\n");
-                    const elementsLength = elements.length;
-                    const now = new Date().getTime();
-                    for (let i = lastAppended; i < elementsLength; i++) {
-                        const element = elements[i];
-                        if (!element || element.length == 0)
-                            return;
-                        const o = JSON.parse(element);
-                        if (o.expired) {
-                            $('#value'+o.id).slideUp(100, (e) => $(e.target).remove());
-                        } else {
-                            const d = window.atob(o.data);
-                            const delay = Math.max(0, start-now);
-                            $('<li class="list-group-item" id="value'+o.id+'"/>').append(valueGetElement(o)).appendTo(group).hide().delay(delay).slideDown(100);
-                            lastAppended = i+1;
-                            start = Math.max(start, now)+25;
-                        }
-                    }
-                } else if (this.status !== 0) {
-                    group.empty().append($('<li class="list-group-item list-group-item-danger"/>').text("Error loading content: " + this.statusText));
-                }
-            }
-        };
-        request.onerror = function(event) {
-            clearGetRequest();
-            group.empty().append($('<li class="list-group-item list-group-item-danger"/>').text("Error loading content."));
-        };
-        request.open(method, server + input, true);
-        request.send(null);
-        setGetRequest();
-        return false;
-    };
-
-    onPut = function( ) {
-        const key = $("#getKey").val();
-        const value = $("#putValue").val();
-        $.ajax({
-            url: server + key,
-            type: 'POST',
-            data: JSON.stringify({
-                data:window.btoa(value)
-            }),
-            contentType: 'application/json; charset=utf-8',
-            dataType: 'json',
-            success: function( result ) {
-                putBtn.button('reset');
-                //$('<li class="list-group-item list-group-item-success"/>').append(valueGetElement(result)).appendTo(group.empty());
-            },
-            error: function(result) {
-                putBtn.button('reset');
-                group.empty().append($('<li class="list-group-item list-group-item-danger"/>').text(result.statusText));
-            }
-        });
-        putBtn.button('loading');
-        return false;
-    };
-
-    const serverValue = $("#serverValue");
-    const serverStatus = $("#serverStatus");
-    const serverBtn = $("#serverBtn");
-    setServer = function(event) {
-        server = serverValue.val() + '/';
-        serverStatus.empty();
-        serverBtn.button('loading');
-        $.getJSON(server, function(data){
-            $('<span><b>Node</b> '+data.node_id+'</span>').appendTo(serverStatus).hide().fadeIn();
-        }).fail(function(error) {
-            var message = " Cant' access node."
-            if (serverValue.val().indexOf("https") != -1){
-                message += "</br></br>Self-signed certificate must be allowed in browser."
-            }
-            serverStatus.html("<div class='alert alert-danger' style='margin-bottom: 0px;'>" +
-                              "<span class='glyphicon glyphicon-remove' aria-hidden='true'></span>" +
-                              message + "</div>");
-        }).always(function(error) {
-            serverBtn.button('reset');
-        });
-        return false;
-    };
-    setServer();
-});
-	</script>
+<head>
+    <meta charset="utf-8" />
+    <meta name="viewport" content="width=device-width, initial-scale=1.0, maximum-scale=1.0, user-scalable=no">
+    <title>OpenDHT web node</title>
+    <link rel="stylesheet" href="https://cdnjs.cloudflare.com/ajax/libs/twitter-bootstrap/5.3.3/css/bootstrap.min.css" />
+    <script src="https://cdn.jsdelivr.net/npm/bootstrap@5.3.3/dist/js/bootstrap.bundle.min.js" crossorigin="anonymous"></script>
+    <link rel="stylesheet" href="https://cdn.jsdelivr.net/npm/bootstrap-icons@1.11.3/font/bootstrap-icons.min.css">
 </head>
+
 <body>
-    <div class="container" style="max-width: 730px;">
-        <header class="page-header">
+    <div class="container" style="max-width: 880px;">
+        <header class="m-3 p-3">
             <div class="row">
-                <div class="col-sm-5">
-                    <h1>OpenDHT tester</h1>
+                <div class="col-sm-5 d-flex align-items-center">
+                    <h2>OpenDHT web node</h2>
                 </div>
                 <div class="col-sm-7">
-                    <div class="well well-sm" style="margin-top:10px; margin-bottom:0px;">
-                        <form id="serverForm" class="form-inline" onsubmit="return setServer();" style="margin-bottom:4px;">
+                    <div class="card card-body bg-light" style="margin-top:10px; margin-bottom:0px;">
+                        <form id="serverForm" class="form-inline" onsubmit="return setServer();"
+                            style="margin-bottom:4px;">
                             <div class="input-group">
-                                <input type="text" class="form-control" id="serverValue" placeholder="Proxy server" value="http://127.0.0.1:8080"/>
-                                <span class="input-group-btn">
-                                    <button id="serverBtn" type="submit" class="btn btn-default" data-loading-text="<i class='fa fa-circle-o-notch fa-spin'></i>"><span class="glyphicon glyphicon-refresh" aria-hidden="true"></span></button>
-                                </span>
+                                <input type="text" class="form-control" id="serverValue" placeholder="Proxy server" value="http://127.0.0.1:8080" />
+                                <button id="serverBtn" type="submit" class="btn btn-outline-secondary">
+                                    <div class='spinner-border spinner-border-sm' role='status'
+                                        style="display: none;">
+                                        <span class='visually-hidden'>Loading...</span>
+                                    </div>
+                                    <span class="btn-icon"><i class="bi bi-arrow-clockwise"></i></span>
+                                </button>
                             </div>
                         </form>
-                        <div id="serverStatus"><i class='fa fa-circle-o-notch fa-spin'></i></div>
+                        <div id="serverStatus">
+                            <div class='spinner-border spinner-border-sm' role='status'><span
+                                    class='visually-hidden'>Loading...</span></div>
+                        </div>
                     </div>
                 </div>
             </div>
         </header>
-        <div class="panel panel-default" id="dhtResult">
-                <div class="panel-heading">
-                    <div class="row">
-                    <div class="col-xs-6">
-                        <form class="form-inline" onsubmit="return onGet('GET');">
-                            <div class="input-group">
+        <div class="list-group" id="dhtResult">
+            <div class="list-group-item list-group-item-light">
+                <div class="row">
+                    <div class="col-sm-6">
+                        <form class="form-inline" onsubmit="return onGet('');">
+                            <div class="input-group" id="getTools">
                                 <input type="text" class="form-control" id="getKey" placeholder="Key" aria-label="Key" />
-                                <span class="input-group-btn" id="getTools">
-                                    <button id="getBtn" class="btn btn-default" data-loading-text="<i class='fa fa-circle-o-notch fa-spin'></i>" type="submit">Get</button>
-                                    <button id="getDropdown"type="button" class="btn btn-default dropdown-toggle" data-toggle="dropdown" aria-haspopup="true" aria-expanded="false"> <span class="caret"></span> <span class="sr-only">Toggle Dropdown</span> </button>
-                                    <ul class="dropdown-menu">
-                                        <li><a id="listenBtn" href="#">Listen</a></li>
-                                    </ul>
-                                    <button id="getStopBtn" class="btn btn-default" type="submit"><span class="glyphicon glyphicon-remove" aria-hidden="true"></span></button>
-                                </span>
+                                <button id="getBtn" class="btn btn-outline-secondary" type="submit">
+                                    <div class='spinner-border spinner-border-sm' role='status' style="display: none;">
+                                        <span class='visually-hidden'>Loading...</span>
+                                    </div>
+                                    <span class="btn-text">Get</span>
+                                </button>
+                                <button id="menuBtn" type="button" class="btn btn-outline-secondary dropdown-toggle dropdown-toggle-split" data-bs-toggle="dropdown" aria-expanded="false">
+                                    <span class="visually-hidden">Toggle Dropdown</span>
+                                </button>
+                                <ul class="dropdown-menu dropdown-menu-end">
+                                    <li><a class="dropdown-item" id="listenBtn" href="#">Listen</a></li>
+                                </ul>
+                                <button id="getStopBtn" class="btn btn-outline-secondary">Stop</button>
                             </div>
                         </form>
                     </div>
-                    <div class="col-xs-6">
+                    <div class="col-sm-6">
                         <form class="form-inline" onsubmit="return onPut();">
                             <div class="input-group">
-                                <input type="text" class="form-control input-group-input" id="putValue" placeholder="Value" />
-                                <span class="input-group-btn">
-                                    <button id="putBtn" type="submit" class="btn btn-default" data-loading-text="<i class='fa fa-circle-o-notch fa-spin'></i> Loading">Put</button>
-                                </span>
+                                <input type="text" class="form-control input-group-input" id="putValue"
+                                    placeholder="Value" />
+                                <button id="putBtn" type="submit" class="btn btn-outline-secondary">
+                                    <div class='spinner-border spinner-border-sm' role='status' style="display: none;">
+                                        <span class='visually-hidden'>Loading...</span>
+                                    </div>
+                                    <span class="btn-text">Put</span>
+                                </button>
                             </div>
                         </form>
                     </div>
@@ -192,5 +80,185 @@ $(function() {
             </div>
         </div>
     </div>
+
+    <script type="text/javascript">"use strict";
+        const valueGetElement = (o) => window.atob(o.data);
+        const getTools = document.getElementById("getTools");
+        const getBtn = document.getElementById("getBtn");
+        const menuBtn = document.getElementById("menuBtn");
+        const getStopBtn = document.getElementById("getStopBtn");
+        const listenBtn = document.getElementById("listenBtn");
+        const putBtn = document.getElementById("putBtn");
+        const group = document.getElementById("dhtResult");
+        const serverValue = document.getElementById("serverValue");
+        const serverStatus = document.getElementById("serverStatus");
+        const serverBtn = document.getElementById("serverBtn");
+
+        var request = undefined;
+        const setGetRequest = () => {
+            getBtn.disabled = true;
+            getBtn.querySelector('.spinner-border').style.display = 'inline-block';
+            getBtn.querySelector('.btn-text').style.display = 'none';
+            menuBtn.style.display = 'none';
+            getTools.appendChild(getStopBtn);
+        };
+        const clearGetRequest = () => {
+            if (request === undefined)
+                return;
+            request.abort();
+            request = undefined;
+            if (getStopBtn.parentNode) {
+                getStopBtn.parentNode.removeChild(getStopBtn);
+            }
+            menuBtn.style.display = '';
+            getBtn.disabled = false;
+            getBtn.querySelector('.spinner-border').style.display = 'none';
+            getBtn.querySelector('.btn-text').style.display = 'inline';
+        };
+
+        const clearResults = () => {
+            while (group.children.length > 1) {
+                group.removeChild(group.lastChild);
+            }
+        };
+
+        var server;
+        const setServer = (event) => {
+            server = serverValue.value + '/';
+            serverStatus.innerHTML = '';
+            serverBtn.disabled = true;
+            serverBtn.querySelector('.spinner-border').style.display = 'inline-block';
+            serverBtn.querySelector('.btn-icon').style.display = 'none';
+            var xhr = new XMLHttpRequest();
+            xhr.open('GET', server, true);
+            xhr.onreadystatechange = () => {
+                if (xhr.readyState === XMLHttpRequest.DONE) {
+                    serverBtn.disabled = false;
+                    serverBtn.querySelector('.spinner-border').style.display = 'none';
+                    serverBtn.querySelector('.btn-icon').style.display = 'inline';
+                    if (xhr.status === 200) {
+                        const data = JSON.parse(xhr.responseText);
+                        const span = document.createElement('span');
+                        span.innerHTML = '<b>Node</b> ' + data.node_id;
+                        serverStatus.appendChild(span);
+                    } else {
+                        const message = " Can't access node.";
+                        if (serverValue.value.indexOf("https") != -1) {
+                            message += "</br></br>Self-signed certificate must be allowed in browser.";
+                        }
+                        serverStatus.innerHTML = "<div class='alert alert-danger' style='margin-bottom: 0px;'>" +
+                            "<span class='glyphicon glyphicon-remove' aria-hidden='true'></span>" +
+                            message + "</div>";
+                    }
+                }
+            };
+            xhr.send();
+            return false;
+        };
+        setServer();
+
+        const onGet = (path) => {
+            if (request !== undefined)
+                return false;
+            const input = document.getElementById("getKey").value;
+            clearResults()
+            let lastAppended = 0;
+            let start = new Date().getTime();
+            request = new XMLHttpRequest();
+            request.onreadystatechange = (event) => {
+                if (request.readyState >= 2) { // LOADING
+                    if (request.status === 200) {
+                        const elements = request.responseText.split("\n");
+                        const elementsLength = elements.length;
+                        const now = new Date().getTime();
+                        for (let i = lastAppended; i < elementsLength; i++) {
+                            const element = elements[i];
+                            if (!element || element.length == 0)
+                                break;
+                            const o = JSON.parse(element);
+                            if (o.expired) {
+                                const expiredElement = document.getElementById('value' + o.id);
+                                if (expiredElement) {
+                                    expiredElement.style.display = 'none';
+                                    if (expiredElement.parentNode) {
+                                        expiredElement.parentNode.removeChild(expiredElement);
+                                    }
+                                }
+                            } else {
+                                const d = window.atob(o.data);
+                                const delay = Math.max(0, start - now);
+                                const li = document.createElement('li');
+                                li.className = 'list-group-item';
+                                li.id = 'value' + o.id;
+                                li.innerHTML = valueGetElement(o);
+                                li.style.opacity = 0;
+                                li.style.transition = 'opacity 0.1s ease-in';
+                                group.appendChild(li);
+                                setTimeout(() => {
+                                    li.style.opacity = 1;
+                                }, delay);
+                                lastAppended = i + 1;
+                                start = Math.max(start, now) + 25;
+                            }
+                        }
+                    } else if (request.status !== 0) {
+                        clearResults()
+                        const li = document.createElement('li');
+                        li.className = 'list-group-item list-group-item-danger';
+                        li.textContent = "Error loading content: " + request.statusText;
+                        group.appendChild(li);
+                    }
+                }
+                if (request.readyState == 4) { // DONE
+                    clearGetRequest();
+                }
+            };
+            request.onerror = (event) => {
+                clearGetRequest();
+                clearResults()
+                const li = document.createElement('li');
+                li.className = 'list-group-item list-group-item-danger';
+                li.textContent = "Error loading content.";
+                group.appendChild(li);
+            };
+            request.open('GET', server + 'key/' + input + path, true);
+            request.send(null);
+            setGetRequest();
+            return false;
+        };
+        const onPut = () => {
+            const key = document.getElementById("getKey").value;
+            const value = document.getElementById("putValue").value;
+            putBtn.disabled = true;
+            putBtn.querySelector('.spinner-border').style.display = 'inline-block';
+            putBtn.querySelector('.btn-text').style.display = 'none';
+            fetch(server + key, {
+                method: 'POST',
+                headers: { 'Content-Type': 'application/json; charset=utf-8' },
+                body: JSON.stringify({ data: window.btoa(value) })
+            })
+            .then(response => {
+                if (!response.ok)
+                    throw new Error(response.statusText);
+            })
+            .catch(error => {
+                group.innerHTML = '';
+                const li = document.createElement('li');
+                li.className = 'list-group-item list-group-item-danger';
+                li.textContent = error.message;
+                group.appendChild(li);
+            })
+            .finally(() => {
+                putBtn.disabled = false;
+                putBtn.querySelector('.spinner-border').style.display = 'none';
+                putBtn.querySelector('.btn-text').style.display = 'inline';
+            });
+            return false;
+        };
+
+        listenBtn.addEventListener('click', () => { onGet('/listen'); });
+        getStopBtn.parentNode.removeChild(getStopBtn);
+        getStopBtn.addEventListener('click', clearGetRequest);
+    </script>
 </body>
-</html>
+</html>