thread pool: add ExecutionContext

dht, dhtrunner: add shutdown stop argument
proxy tester: increase shutdown timeout
2026-11-09 04:49:14 +00:00 · 2021-05-31 15:46:41 -04:00 · 2021-05-31 14:43:57 -04:00 · 2021-05-31 14:43:57 -04:00 · 2021-05-28 13:17:25 -04:00 · 2021-05-28 13:17:25 -04:00
43 changed files with 828 additions and 310 deletions
@@ -7,7 +7,7 @@ include(FindPkgConfig)
 include(cmake/CheckAtomic.cmake)

 set (opendht_VERSION_MAJOR 2)
-set (opendht_VERSION_MINOR 1.9)
+set (opendht_VERSION_MINOR 2.0)
 set (opendht_VERSION ${opendht_VERSION_MAJOR}.${opendht_VERSION_MINOR})
 set (PACKAGE_VERSION ${opendht_VERSION})
 set (VERSION "${opendht_VERSION}")
@@ -71,8 +71,11 @@ if (NOT MSVC)
        )
    endif()

-    if (OPENDHT_HTTP)
+    if (OPENDHT_HTTP OR OPENDHT_PEER_DISCOVERY)
        find_path(ASIO_INCLUDE_DIR asio.hpp REQUIRED)
+    endif ()
+
+    if (OPENDHT_HTTP)
        find_package(Restinio REQUIRED)
        find_library(FMT_LIBRARY fmt)
        add_library(fmt SHARED IMPORTED)
@@ -1,7 +1,7 @@
 dnl define macros
 m4_define([opendht_major_version], 2)
-m4_define([opendht_minor_version], 1)
-m4_define([opendht_patch_version], 9)
+m4_define([opendht_minor_version], 2)
+m4_define([opendht_patch_version], 0)
 m4_define([opendht_version],
 		  [opendht_major_version.opendht_minor_version.opendht_patch_version])

@@ -17,9 +17,9 @@ RUN apt-get update && apt-get install -y \

 RUN echo "*** Downloading RESTinio ***" \
    && mkdir restinio && cd restinio \
-    && wget https://github.com/aberaud/restinio/archive/2c0b6f5e5ba04d7a74e8406a3df1fd433680599d.tar.gz \
-    && ls -l && tar -xzf 2c0b6f5e5ba04d7a74e8406a3df1fd433680599d.tar.gz \
-    && cd restinio-2c0b6f5e5ba04d7a74e8406a3df1fd433680599d/dev \
+    && wget https://github.com/aberaud/restinio/archive/2224ffedef52cb2b74645d63d871d61dbd0f165e.tar.gz \
+    && ls -l && tar -xzf 2224ffedef52cb2b74645d63d871d61dbd0f165e.tar.gz \
+    && cd restinio-2224ffedef52cb2b74645d63d871d61dbd0f165e/dev \
    && cmake -DCMAKE_INSTALL_PREFIX=/usr -DRESTINIO_TEST=OFF -DRESTINIO_SAMPLE=OFF \
             -DRESTINIO_INSTALL_SAMPLES=OFF -DRESTINIO_BENCH=OFF -DRESTINIO_INSTALL_BENCHES=OFF \
             -DRESTINIO_FIND_DEPS=ON -DRESTINIO_ALLOW_SOBJECTIZER=Off -DRESTINIO_USE_BOOST_ASIO=none . \
@@ -18,9 +18,9 @@ ENV CXX c++

 RUN echo "*** Downloading RESTinio ***" \
    && mkdir restinio && cd restinio \
-    && wget https://github.com/aberaud/restinio/archive/2c0b6f5e5ba04d7a74e8406a3df1fd433680599d.tar.gz \
-    && ls -l && tar -xzf 2c0b6f5e5ba04d7a74e8406a3df1fd433680599d.tar.gz \
-    && cd restinio-2c0b6f5e5ba04d7a74e8406a3df1fd433680599d/dev \
+    && wget https://github.com/aberaud/restinio/archive/2224ffedef52cb2b74645d63d871d61dbd0f165e.tar.gz \
+    && ls -l && tar -xzf 2224ffedef52cb2b74645d63d871d61dbd0f165e.tar.gz \
+    && cd restinio-2224ffedef52cb2b74645d63d871d61dbd0f165e/dev \
    && cmake -DCMAKE_INSTALL_PREFIX=/usr -DRESTINIO_TEST=OFF -DRESTINIO_SAMPLE=OFF \
             -DRESTINIO_INSTALL_SAMPLES=OFF -DRESTINIO_BENCH=OFF -DRESTINIO_INSTALL_BENCHES=OFF \
             -DRESTINIO_FIND_DEPS=ON -DRESTINIO_ALLOW_SOBJECTIZER=Off -DRESTINIO_USE_BOOST_ASIO=none . \
@@ -132,7 +132,7 @@ struct OPENDHT_PUBLIC Config {
    /* If non-0, overrides the default maximum store size. -1 means no limit.  */
    ssize_t max_store_size {0};

-    /** 
+    /**
     * Use appropriate bahavior for a public IP, stable node:
     *   - No connectivity change triggered when a search fails
     *   - Larger listen refresh time
@@ -148,7 +148,7 @@ struct OPENDHT_PUBLIC SecureDhtConfig
    Config node_config {};
    crypto::Identity id {};

-    /** 
+    /**
     * Cache all encountered public keys and certificates,
     * for use by the certificate store, putEncrypted and putSigned
     */
@@ -164,6 +164,7 @@ using GetCallback = std::function<bool(const std::vector<std::shared_ptr<Value>>
 using ValueCallback = std::function<bool(const std::vector<std::shared_ptr<Value>>& values, bool expired)>;
 using GetCallbackSimple = std::function<bool(std::shared_ptr<Value> value)>;
 using ShutdownCallback = std::function<void()>;
+using IdentityAnnouncedCb = std::function<void(bool)>;

 using CertificateStoreQuery = std::function<std::vector<std::shared_ptr<crypto::Certificate>>(const InfoHash& pk_id)>;

@@ -178,13 +179,13 @@ typedef bool (*FilterRaw)(const Value&, void *user_data);

 using DoneCallbackSimple = std::function<void(bool success)>;

-OPENDHT_PUBLIC GetCallbackSimple bindGetCb(const GetCallbackRaw& raw_cb, void* user_data);
-OPENDHT_PUBLIC GetCallback bindGetCb(const GetCallbackSimple& cb);
-OPENDHT_PUBLIC ValueCallback bindValueCb(const ValueCallbackRaw& raw_cb, void* user_data);
-OPENDHT_PUBLIC ShutdownCallback bindShutdownCb(const ShutdownCallbackRaw& shutdown_cb_raw, void* user_data);
+OPENDHT_PUBLIC GetCallbackSimple bindGetCb(GetCallbackRaw raw_cb, void* user_data);
+OPENDHT_PUBLIC GetCallback bindGetCb(GetCallbackSimple cb);
+OPENDHT_PUBLIC ValueCallback bindValueCb(ValueCallbackRaw raw_cb, void* user_data);
+OPENDHT_PUBLIC ShutdownCallback bindShutdownCb(ShutdownCallbackRaw shutdown_cb_raw, void* user_data);
 OPENDHT_PUBLIC DoneCallback bindDoneCb(DoneCallbackSimple donecb);
-OPENDHT_PUBLIC DoneCallback bindDoneCb(const DoneCallbackRaw& raw_cb, void* user_data);
-OPENDHT_PUBLIC DoneCallbackSimple bindDoneCbSimple(const DoneCallbackSimpleRaw& raw_cb, void* user_data);
-OPENDHT_PUBLIC Value::Filter bindFilterRaw(const FilterRaw& raw_filter, void* user_data);
+OPENDHT_PUBLIC DoneCallback bindDoneCb(DoneCallbackRaw raw_cb, void* user_data);
+OPENDHT_PUBLIC DoneCallbackSimple bindDoneCbSimple(DoneCallbackSimpleRaw raw_cb, void* user_data);
+OPENDHT_PUBLIC Value::Filter bindFilterRaw(FilterRaw raw_filter, void* user_data);

 }
@@ -80,6 +80,8 @@ struct OPENDHT_PUBLIC PublicKey
     * Takes ownership of an existing gnutls_pubkey.
     */
    PublicKey(gnutls_pubkey_t k) : pk(k) {}
+
+    /** Import public key from serialized data */
    PublicKey(const uint8_t* dat, size_t dat_size);
    PublicKey(const Blob& pk) : PublicKey(pk.data(), pk.size()) {}
    PublicKey(PublicKey&& o) noexcept : pk(o.pk) { o.pk = nullptr; };
@@ -106,12 +108,12 @@ struct OPENDHT_PUBLIC PublicKey
    PkId getLongId() const;

    bool checkSignature(const uint8_t* data, size_t data_len, const uint8_t* signature, size_t signature_len) const;
-    bool checkSignature(const Blob& data, const Blob& signature) const {
+    inline bool checkSignature(const Blob& data, const Blob& signature) const {
        return checkSignature(data.data(), data.size(), signature.data(), signature.size());
    }

    Blob encrypt(const uint8_t* data, size_t data_len) const;
-    Blob encrypt(const Blob& data) const {
+    inline Blob encrypt(const Blob& data) const {
        return encrypt(data.data(), data.size());
    }

@@ -170,14 +172,16 @@ struct OPENDHT_PUBLIC PrivateKey
     * Sign the provided binary object.
     * @returns the signature data.
     */
-    Blob sign(const Blob&) const;
+    Blob sign(const uint8_t* data, size_t data_len) const;
+    inline Blob sign(const Blob& dat) const { return sign(dat.data(), dat.size()); }

    /**
     * Try to decrypt the provided cypher text.
     * In case of failure a CryptoException is thrown.
     * @returns the decrypted data.
     */
-    Blob decrypt(const Blob& cypher) const;
+    Blob decrypt(const uint8_t* cypher, size_t cypher_len) const;
+    Blob decrypt(const Blob& cypher) const { return decrypt(cypher.data(), cypher.size()); }

    /**
     * Generate a new RSA key pair
@@ -310,6 +314,24 @@ private:
    gnutls_x509_crq_t request {nullptr};
 };

+class OPENDHT_PUBLIC OcspRequest
+{
+public:
+    OcspRequest(gnutls_ocsp_req_t r) : request(r) {};
+    OcspRequest(const uint8_t* dat_ptr, size_t dat_size);
+    ~OcspRequest();
+
+    /*
+     * Get OCSP Request in readable format.
+     */
+    std::string toString(const bool compact = true) const;
+
+    Blob pack() const;
+    Blob getNonce() const;
+private:
+    gnutls_ocsp_req_t request;
+};
+
 class OPENDHT_PUBLIC OcspResponse
 {
 public:
@@ -331,11 +353,11 @@ public:
    gnutls_ocsp_cert_status_t getCertificateStatus() const;

    /*
-     * Verify OCSP response.
-     * Return OCSP verify reason.
+     * Verify OCSP response and return OCSP status.
+     * Throws CryptoException in case of error in the response.
     * http://www.gnu.org/software/gnutls/reference/gnutls-ocsp.html#gnutls-ocsp-verify-reason-t
     */
-    gnutls_ocsp_verify_reason_t verifyDirect(const Certificate& crt, const Blob& nonce);
+    gnutls_ocsp_cert_status_t verifyDirect(const Certificate& crt, const Blob& nonce);

 private:
    gnutls_ocsp_resp_t response;
@@ -749,8 +771,10 @@ OPENDHT_PUBLIC Blob aesEncrypt(const Blob& data, const std::string& password);
 /**
 * AES-GCM decryption.
 */
-OPENDHT_PUBLIC Blob aesDecrypt(const Blob& data, const Blob& key);
-OPENDHT_PUBLIC Blob aesDecrypt(const Blob& data, const std::string& password);
+OPENDHT_PUBLIC Blob aesDecrypt(const uint8_t* data, size_t data_length, const Blob& key);
+OPENDHT_PUBLIC inline Blob aesDecrypt(const Blob& data, const Blob& key) { return aesDecrypt(data.data(), data.size(), key); }
+OPENDHT_PUBLIC Blob aesDecrypt(const uint8_t* data, size_t data_length, const std::string& password);
+OPENDHT_PUBLIC inline Blob aesDecrypt(const Blob& data, const std::string& password) { return aesDecrypt(data.data(), data.size(), password); }

 }
 }
@@ -144,17 +144,18 @@ public:
    static const ValueType TYPE;

    TrustRequest() {}
-    TrustRequest(std::string s) : service(s) {}
-    TrustRequest(std::string s, const Blob& d) : service(s), payload(d) {}
+    TrustRequest(std::string s, std::string ci = {}) : service(s), conversationId(ci) {}
+    TrustRequest(std::string s, std::string ci, const Blob& d) : service(s), conversationId(ci), payload(d) {}

    static Value::Filter getFilter() {
        return EncryptedValue::getFilter();
    }

    std::string service;
+    std::string conversationId;
    Blob payload;
    bool confirm {false};
-    MSGPACK_DEFINE_MAP(service, payload, confirm)
+    MSGPACK_DEFINE_MAP(service, conversationId, payload, confirm)
 };

 class OPENDHT_PUBLIC IceCandidates : public EncryptedValue<IceCandidates>
@@ -61,9 +61,6 @@ struct LocalListener;
 */
 class OPENDHT_PUBLIC Dht final : public DhtInterface {
 public:
-
-    Dht();
-
    /**
     * Initialise the Dht with two open sockets (for IPv4 and IP6)
     * and an ID for the node.
@@ -98,7 +95,7 @@ public:
    /**
     * Performs final operations before quitting.
     */
-    void shutdown(ShutdownCallback cb) override;
+    void shutdown(ShutdownCallback cb, bool stop = false) override;

    /**
     * Returns true if the node is running (have access to an open socket).
@@ -515,6 +512,8 @@ private:

    void confirmNodes();
    void expire();
+
+    void onConnected();
    void onDisconnected();

    /**
@@ -21,6 +21,8 @@
 #include "infohash.h"
 #include "log_enable.h"

+#include <queue>
+
 namespace dht {

 namespace net {
@@ -46,6 +48,10 @@ public:
    virtual NodeStatus getStatus(sa_family_t af) const = 0;
    virtual NodeStatus getStatus() const = 0;

+    void addOnConnectedCallback(std::function<void()> cb) {
+        onConnectCallbacks_.emplace(std::move(cb));
+    }
+
    virtual net::DatagramSocket* getSocket() const { return {}; };

    /**
@@ -55,8 +61,10 @@ public:

    /**
     * Performs final operations before quitting.
+     * stop: if true, cancel ongoing operations and call their 'done'
+     *       callbacks synchronously.
     */
-    virtual void shutdown(ShutdownCallback cb) = 0;
+    virtual void shutdown(ShutdownCallback cb, bool stop = false) = 0;

    /**
     * Returns true if the node is running (have access to an open socket).
@@ -264,6 +272,7 @@ public:

 protected:
    std::shared_ptr<Logger> logger_ {};
+    std::queue<std::function<void()>> onConnectCallbacks_ {};
 };

 } // namespace dht
@@ -85,7 +85,7 @@ public:
    /**
     * Performs final operations before quitting.
     */
-    void shutdown(ShutdownCallback cb) override;
+    void shutdown(ShutdownCallback cb, bool) override;

    /**
     * Returns true if the node is running (have access to an open socket).
@@ -107,13 +107,13 @@ public:
     */
    virtual void get(const InfoHash& key, GetCallback cb, DoneCallback donecb={}, Value::Filter&& f={}, Where&& w = {}) override;
    virtual void get(const InfoHash& key, GetCallback cb, DoneCallbackSimple donecb={}, Value::Filter&& f={}, Where&& w = {}) override {
-        get(key, cb, bindDoneCb(donecb), std::forward<Value::Filter>(f), std::forward<Where>(w));
+        get(key, cb, bindDoneCb(std::move(donecb)), std::forward<Value::Filter>(f), std::forward<Where>(w));
    }
    virtual void get(const InfoHash& key, GetCallbackSimple cb, DoneCallback donecb={}, Value::Filter&& f={}, Where&& w = {}) override {
-        get(key, bindGetCb(cb), donecb, std::forward<Value::Filter>(f), std::forward<Where>(w));
+        get(key, bindGetCb(cb), std::move(donecb), std::forward<Value::Filter>(f), std::forward<Where>(w));
    }
    virtual void get(const InfoHash& key, GetCallbackSimple cb, DoneCallbackSimple donecb, Value::Filter&& f={}, Where&& w = {}) override {
-        get(key, bindGetCb(cb), bindDoneCb(donecb), std::forward<Value::Filter>(f), std::forward<Where>(w));
+        get(key, bindGetCb(cb), bindDoneCb(std::move(donecb)), std::forward<Value::Filter>(f), std::forward<Where>(w));
    }

    /**
@@ -134,7 +134,7 @@ public:
            time_point created=time_point::max(),
            bool permanent = false) override
    {
-        put(key, v, bindDoneCb(cb), created, permanent);
+        put(key, v, bindDoneCb(std::move(cb)), created, permanent);
    }

    void put(const InfoHash& key,
@@ -143,7 +143,7 @@ public:
            time_point created=time_point::max(),
            bool permanent = false) override
    {
-        put(key, std::make_shared<Value>(std::move(v)), cb, created, permanent);
+        put(key, std::make_shared<Value>(std::move(v)), std::move(cb), created, permanent);
    }
    void put(const InfoHash& key,
            Value&& v,
@@ -151,7 +151,7 @@ public:
            time_point created=time_point::max(),
            bool permanent = false) override
    {
-        put(key, std::forward<Value>(v), bindDoneCb(cb), created, permanent);
+        put(key, std::forward<Value>(v), bindDoneCb(std::move(cb)), created, permanent);
    }

    /**
@@ -176,14 +176,14 @@ public:
    virtual size_t listen(const InfoHash&, ValueCallback, Value::Filter={}, Where={}) override;

    virtual size_t listen(const InfoHash& key, GetCallback cb, Value::Filter f={}, Where w={}) override {
-        return listen(key, [cb](const std::vector<Sp<Value>>& vals, bool expired){
+        return listen(key, [cb=std::move(cb)](const std::vector<Sp<Value>>& vals, bool expired){
            if (not expired)
                return cb(vals);
            return true;
        }, std::forward<Value::Filter>(f), std::forward<Where>(w));
    }
    virtual size_t listen(const InfoHash& key, GetCallbackSimple cb, Value::Filter f={}, Where w={}) override {
-        return listen(key, bindGetCb(cb), std::forward<Value::Filter>(f), std::forward<Where>(w));
+        return listen(key, bindGetCb(std::move(cb)), std::forward<Value::Filter>(f), std::forward<Where>(w));
    }
    /*
     * This function relies on the cache implementation.
@@ -214,7 +214,7 @@ public:
     */
    virtual void query(const InfoHash& /*key*/, QueryCallback /*cb*/, DoneCallback /*done_cb*/ = {}, Query&& /*q*/ = {}) override { }
    virtual void query(const InfoHash& key, QueryCallback cb, DoneCallbackSimple done_cb = {}, Query&& q = {}) override {
-        query(key, cb, bindDoneCb(done_cb), std::forward<Query>(q));
+        query(key, cb, bindDoneCb(std::move(done_cb)), std::forward<Query>(q));
    }

    /**
@@ -378,8 +378,8 @@ private:
     * Retrieve if we can connect to the proxy (update statusIpvX_)
     */
    void handleProxyConfirm(const asio::error_code &ec);
-    Sp<asio::steady_timer> nextProxyConfirmationTimer_;
-    Sp<asio::steady_timer> listenerRestartTimer_;
+    std::unique_ptr<asio::steady_timer> nextProxyConfirmationTimer_;
+    std::unique_ptr<asio::steady_timer> listenerRestartTimer_;

    /**
     * Relaunch LISTEN requests if the client disconnect/reconnect.
@@ -445,7 +445,6 @@ private:
        std::map<InfoHash, std::vector<Listener>> listeners;
        MSGPACK_DEFINE_ARRAY(listeners)
    };
-    std::mutex lockPushListeners_;
    std::map<std::string, PushListener> pushListeners_;
    proxy::ListenToken tokenPushNotif_ {0};
 #endif //OPENDHT_PUSH_NOTIFICATIONS
@@ -65,6 +65,7 @@ public:
        bool peer_publish {false};
        std::shared_ptr<dht::crypto::Certificate> server_ca;
        dht::crypto::Identity client_identity;
+        SockAddr bind4 {}, bind6 {};
    };

    struct Context {
@@ -73,6 +74,7 @@ public:
        std::shared_ptr<PeerDiscovery> peerDiscovery {};
        StatusCallback statusChangedCallback {};
        CertificateStoreQuery certificateStore {};
+        IdentityAnnouncedCb identityAnnouncedCb {};
        Context() {}
    };

@@ -97,7 +99,7 @@ public:
    template <class T>
    void get(InfoHash hash, std::function<bool(std::vector<T>&&)> cb, DoneCallbackSimple dcb={})
    {
-        get(hash, [=](const std::vector<std::shared_ptr<Value>>& vals) {
+        get(hash, [cb=std::move(cb)](const std::vector<std::shared_ptr<Value>>& vals) {
            return cb(unpackVector<T>(vals));
        },
        dcb,
@@ -106,7 +108,7 @@ public:
    template <class T>
    void get(InfoHash hash, std::function<bool(T&&)> cb, DoneCallbackSimple dcb={})
    {
-        get(hash, [=](const std::vector<std::shared_ptr<Value>>& vals) {
+        get(hash, [cb=std::move(cb)](const std::vector<std::shared_ptr<Value>>& vals) {
            for (const auto& v : vals) {
                try {
                    if (not cb(Value::unpack<T>(*v)))
@@ -155,7 +157,7 @@ public:
    std::future<size_t> listen(InfoHash key, ValueCallback vcb, Value::Filter f = {}, Where w = {});

    std::future<size_t> listen(InfoHash key, GetCallback cb, Value::Filter f={}, Where w={}) {
-        return listen(key, [cb](const std::vector<Sp<Value>>& vals, bool expired){
+        return listen(key, [cb=std::move(cb)](const std::vector<Sp<Value>>& vals, bool expired){
            if (not expired)
                return cb(vals);
            return true;
@@ -169,7 +171,7 @@ public:
    template <class T>
    std::future<size_t> listen(InfoHash hash, std::function<bool(std::vector<T>&&)> cb)
    {
-        return listen(hash, [=](const std::vector<std::shared_ptr<Value>>& vals) {
+        return listen(hash, [cb=std::move(cb)](const std::vector<std::shared_ptr<Value>>& vals) {
            return cb(unpackVector<T>(vals));
        },
        getFilterSet<T>());
@@ -177,7 +179,7 @@ public:
    template <class T>
    std::future<size_t> listen(InfoHash hash, std::function<bool(std::vector<T>&&, bool)> cb)
    {
-        return listen(hash, [=](const std::vector<std::shared_ptr<Value>>& vals, bool expired) {
+        return listen(hash, [cb=std::move(cb)](const std::vector<std::shared_ptr<Value>>& vals, bool expired) {
            return cb(unpackVector<T>(vals), expired);
        },
        getFilterSet<T>());
@@ -186,7 +188,7 @@ public:
    template <typename T>
    std::future<size_t> listen(InfoHash hash, std::function<bool(T&&)> cb, Value::Filter f = {}, Where w = {})
    {
-        return listen(hash, [=](const std::vector<std::shared_ptr<Value>>& vals) {
+        return listen(hash, [cb=std::move(cb)](const std::vector<std::shared_ptr<Value>>& vals) {
            for (const auto& v : vals) {
                try {
                    if (not cb(Value::unpack<T>(*v)))
@@ -202,7 +204,7 @@ public:
    template <typename T>
    std::future<size_t> listen(InfoHash hash, std::function<bool(T&&, bool)> cb, Value::Filter f = {}, Where w = {})
    {
-        return listen(hash, [=](const std::vector<std::shared_ptr<Value>>& vals, bool expired) {
+        return listen(hash, [cb=std::move(cb)](const std::vector<std::shared_ptr<Value>>& vals, bool expired) {
            for (const auto& v : vals) {
                try {
                    if (not cb(Value::unpack<T>(*v), expired))
@@ -359,6 +361,7 @@ public:
    std::string getSearchLog(const InfoHash&, sa_family_t af = AF_UNSPEC) const;
    std::vector<SockAddr> getPublicAddress(sa_family_t af = AF_UNSPEC);
    std::vector<std::string> getPublicAddressStr(sa_family_t af = AF_UNSPEC);
+    void getPublicAddress(std::function<void(std::vector<SockAddr>&&)>, sa_family_t af = AF_UNSPEC);

    // securedht methods

@@ -379,22 +382,12 @@ public:
        config.threaded = threaded;
        run(port, config);
    }
-    void run(in_port_t port, const Config& config, Context&& context = {});
-
-    /**
-     * @param local4: Local IPv4 address and port to bind. Can be null.
-     * @param local6: Local IPv6 address and port to bind. Can be null.
-     *         You should allways bind to a global IPv6 address.
-     * @param identity: RSA key pair to use for cryptographic operations.
-     * @param threaded: If false, loop() must be called periodically. Otherwise a thread is launched.
-     * @param cb: Optional callback to receive general state information.
-     */
-    void run(SockAddr& local4, SockAddr& local6, const Config& config, Context&& context = {});
+    void run(in_port_t port, Config& config, Context&& context = {});

    /**
     * Same as @run(sockaddr_in, sockaddr_in6, Identity, bool, StatusCallback), but with string IP addresses and service (port).
     */
-    void run(const char* ip4, const char* ip6, const char* service, const Config& config, Context&& context = {});
+    void run(const char* ip4, const char* ip6, const char* service, Config& config, Context&& context = {});

    void run(const Config& config, Context&& context);

@@ -415,7 +408,7 @@ public:
    /**
     * Gracefuly disconnect from network.
     */
-    void shutdown(ShutdownCallback cb = {});
+    void shutdown(ShutdownCallback cb = {}, bool stop = false);

    /**
     * Quit and wait for all threads to terminate.
@@ -481,6 +474,7 @@ private:

    /** Current configuration */
    Config config_;
+    IdentityAnnouncedCb identityAnnouncedCb_;

    /**
     * reset dht clients
@@ -335,7 +335,7 @@ public:
    void lookup(Key k, LookupCallback cb = {}, DoneCallbackSimple done_cb = {}, bool exact_match = true);
    void lookup(Key k, LookupCallbackSimple cb = {}, DoneCallbackSimple done_cb = {}, bool exact_match = true)
    {
-        lookup(k, [=](std::vector<std::shared_ptr<Value>>& values, Prefix) { cb(values); }, done_cb, exact_match);
+        lookup(k, [cb=std::move(cb)](std::vector<std::shared_ptr<Value>>& values, Prefix) { cb(values); }, done_cb, exact_match);
    }

    /**
@@ -54,9 +54,9 @@ namespace crypto {
 }

 /**
- * Represents an InfoHash.
- * An InfoHash is a byte array of HASH_LEN bytes.
- * InfoHashes identify nodes and values in the Dht.
+ * Represents an Hash,
+ * a byte array of N bytes.
+ * Hashes identify nodes and values in the Dht.
 */
 template <size_t N>
 class OPENDHT_PUBLIC Hash {
@@ -93,6 +93,8 @@ public:
    iterator end() { return data_.end(); }
    const_iterator cend() const { return data_.cend(); }

+    static constexpr inline Hash zero() noexcept { return Hash{}; }
+
    bool operator==(const Hash& h) const {
        auto a = reinterpret_cast<const uint32_t*>(data_.data());
        auto b = reinterpret_cast<const uint32_t*>(h.data_.data());
@@ -112,6 +114,14 @@ public:
        return false;
    }

+    Hash operator^(const Hash& o) const {
+        Hash result;
+        for(auto i = 0u; i < N; i++) {
+            result[i] = data_[i] ^ o.data_[i];
+        }
+        return result;
+    }
+
    explicit operator bool() const {
        auto a = reinterpret_cast<const uint32_t*>(data_.data());
        auto b = reinterpret_cast<const uint32_t*>(data_.data() + N);
@@ -142,10 +152,6 @@ public:
        return 8 * i + j;
    }

-    /**
-     * Forget about the ``XOR-metric''.  An id is just a path from the
-     * root of the tree, so bits are numbered from the start.
-     */
    static inline int cmp(const Hash& id1, const Hash& id2) {
        return std::memcmp(id1.data_.data(), id2.data_.data(), N);
    }
@@ -179,16 +185,12 @@ public:
    int
    xorCmp(const Hash& id1, const Hash& id2) const
    {
-        for(unsigned i = 0; i < N; i++) {
-            uint8_t xor1, xor2;
+        for (unsigned i = 0; i < N; i++) {
            if(id1.data_[i] == id2.data_[i])
                continue;
-            xor1 = id1.data_[i] ^ data_[i];
-            xor2 = id2.data_[i] ^ data_[i];
-            if(xor1 < xor2)
-                return -1;
-            else
-                return 1;
+            uint8_t xor1 = id1.data_[i] ^ data_[i];
+            uint8_t xor2 = id2.data_[i] ^ data_[i];
+            return (xor1 < xor2) ? -1 : 1;
        }
        return 0;
    }
@@ -390,8 +392,6 @@ Hash<N>::toString() const
    return std::string(to_c_str(), N*2);
 }

-const InfoHash zeroes {};
-
 struct OPENDHT_PUBLIC NodeExport {
    InfoHash id;
    sockaddr_storage ss;
@@ -214,7 +214,6 @@ public:
    using RequestErrorCb = std::function<bool(const Request&, DhtProtocolException&&)>;
    using RequestExpiredCb = std::function<void(const Request&, bool)>;

-    NetworkEngine(const Sp<Logger>& log, std::mt19937_64& rd, Scheduler& scheduler, std::unique_ptr<DatagramSocket>&& sock);
    NetworkEngine(
            InfoHash& myid,
            NetworkConfig config,
@@ -293,7 +292,7 @@ public:
     */
    Sp<Request>
    sendPing(SockAddr&& sa, RequestCb&& on_done, RequestExpiredCb&& on_expired) {
-        return sendPing(std::make_shared<Node>(zeroes, std::move(sa), rd),
+        return sendPing(std::make_shared<Node>(InfoHash::zero(), std::move(sa), rd),
                std::forward<RequestCb>(on_done),
                std::forward<RequestExpiredCb>(on_expired));
    }
@@ -45,15 +45,13 @@ public:
        return c;
    }

-    SecureDht() {}
-
    /**
     * s, s6: bound socket descriptors for IPv4 and IPv6, respectively.
     *        For the Dht to be initialised, at least one of them must be >= 0.
     * id:    the identity to use for the crypto layer and to compute
     *        our own hash on the Dht.
     */
-    SecureDht(std::unique_ptr<DhtInterface> dht, Config config);
+    SecureDht(std::unique_ptr<DhtInterface> dht, Config config, IdentityAnnouncedCb iacb = {}, const std::shared_ptr<Logger>& l = {});

    virtual ~SecureDht();

@@ -148,8 +146,8 @@ public:
    /**
     * SecureDht to Dht proxy
     */
-    void shutdown(ShutdownCallback cb) override {
-        dht_->shutdown(cb);
+    void shutdown(ShutdownCallback cb, bool stop = false) override {
+        dht_->shutdown(cb, stop);
    }
    void dumpTables() const override {
        dht_->dumpTables();
@@ -68,7 +68,7 @@ public:
     * Build from existing address.
     */
    SockAddr(const sockaddr* sa, socklen_t length) {
-        if (length > sizeof(sockaddr_storage))
+        if (length > static_cast<socklen_t>(sizeof(sockaddr_storage)))
            throw std::runtime_error("Socket address length is too large");
        set(sa, length);
    }
@@ -144,7 +144,7 @@ public:
            if (len) addr.reset((sockaddr*)::calloc(len, 1));
            else     addr.reset();
        }
-        if (len > sizeof(sa_family_t))
+        if (len)
            addr->sa_family = af;
    }

@@ -223,16 +223,16 @@ public:
     */
    sockaddr* get() { return addr.get(); }

-    const sockaddr_in& getIPv4() const {
+    inline const sockaddr_in& getIPv4() const {
        return *reinterpret_cast<const sockaddr_in*>(get());
    }
-    const sockaddr_in6& getIPv6() const {
+    inline const sockaddr_in6& getIPv6() const {
        return *reinterpret_cast<const sockaddr_in6*>(get());
    }
-    sockaddr_in& getIPv4() {
+    inline sockaddr_in& getIPv4() {
        return *reinterpret_cast<sockaddr_in*>(get());
    }
-    sockaddr_in6& getIPv6() {
+    inline sockaddr_in6& getIPv6() {
        return *reinterpret_cast<sockaddr_in6*>(get());
    }

@@ -93,4 +93,71 @@ private:
    void schedule();
 };

+class OPENDHT_PUBLIC ExecutionContext {
+public:
+    ExecutionContext(ThreadPool& pool)
+     : threadPool_(pool), state_(std::make_shared<SharedState>())
+    {}
+
+    ~ExecutionContext() {
+        state_->destroy();
+    }
+
+    /** Wait for ongoing tasks to complete execution and drop other pending tasks */
+    void stop() {
+        state_->destroy(false);
+    }
+
+    void run(std::function<void()>&& task) {
+        std::lock_guard<std::mutex> lock(state_->mtx);
+        if (state_->shutdown_) return;
+        state_->pendingTasks++;
+        threadPool_.get().run([task = std::move(task), state = state_] {
+            state->run(task);
+        });
+    }
+
+private:
+    struct SharedState {
+        std::mutex mtx {};
+        std::condition_variable cv {};
+        unsigned pendingTasks {0};
+        unsigned ongoingTasks {0};
+        /** When true, prevents new tasks to be scheduled */
+        bool shutdown_ {false};
+        /** When true, prevents scheduled tasks to be executed */
+        std::atomic_bool destroyed {false};
+
+        void destroy(bool wait = true) {
+            std::unique_lock<std::mutex> lock(mtx);
+            if (destroyed) return;
+            if (wait) {
+                cv.wait(lock, [this] { return pendingTasks == 0 && ongoingTasks == 0; });
+            }
+            shutdown_ = true;
+            if (not wait) {
+                cv.wait(lock, [this] { return ongoingTasks == 0; });
+            }
+            destroyed = true;
+        }
+
+        void run(const std::function<void()>& task) {
+            {
+                std::lock_guard<std::mutex> lock(mtx);
+                pendingTasks--;
+                ongoingTasks++;
+            }
+                if (destroyed) return;
+                task();
+            {
+                std::lock_guard<std::mutex> lock(mtx);
+                ongoingTasks--;
+                cv.notify_all();
+            }
+        }
+    };
+    std::reference_wrapper<ThreadPool> threadPool_;
+    std::shared_ptr<SharedState> state_;
+};
+
 }
@@ -205,7 +205,7 @@ struct OPENDHT_PUBLIC Value
        }
        static Filter chainOr(Filter&& f1, Filter&& f2) {
            if (not f1 or not f2) return {};
-            return [f1,f2](const Value& v) {
+            return [f1=std::move(f1),f2=std::move(f2)](const Value& v) {
                return f1(v) or f2(v);
            };
        }
@@ -231,8 +231,7 @@ struct OPENDHT_PUBLIC Value
    }

    static Filter TypeFilter(const ValueType& t) {
-        const auto tid = t.id;
-        return [tid](const Value& v) {
+        return [tid = t.id](const Value& v) {
            return v.type == tid;
        };
    }
@@ -36,6 +36,7 @@ from libcpp.memory cimport shared_ptr
 from cython.parallel import parallel, prange
 from cython.operator cimport dereference as deref, preincrement as inc, predecrement as dec
 from cpython cimport ref
+from datetime import timedelta

 cimport opendht_cpp as cpp

@@ -235,8 +236,9 @@ cdef class Where(object):

 cdef class Value(object):
    cdef shared_ptr[cpp.Value] _value
-    def __init__(self, bytes val=b''):
-        self._value.reset(new cpp.Value(val, len(val)))
+    def __init__(self, bytes val=b'', cpp.uint16_t id=0):
+        self._value.reset(new cpp.Value(id, val, len(val)))
+
    def __str__(self):
        return self._value.get().toString().decode()
    property owner:
@@ -270,6 +272,14 @@ cdef class Value(object):
        def __get__(self):
            return self._value.get().size()

+cdef class ValueType(object):
+    cdef cpp.ValueType * _value
+    def __init__(self, cpp.uint16_t id, str name, expiration: timedelta):
+        if not isinstance(expiration, timedelta):
+            raise TypeError("expiration argument must be of type timedelta")
+        cdef cpp.seconds duration = cpp.seconds(int(expiration.total_seconds()))
+        self._value =  new cpp.ValueType(id, name.encode(), duration)
+
 cdef class NodeSetIter(object):
    cdef map[cpp.InfoHash, shared_ptr[cpp.Node]]* _nodes
    cdef map[cpp.InfoHash, shared_ptr[cpp.Node]].iterator _curIter
@@ -513,6 +523,8 @@ cdef class DhtRunner(_WithID):
        cb_obj = {'shutdown':shutdown_cb}
        ref.Py_INCREF(cb_obj)
        self.thisptr.get().shutdown(cpp.bindShutdownCb(shutdown_callback, <void*>cb_obj))
+    def registerType(self, ValueType type):
+        self.thisptr.get().registerType(deref(type._value))
    def enableLogging(self):
        cpp.enableLogging(self.thisptr.get()[0])
    def disableLogging(self):
@@ -26,6 +26,14 @@ from libc.string cimport const_char, const_uchar
 ctypedef uint16_t in_port_t
 ctypedef unsigned short int sa_family_t;

+cdef extern from "<iostream>" namespace "std::chrono" nogil:
+    cdef cppclass duration[ulong]:
+        duration() except +
+
+    cdef cppclass seconds:
+        duration seconds(uint64_t) except +
+        duration seconds()
+
 cdef extern from "<memory>" namespace "std" nogil:
    cdef cppclass shared_ptr[T]:
        shared_ptr() except +
@@ -131,7 +139,6 @@ cdef extern from "opendht/value.h" namespace "dht::Value":
 cdef extern from "opendht/value.h" namespace "dht::Value::Field":
    cdef Field None
    cdef Field Id
-    cdef Field ValueType
    cdef Field OwnerPk
    cdef Field SeqNum
    cdef Field UserType
@@ -141,7 +148,7 @@ cdef extern from "opendht/value.h" namespace "dht":
    cdef cppclass Value:
        Value() except +
        Value(vector[uint8_t]) except +
-        Value(const uint8_t* dat_ptr, size_t dat_len) except +
+        Value(const uint16_t t, const uint8_t* dat_ptr, size_t dat_len) except +
        string toString() const
        size_t size() const
        uint64_t id
@@ -150,6 +157,12 @@ cdef extern from "opendht/value.h" namespace "dht":
        vector[uint8_t] data
        string user_type

+    cdef cppclass ValueType:
+        ValueType(uint16_t id, string name, seconds expiration) except +
+        uint16_t id
+        string name
+        seconds expiration
+
    cdef cppclass Query:
        Query() except +
        Query(Select s, Where w) except +
@@ -234,6 +247,7 @@ cdef extern from "opendht/dhtrunner.h" namespace "dht":
        void run(const_char*, const_char*, const_char*, Config config)
        void join()
        void shutdown(ShutdownCallback)
+        void registerType(ValueType& value);
        bool isRunning()
        SockAddr getBound(sa_family_t af) const
        string getStorageLog() const
@@ -26,7 +26,7 @@ namespace dht {


 GetCallbackSimple
-bindGetCb(const GetCallbackRaw& raw_cb, void* user_data)
+bindGetCb(GetCallbackRaw raw_cb, void* user_data)
 {
    if (not raw_cb) return {};
    return [=](const std::shared_ptr<Value>& value) {
@@ -35,10 +35,10 @@ bindGetCb(const GetCallbackRaw& raw_cb, void* user_data)
 }

 GetCallback
-bindGetCb(const GetCallbackSimple& cb)
+bindGetCb(GetCallbackSimple cb)
 {
    if (not cb) return {};
-    return [=](const std::vector<std::shared_ptr<Value>>& values) {
+    return [cb=std::move(cb)](const std::vector<std::shared_ptr<Value>>& values) {
        for (const auto& v : values)
            if (not cb(v))
                return false;
@@ -47,7 +47,7 @@ bindGetCb(const GetCallbackSimple& cb)
 }

 ValueCallback
-bindValueCb(const ValueCallbackRaw& raw_cb, void* user_data)
+bindValueCb(ValueCallbackRaw raw_cb, void* user_data)
 {
    if (not raw_cb) return {};
    return [=](const std::vector<std::shared_ptr<Value>>& values, bool expired) {
@@ -59,7 +59,7 @@ bindValueCb(const ValueCallbackRaw& raw_cb, void* user_data)
 }

 ShutdownCallback
-bindShutdownCb(const ShutdownCallbackRaw& shutdown_cb_raw, void* user_data)
+bindShutdownCb(ShutdownCallbackRaw shutdown_cb_raw, void* user_data)
 {
    return [=]() { shutdown_cb_raw(user_data); };
 }
@@ -69,11 +69,11 @@ bindDoneCb(DoneCallbackSimple donecb)
 {
    if (not donecb) return {};
    using namespace std::placeholders;
-    return std::bind(donecb, _1);
+    return std::bind(std::move(donecb), _1);
 }

 DoneCallback
-bindDoneCb(const DoneCallbackRaw& raw_cb, void* user_data)
+bindDoneCb(DoneCallbackRaw raw_cb, void* user_data)
 {
    if (not raw_cb) return {};
    return [=](bool success, const std::vector<std::shared_ptr<Node>>& nodes) {
@@ -82,7 +82,7 @@ bindDoneCb(const DoneCallbackRaw& raw_cb, void* user_data)
 }

 DoneCallbackSimple
-bindDoneCbSimple(const DoneCallbackSimpleRaw& raw_cb, void* user_data) {
+bindDoneCbSimple(DoneCallbackSimpleRaw raw_cb, void* user_data) {
    if (not raw_cb) return {};
    return [=](bool success) {
        raw_cb(success, user_data);
@@ -189,19 +189,15 @@ void
 PEMCache::fillX509Store(SSL_CTX* ctx)
 {
    if (logger)
-        logger->w("adding %d decoded certs to X509 store", pems_.size());
-    X509_STORE* store = SSL_CTX_get_cert_store(ctx);
-    if (store == nullptr) {
-        if (logger)
-            logger->e("couldn't get the context cert store");
-        return;
-    }
-    for (const auto& pem : pems_) {
-        if (X509_STORE_add_cert(store, pem.get()) == 1)
-            continue;
-        if (logger)
-            logger->d("couldn't add local certificate");
-    }
+        logger->d("adding %d decoded certs to X509 store", pems_.size());
+    if (X509_STORE* store = SSL_CTX_get_cert_store(ctx)) {
+        for (const auto& pem : pems_) {
+            if (X509_STORE_add_cert(store, pem.get()) != 1)
+                if (logger)
+                    logger->w("couldn't add local certificate");
+        }
+    } else if (logger)
+        logger->e("couldn't get the context cert store");
 }

 } /*namespace http*/
@@ -54,7 +54,7 @@ static std::uniform_int_distribution<uint8_t> rand_byte;
 #define GNUTLS_PKCS_PBES2_AES_256 GNUTLS_PKCS_USE_PBES2_AES_256
 #endif

-#define DHT_AES_LEGACY_ENCRYPT 1
+#define DHT_AES_LEGACY_ENCRYPT 0
 #define DHT_AES_LEGACY_DECRYPT 1

 namespace dht {
@@ -128,37 +128,37 @@ Blob aesEncrypt(const Blob& data, const std::string& password)
    return encrypted;
 }

-Blob aesDecrypt(const Blob& data, const Blob& key)
+Blob aesDecrypt(const uint8_t* data, size_t data_length, const Blob& key)
 {
    if (not aesKeySizeGood(key.size()))
        throw DecryptError("Wrong key size");

-    if (data.size() <= GCM_IV_SIZE + GCM_DIGEST_SIZE)
+    if (data_length <= GCM_IV_SIZE + GCM_DIGEST_SIZE)
        throw DecryptError("Wrong data size");

    std::array<uint8_t, GCM_DIGEST_SIZE> digest;

    struct gcm_aes_ctx aes;
    gcm_aes_set_key(&aes, key.size(), key.data());
-    gcm_aes_set_iv(&aes, GCM_IV_SIZE, data.data());
+    gcm_aes_set_iv(&aes, GCM_IV_SIZE, data);

-    size_t data_sz = data.size() - GCM_IV_SIZE - GCM_DIGEST_SIZE;
+    size_t data_sz = data_length - GCM_IV_SIZE - GCM_DIGEST_SIZE;
    Blob ret(data_sz);
-    gcm_aes_decrypt(&aes, data_sz, ret.data(), data.data() + GCM_IV_SIZE);
+    gcm_aes_decrypt(&aes, data_sz, ret.data(), data + GCM_IV_SIZE);
    gcm_aes_digest(&aes, GCM_DIGEST_SIZE, digest.data());

-    if (not std::equal(digest.begin(), digest.end(), data.end() - GCM_DIGEST_SIZE)) {
+    if (not std::equal(digest.begin(), digest.end(), data + data_length - GCM_DIGEST_SIZE)) {
 #if DHT_AES_LEGACY_DECRYPT
-        //gcm_aes_decrypt(&aes, data_sz, ret.data(), data.data() + GCM_IV_SIZE);
+        //gcm_aes_decrypt(&aes, data_sz, ret.data(), data + GCM_IV_SIZE);
        Blob ret_tmp(data_sz);
        struct gcm_aes_ctx aes_d;
        gcm_aes_set_key(&aes_d, key.size(), key.data());
-        gcm_aes_set_iv(&aes_d, GCM_IV_SIZE, data.data());
+        gcm_aes_set_iv(&aes_d, GCM_IV_SIZE, data);
        gcm_aes_update(&aes_d, ret.size(), ret.data());
        gcm_aes_encrypt(&aes_d, ret.size(), ret_tmp.data(), ret.data());
        gcm_aes_digest(&aes_d, GCM_DIGEST_SIZE, digest.data());

-        if (not std::equal(digest.begin(), digest.end(), data.end() - GCM_DIGEST_SIZE))
+        if (not std::equal(digest.begin(), digest.end(), data + data_length - GCM_DIGEST_SIZE))
            throw DecryptError("Can't decrypt data");
 #else
        throw DecryptError("Can't decrypt data");
@@ -168,14 +168,13 @@ Blob aesDecrypt(const Blob& data, const Blob& key)
    return ret;
 }

-Blob aesDecrypt(const Blob& data, const std::string& password)
+Blob aesDecrypt(const uint8_t* data, size_t data_len, const std::string& password)
 {
-    if (data.size() <= PASSWORD_SALT_LENGTH)
+    if (data_len <= PASSWORD_SALT_LENGTH)
        throw DecryptError("Wrong data size");
-    Blob salt {data.begin(), data.begin()+PASSWORD_SALT_LENGTH};
+    Blob salt {data, data+PASSWORD_SALT_LENGTH};
    Blob key = stretchKey(password, salt, 256/8);
-    Blob encrypted {data.begin()+PASSWORD_SALT_LENGTH, data.end()};
-    return aesDecrypt(encrypted, key);
+    return aesDecrypt(data+PASSWORD_SALT_LENGTH, data_len - PASSWORD_SALT_LENGTH, key);
 }

 Blob stretchKey(const std::string& password, Blob& salt, size_t key_length)
@@ -292,14 +291,14 @@ PrivateKey::operator=(PrivateKey&& o) noexcept
 }

 Blob
-PrivateKey::sign(const Blob& data) const
+PrivateKey::sign(const uint8_t* data, size_t data_length) const
 {
    if (!key)
        throw CryptoException("Can't sign data: no private key set !");
-    if (std::numeric_limits<unsigned>::max() < data.size())
+    if (std::numeric_limits<unsigned>::max() < data_length)
        throw CryptoException("Can't sign data: too large !");
    gnutls_datum_t sig;
-    const gnutls_datum_t dat {(unsigned char*)data.data(), (unsigned)data.size()};
+    const gnutls_datum_t dat {(unsigned char*)data, (unsigned)data_length};
    if (gnutls_privkey_sign_data(key, GNUTLS_DIG_SHA512, 0, &dat, &sig) != GNUTLS_E_SUCCESS)
        throw CryptoException("Can't sign data !");
    Blob ret(sig.data, sig.data+sig.size);
@@ -321,7 +320,7 @@ PrivateKey::decryptBloc(const uint8_t* src, size_t src_size) const
 }

 Blob
-PrivateKey::decrypt(const Blob& cipher) const
+PrivateKey::decrypt(const uint8_t* cypher, size_t cypher_len) const
 {
    if (!key)
        throw CryptoException("Can't decrypt data without private key !");
@@ -334,12 +333,12 @@ PrivateKey::decrypt(const Blob& cipher) const
        throw CryptoException("Must be an RSA key");

    unsigned cypher_block_sz = key_len / 8;
-    if (cipher.size() < cypher_block_sz)
+    if (cypher_len < cypher_block_sz)
        throw DecryptError("Unexpected cipher length");
-    else if (cipher.size() == cypher_block_sz)
-        return decryptBloc(cipher.data(), cypher_block_sz);
+    else if (cypher_len == cypher_block_sz)
+        return decryptBloc(cypher, cypher_block_sz);

-    return aesDecrypt(Blob {cipher.begin() + cypher_block_sz, cipher.end()}, decryptBloc(cipher.data(), cypher_block_sz));
+    return aesDecrypt(cypher + cypher_block_sz, cypher_len - cypher_block_sz, decryptBloc(cypher, cypher_block_sz));
 }

 Blob
@@ -711,13 +710,13 @@ CertificateRequest::sign(const PrivateKey& key, const std::string& password)
 }

 bool
-CertificateRequest::verify() const 
+CertificateRequest::verify() const
 {
    return gnutls_x509_crq_verify(request, 0) >= 0;
 }

-Blob 
-CertificateRequest::pack() const 
+Blob
+CertificateRequest::pack() const
 {
    gnutls_datum_t dat {nullptr, 0};
    if (auto err = gnutls_x509_crq_export2(request, GNUTLS_X509_FMT_PEM, &dat))
@@ -728,7 +727,7 @@ CertificateRequest::pack() const
 }

 std::string
-CertificateRequest::toString() const 
+CertificateRequest::toString() const
 {
    gnutls_datum_t dat {nullptr, 0};
    if (auto err = gnutls_x509_crq_export2(request, GNUTLS_X509_FMT_PEM, &dat))
@@ -1148,8 +1147,8 @@ void
 setRandomSerial(gnutls_x509_crt_t cert)
 {
    random_device rdev;
-    std::uniform_int_distribution<uint64_t> dist{};
-    uint64_t cert_serial = dist(rdev);
+    std::uniform_int_distribution<int64_t> dist{1};
+    int64_t cert_serial = dist(rdev);
    gnutls_x509_crt_set_serial(cert, &cert_serial, sizeof(cert_serial));
 }

@@ -1241,6 +1240,70 @@ Certificate::getRevocationLists() const
    return ret;
 }

+// OcspRequest
+
+OcspRequest::OcspRequest(const uint8_t* dat_ptr, size_t dat_size)
+{
+    int ret = gnutls_ocsp_req_init(&request);
+    if (ret < 0)
+        throw CryptoException(gnutls_strerror(ret));
+    gnutls_datum_t dat = {(unsigned char*)dat_ptr,(unsigned int)dat_size};
+    ret = gnutls_ocsp_req_import(request, &dat);
+    if (ret < 0){
+        gnutls_ocsp_req_deinit(request);
+        throw CryptoException(gnutls_strerror(ret));
+    }
+}
+
+OcspRequest::~OcspRequest()
+{
+    if (request) {
+        gnutls_ocsp_req_deinit(request);
+        request = nullptr;
+    }
+}
+
+std::string
+OcspRequest::toString(const bool compact) const
+{
+    int ret;
+    gnutls_datum_t dat;
+    ret = gnutls_ocsp_req_print(request, compact ? GNUTLS_OCSP_PRINT_COMPACT : GNUTLS_OCSP_PRINT_FULL, &dat);
+
+    std::string str;
+    if (ret == 0) {
+        str = std::string((const char*)dat.data, (size_t)dat.size);
+        gnutls_free(dat.data);
+    } else
+        throw CryptoException(gnutls_strerror(ret));
+    return str;
+}
+
+Blob
+OcspRequest::pack() const
+{
+    gnutls_datum_t dat;
+    int err = gnutls_ocsp_req_export(request, &dat);
+    if (err < 0)
+        throw CryptoException(gnutls_strerror(err));
+    Blob ret {dat.data, dat.data + dat.size};
+    gnutls_free(dat.data);
+    return ret;
+}
+
+Blob
+OcspRequest::getNonce() const
+{
+    gnutls_datum_t dat;
+    unsigned critical;
+    int err = gnutls_ocsp_req_get_nonce(request, &critical, &dat);
+    if (err < 0)
+        throw CryptoException(gnutls_strerror(err));
+    Blob ret {dat.data, dat.data + dat.size};
+    gnutls_free(dat.data);
+    return ret;
+}
+
 // OcspResponse

 OcspResponse::OcspResponse(const uint8_t* dat_ptr, size_t dat_size)
@@ -1265,10 +1328,12 @@ Blob
 OcspResponse::pack() const
 {
    gnutls_datum_t dat;
-    int ret = gnutls_ocsp_resp_export(response, &dat);
-    if (ret < 0)
-        throw CryptoException(gnutls_strerror(ret));
-    return {dat.data, dat.data + dat.size};
+    int err = gnutls_ocsp_resp_export(response, &dat);
+    if (err < 0)
+        throw CryptoException(gnutls_strerror(err));
+    Blob ret {dat.data, dat.data + dat.size};
+    gnutls_free(dat.data);
+    return ret;
 }

 std::string
@@ -1298,7 +1363,7 @@ OcspResponse::getCertificateStatus() const
    return (gnutls_ocsp_cert_status_t) status;
 }

-gnutls_ocsp_verify_reason_t
+gnutls_ocsp_cert_status_t
 OcspResponse::verifyDirect(const Certificate& crt, const Blob& nonce)
 {
    // Check OCSP response
@@ -1309,39 +1374,52 @@ OcspResponse::verifyDirect(const Certificate& crt, const Blob& nonce)
    if (status != GNUTLS_OCSP_RESP_SUCCESSFUL)
        throw CryptoException("OCSP request unsuccessful: " + std::to_string(ret));

-    // Check whether the OCSP response is about the provided certificate.
-    if ((ret = gnutls_ocsp_resp_check_crt(response, 0, crt.cert)) < 0)
-        throw CryptoException(gnutls_strerror(ret));
+    if (not nonce.empty()) {
+        // Ensure no replay attack has been done
+        gnutls_datum_t rnonce;
+        ret = gnutls_ocsp_resp_get_nonce(response, NULL, &rnonce);
+        if (ret < 0)
+            throw CryptoException(gnutls_strerror(ret));
+        if (rnonce.size != nonce.size() || memcmp(nonce.data(), rnonce.data, nonce.size()) != 0){
+            gnutls_free(rnonce.data);
+            throw CryptoException(gnutls_strerror(GNUTLS_E_OCSP_RESPONSE_ERROR));
+        }
+        gnutls_free(rnonce.data);
+    }

    // Verify signature of the Basic OCSP response against the public key in the issuer certificate.
    unsigned int verify = 0;
    ret = gnutls_ocsp_resp_verify_direct(response, crt.issuer->cert, &verify, 0);
    if (ret < 0)
        throw CryptoException(gnutls_strerror(ret));
+    if (verify) {
+        if (verify & GNUTLS_OCSP_VERIFY_SIGNER_NOT_FOUND)
+            throw CryptoException("Signer cert not found");
+        if (verify & GNUTLS_OCSP_VERIFY_SIGNER_KEYUSAGE_ERROR)
+            throw CryptoException("Signer cert keyusage error");
+        if (verify & GNUTLS_OCSP_VERIFY_UNTRUSTED_SIGNER)
+            throw CryptoException("Signer cert is not trusted");
+        if (verify & GNUTLS_OCSP_VERIFY_INSECURE_ALGORITHM)
+            throw CryptoException("Insecure algorithm");
+        if (verify & GNUTLS_OCSP_VERIFY_SIGNATURE_FAILURE)
+            throw CryptoException("Signature failure");
+        if (verify & GNUTLS_OCSP_VERIFY_CERT_NOT_ACTIVATED)
+            throw CryptoException("Signer cert not yet activated");
+        if (verify & GNUTLS_OCSP_VERIFY_CERT_EXPIRED)
+            throw CryptoException("Signer cert expired");
+        throw CryptoException(gnutls_strerror(GNUTLS_E_OCSP_RESPONSE_ERROR));
+    }
+
+    // Check whether the OCSP response is about the provided certificate.
+    if ((ret = gnutls_ocsp_resp_check_crt(response, 0, crt.cert)) < 0)
+        throw CryptoException(gnutls_strerror(ret));

    // Check certificate revocation status
    unsigned status_ocsp;
    ret = gnutls_ocsp_resp_get_single(response, 0, NULL, NULL, NULL, NULL, &status_ocsp, NULL, NULL, NULL, NULL);
    if (ret < 0)
        throw CryptoException(gnutls_strerror(ret));
-    gnutls_ocsp_cert_status_t certStatus = (gnutls_ocsp_cert_status_t)ret;
-    if (certStatus == GNUTLS_OCSP_CERT_REVOKED)
-        throw CryptoException("Certificate was revoked");
-    else if (certStatus != GNUTLS_OCSP_CERT_GOOD)
-        throw CryptoException("Certificate is unknown");
-
-    // Ensure no replay attack has been done
-    gnutls_datum_t rnonce;
-    ret = gnutls_ocsp_resp_get_nonce(response, NULL, &rnonce);
-    if (ret < 0)
-        throw CryptoException(gnutls_strerror(ret));
-    if (rnonce.size != nonce.size() || memcmp(nonce.data(), rnonce.data, nonce.size()) != 0){
-        gnutls_free(rnonce.data);
-        throw CryptoException(gnutls_strerror(GNUTLS_E_OCSP_RESPONSE_ERROR));
-    }
-    gnutls_free(rnonce.data);
-
-    return (gnutls_ocsp_verify_reason_t) verify;
+    return (gnutls_ocsp_cert_status_t)status_ocsp;
 }

 // RevocationList
@@ -52,12 +52,7 @@ Dht::updateStatus(sa_family_t af)
        if (other.status == NodeStatus::Disconnected && d.status == NodeStatus::Disconnected)
            onDisconnected();
        else if (other.status == NodeStatus::Connected || d.status == NodeStatus::Connected) {
-            // On connected
-            if (bootstrapJob) {
-                bootstrapJob->cancel();
-                bootstrapJob.reset();
-            }
-            bootstrap_period = std::chrono::seconds(10);
+            onConnected();
        }
    }
    return d.status;
@@ -82,11 +77,27 @@ Dht::Kad::getStatus(time_point now) const
 }

 void
-Dht::shutdown(ShutdownCallback cb)
+Dht::shutdown(ShutdownCallback cb, bool stop)
 {
    if (not persistPath.empty())
        saveState(persistPath);

+    if (stop) {
+        for (auto dht : {&dht4, &dht6}) {
+            for (auto& sr : dht->searches) {
+                for (const auto& r : sr.second->callbacks)
+                    r.second.done_cb(false, {});
+                sr.second->callbacks.clear();
+                for (const auto& a : sr.second->announce) {
+                    if (a.callback) a.callback(false, {});
+                }
+                sr.second->announce.clear();
+                sr.second->listeners.clear();
+            }
+        }
+        network_engine.clear();
+    }
+
    if (not maintain_storage) {
        if (cb) cb();
        return;
@@ -491,9 +502,9 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr) {
                        sr->id.toString().c_str(), sn->node->toString().c_str(), a.value->id);
                sn->acked[a.value->id] = {
                    network_engine.sendRefreshValue(sn->node, sr->id, a.value->id, sn->token, onDone,
-                    [this, ws, node=sn->node, v=a.value, 
-                     onDone, 
-                     onExpired, 
+                    [this, ws, node=sn->node, v=a.value,
+                     onDone,
+                     onExpired,
                     created = a.permanent ? time_point::max() : a.created,
                     next_refresh_time
                    ](const net::Request& /*req*/, net::DhtProtocolException&& e){
@@ -591,7 +602,7 @@ Dht::searchSynchedNodeListen(const Sp<Search>& sr, SearchNode& n)
    std::weak_ptr<Search> ws = sr;
    for (const auto& l : sr->listeners) {
        const auto& query = l.second.query;
-        
+
        auto r = n.listenStatus.find(query);
        if (n.getListenTime(r, listenExp) > scheduler.time())
            continue;
@@ -725,7 +736,7 @@ Dht::searchStep(Sp<Search> sr)

    while (sr->currentlySolicitedNodeCount() < MAX_REQUESTED_SEARCH_NODES and searchSendGetValues(sr));

-    
+
    if (sr->getNumberOfConsecutiveBadNodes() >= std::min<size_t>(sr->nodes.size(), SEARCH_MAX_BAD_NODES))
    {
        if (logger_)
@@ -1747,8 +1758,6 @@ fromDhtConfig(const Config& config)
    return netConf;
 }

-Dht::Dht() : store(), network_engine(logger_, rd, scheduler, {}) {}
-
 Dht::Dht(std::unique_ptr<net::DatagramSocket>&& sock, const Config& config, const Sp<Logger>& l)
    : DhtInterface(l),
    myid(config.node_id ? config.node_id : InfoHash::getRandom(rd)),
@@ -1983,6 +1992,21 @@ Dht::expire()
    scheduler.add(expire_stuff_time, std::bind(&Dht::expire, this));
 }

+void
+Dht::onConnected()
+{
+    if (bootstrapJob) {
+        bootstrapJob->cancel();
+        bootstrapJob.reset();
+    }
+    bootstrap_period = std::chrono::seconds(10);
+    auto callbacks = std::move(onConnectCallbacks_);
+    while (not callbacks.empty()) {
+        callbacks.front()();
+        callbacks.pop();
+    }
+}
+
 void
 Dht::onDisconnected()
 {
@@ -2528,7 +2552,7 @@ Dht::saveState(const std::string& path) const
    state.nodes = exportNodes();
    state.values = exportValues();
    std::ofstream file(path);
-    msgpack::pack(file, state);    
+    msgpack::pack(file, state);
 }

 void
@@ -155,10 +155,10 @@ DhtProxyClient::startProxy()
    if (logger_)
        logger_->d("[proxy:client] start proxy with %s", proxyUrl_.c_str());

-    nextProxyConfirmationTimer_ = std::make_shared<asio::steady_timer>(httpContext_, std::chrono::steady_clock::now());
+    nextProxyConfirmationTimer_ = std::make_unique<asio::steady_timer>(httpContext_, std::chrono::steady_clock::now());
    nextProxyConfirmationTimer_->async_wait(std::bind(&DhtProxyClient::handleProxyConfirm, this, std::placeholders::_1));

-    listenerRestartTimer_ = std::make_shared<asio::steady_timer>(httpContext_);
+    listenerRestartTimer_ = std::make_unique<asio::steady_timer>(httpContext_);

    loopSignal_();
 }
@@ -242,7 +242,7 @@ DhtProxyClient::cancelAllListeners()
 }

 void
-DhtProxyClient::shutdown(ShutdownCallback cb)
+DhtProxyClient::shutdown(ShutdownCallback cb, bool)
 {
    stop();
    if (cb)
@@ -723,6 +723,15 @@ DhtProxyClient::onProxyInfos(const Json::Value& proxyInfos, const sa_family_t fa
        if (oldStatus == NodeStatus::Disconnected || oldStatus == NodeStatus::Connecting) {
            listenerRestartTimer_->expires_at(std::chrono::steady_clock::now());
            listenerRestartTimer_->async_wait(std::bind(&DhtProxyClient::restartListeners, this, std::placeholders::_1));
+            if (not onConnectCallbacks_.empty()) {
+                std::lock_guard<std::mutex> lock(lockCallbacks_);
+                callbacks_.emplace_back([cbs = std::move(onConnectCallbacks_)]() mutable {
+                    while (not cbs.empty()) {
+                        cbs.front()();
+                        cbs.pop();
+                    }
+                });
+            }
        }
        nextProxyConfirmationTimer_->expires_at(std::chrono::steady_clock::now() + std::chrono::minutes(15));
        nextProxyConfirmationTimer_->async_wait(std::bind(&DhtProxyClient::handleProxyConfirm, this, std::placeholders::_1));
@@ -822,7 +822,7 @@ DhtProxyServer::subscribe(restinio::request_handle_t request,
            logger_->d("[proxy:server] [subscribe %s] [client %s] [session %s]", infoHash.toString().c_str(), clientId.c_str(), sessionId.c_str());

        // Insert new or return existing push listeners of a token
-        std::lock_guard<std::mutex> lock(lockPushListeners_);
+        std::lock_guard<std::mutex> lock(lockListener_);
        auto& pushListener = pushListeners_[pushToken];
        auto& pushListeners = pushListener.listeners[infoHash];

@@ -76,19 +76,17 @@ DhtRunner::~DhtRunner()
 }

 void
-DhtRunner::run(in_port_t port, const Config& config, Context&& context)
+DhtRunner::run(in_port_t port, Config& config, Context&& context)
 {
-    SockAddr sin4;
-    sin4.setFamily(AF_INET);
-    sin4.setPort(port);
-    SockAddr sin6;
-    sin6.setFamily(AF_INET6);
-    sin6.setPort(port);
-    run(sin4, sin6, config, std::move(context));
+    config.bind4.setFamily(AF_INET);
+    config.bind4.setPort(port);
+    config.bind6.setFamily(AF_INET6);
+    config.bind6.setPort(port);
+    run(config, std::move(context));
 }

 void
-DhtRunner::run(const char* ip4, const char* ip6, const char* service, const Config& config, Context&& context)
+DhtRunner::run(const char* ip4, const char* ip6, const char* service, Config& config, Context&& context)
 {
    auto res4 = SockAddr::resolve(ip4, service);
    auto res6 = SockAddr::resolve(ip6, service);
@@ -96,44 +94,9 @@ DhtRunner::run(const char* ip4, const char* ip6, const char* service, const Conf
        res4.emplace_back();
    if (res6.empty())
        res6.emplace_back();
-    run(res4.front(), res6.front(), config, std::move(context));
-}
-
-void
-DhtRunner::run(SockAddr& local4, SockAddr& local6, const Config& config, Context&& context)
-{
-    if (running == State::Idle) {
-        auto state_path = config.dht_config.node_config.persist_path;
-        if (not state_path.empty()) {
-            state_path += "_port.txt";
-            std::ifstream inConfig(state_path);
-            if (inConfig.is_open()) {
-                in_port_t port;
-                if (inConfig >> port) {
-                    if (context.logger)
-                        context.logger->d("[runner %p] Using IPv4 port %hu from saved configuration", this, port);
-                    if (local4.getPort() == 0)
-                        local4.setPort(port);
-                }
-                if (inConfig >> port) {
-                    if (context.logger)
-                        context.logger->d("[runner %p] Using IPv6 port %hu from saved configuration", this, port);
-                    if (local6.getPort() == 0)
-                        local6.setPort(port);
-                }
-            }
-        }
-
-        if (not context.sock)
-            context.sock.reset(new net::UdpSocket(local4, local6, context.logger));
-
-        if (not state_path.empty()) {
-            std::ofstream outConfig(state_path);
-            outConfig << context.sock->getBoundRef(AF_INET).getPort() << std::endl;
-            outConfig << context.sock->getBoundRef(AF_INET6).getPort() << std::endl;
-        }
-        run(config, std::move(context));
-    }
+    config.bind4 = std::move(res4.front());
+    config.bind6 = std::move(res6.front());
+    run(config, std::move(context));
 }

 void
@@ -141,8 +104,51 @@ DhtRunner::run(const Config& config, Context&& context)
 {
    std::lock_guard<std::mutex> lck(dht_mtx);
    auto expected = State::Idle;
-    if (not running.compare_exchange_strong(expected, State::Running))
+    if (not running.compare_exchange_strong(expected, State::Running)) {
+        if (context.logger)
+            context.logger->w("[runner %p] Node is already running. Call join() first before calling run() again.");
        return;
+    }
+
+    auto local4 = config.bind4;
+    auto local6 = config.bind6;
+    if (not local4 and not local6) {
+        if (context.logger)
+            context.logger->w("[runner %p] No address to bind specified in the configuration, using default addresses");
+        local4.setFamily(AF_INET);
+        local6.setFamily(AF_INET6);
+    }
+    auto state_path = config.dht_config.node_config.persist_path;
+    if (not state_path.empty() && (local4.getPort() == 0 || local6.getPort() == 0)) {
+        state_path += "_port.txt";
+        std::ifstream inConfig(state_path);
+        if (inConfig.is_open()) {
+            in_port_t port;
+            if (inConfig >> port) {
+                if (local4.getPort() == 0) {
+                    if (context.logger)
+                        context.logger->d("[runner %p] Using IPv4 port %hu from saved configuration", this, port);
+                    local4.setPort(port);
+                }
+            }
+            if (inConfig >> port) {
+                if (local6.getPort() == 0) {
+                    if (context.logger)
+                        context.logger->d("[runner %p] Using IPv6 port %hu from saved configuration", this, port);
+                    local6.setPort(port);
+                }
+            }
+        }
+    }
+
+    if (not context.sock)
+        context.sock.reset(new net::UdpSocket(local4, local6, context.logger));
+
+    if (not state_path.empty()) {
+        std::ofstream outConfig(state_path);
+        outConfig << context.sock->getBoundRef(AF_INET).getPort() << std::endl;
+        outConfig << context.sock->getBoundRef(AF_INET6).getPort() << std::endl;
+    }

    if (context.logger) {
        logger_ = context.logger;
@@ -167,12 +173,13 @@ DhtRunner::run(const Config& config, Context&& context)
        return ret;
    });

-    auto dht = std::unique_ptr<DhtInterface>(new Dht(std::move(context.sock), SecureDht::getConfig(config.dht_config), context.logger));
-    dht_ = std::unique_ptr<SecureDht>(new SecureDht(std::move(dht), config.dht_config));
-
 #ifdef OPENDHT_PROXY_CLIENT
    config_ = config;
+    identityAnnouncedCb_ = context.identityAnnouncedCb;
 #endif
+    auto dht = std::unique_ptr<DhtInterface>(new Dht(std::move(context.sock), SecureDht::getConfig(config.dht_config), context.logger));
+    dht_ = std::unique_ptr<SecureDht>(new SecureDht(std::move(dht), config.dht_config, std::move(context.identityAnnouncedCb), context.logger));
+
    enableProxy(not config.proxy_server.empty());
    if (context.logger and dht_via_proxy_) {
        dht_via_proxy_->setLogger(context.logger);
@@ -261,29 +268,31 @@ DhtRunner::run(const Config& config, Context&& context)
 }

 void
-DhtRunner::shutdown(ShutdownCallback cb) {
+DhtRunner::shutdown(ShutdownCallback cb, bool stop) {
+    std::unique_lock<std::mutex> lck(storage_mtx);
    auto expected = State::Running;
    if (not running.compare_exchange_strong(expected, State::Stopping)) {
        if (expected == State::Stopping and ongoing_ops) {
-            std::lock_guard<std::mutex> lck(storage_mtx);
            shutdownCallbacks_.emplace_back(std::move(cb));
        }
-        else if (cb) cb();
+        else if (cb) {
+            lck.unlock();
+            cb();
+        }
        return;
    }
    if (logger_)
        logger_->d("[runner %p] state changed to Stopping, %zu ongoing ops", this, ongoing_ops.load());
-    std::lock_guard<std::mutex> lck(storage_mtx);
    ongoing_ops++;
    shutdownCallbacks_.emplace_back(std::move(cb));
-    pending_ops_prio.emplace([=](SecureDht&) mutable {
+    pending_ops.emplace([=](SecureDht&) mutable {
        auto onShutdown = [this]{ opEnded(); };
 #ifdef OPENDHT_PROXY_CLIENT
        if (dht_via_proxy_)
-            dht_via_proxy_->shutdown(onShutdown);
+            dht_via_proxy_->shutdown(onShutdown, stop);
 #endif
        if (dht_)
-            dht_->shutdown(onShutdown);
+            dht_->shutdown(onShutdown, stop);
    });
    cv.notify_all();
 }
@@ -312,11 +321,11 @@ DhtRunner::bindOpDoneCallback(DoneCallbackSimple&& cb) {

 bool
 DhtRunner::checkShutdown() {
-    if (running != State::Stopping or ongoing_ops)
-        return false;
    decltype(shutdownCallbacks_) cbs;
    {
        std::lock_guard<std::mutex> lck(storage_mtx);
+        if (running != State::Stopping or ongoing_ops)
+            return false;
        cbs = std::move(shutdownCallbacks_);
    }
    for (auto& cb : cbs)
@@ -348,9 +357,13 @@ DhtRunner::join()

    {
        std::lock_guard<std::mutex> lck(storage_mtx);
+        if (ongoing_ops and logger_) {
+            logger_->w("[runner %p] stopping with %zu remaining ops", this, ongoing_ops.load());
+        }
        pending_ops = decltype(pending_ops)();
        pending_ops_prio = decltype(pending_ops_prio)();
        ongoing_ops = 0;
+        shutdownCallbacks_.clear();
    }
    {
        std::lock_guard<std::mutex> lck(dht_mtx);
@@ -593,6 +606,18 @@ DhtRunner::getPublicAddressStr(sa_family_t af)
    return ret;
 }

+void
+DhtRunner::getPublicAddress(std::function<void(std::vector<SockAddr>&&)> cb, sa_family_t af)
+{
+    std::lock_guard<std::mutex> lck(storage_mtx);
+    ongoing_ops++;
+    pending_ops_prio.emplace([cb = std::move(cb), this, af](SecureDht& dht){
+        cb(dht.getPublicAddress(af));
+        opEnded();
+    });
+    cv.notify_all();
+}
+
 void
 DhtRunner::registerCertificate(std::shared_ptr<crypto::Certificate> cert) {
    std::lock_guard<std::mutex> lck(dht_mtx);
@@ -690,11 +715,12 @@ DhtRunner::loop_()
 void
 DhtRunner::get(InfoHash hash, GetCallback vcb, DoneCallback dcb, Value::Filter f, Where w)
 {
+    std::unique_lock<std::mutex> lck(storage_mtx);
    if (running != State::Running) {
+        lck.unlock();
        if (dcb) dcb(false, {});
        return;
    }
-    std::lock_guard<std::mutex> lck(storage_mtx);
    ongoing_ops++;
    pending_ops.emplace([=](SecureDht& dht) mutable {
        dht.get(hash, std::move(vcb), bindOpDoneCallback(std::move(dcb)), std::move(f), std::move(w));
@@ -709,11 +735,12 @@ DhtRunner::get(const std::string& key, GetCallback vcb, DoneCallbackSimple dcb,
 }
 void
 DhtRunner::query(const InfoHash& hash, QueryCallback cb, DoneCallback done_cb, Query q) {
+    std::unique_lock<std::mutex> lck(storage_mtx);
    if (running != State::Running) {
+        lck.unlock();
        if (done_cb) done_cb(false, {});
        return;
    }
-    std::lock_guard<std::mutex> lck(storage_mtx);
    ongoing_ops++;
    pending_ops.emplace([=](SecureDht& dht) mutable {
        dht.query(hash, std::move(cb), bindOpDoneCallback(std::move(done_cb)), std::move(q));
@@ -725,11 +752,12 @@ std::future<size_t>
 DhtRunner::listen(InfoHash hash, ValueCallback vcb, Value::Filter f, Where w)
 {
    auto ret_token = std::make_shared<std::promise<size_t>>();
+    std::unique_lock<std::mutex> lck(storage_mtx);
    if (running != State::Running) {
+        lck.unlock();
        ret_token->set_value(0);
        return ret_token->get_future();
    }
-    std::lock_guard<std::mutex> lck(storage_mtx);
    pending_ops.emplace([=](SecureDht& dht) mutable {
 #ifdef OPENDHT_PROXY_CLIENT
        auto tokenbGlobal = listener_token_++;
@@ -810,11 +838,12 @@ DhtRunner::cancelListen(InfoHash h, std::shared_future<size_t> ftoken)
 void
 DhtRunner::put(InfoHash hash, Value&& value, DoneCallback cb, time_point created, bool permanent)
 {
+    std::unique_lock<std::mutex> lck(storage_mtx);
    if (running != State::Running) {
+        lck.unlock();
        if (cb) cb(false, {});
        return;
    }
-    std::lock_guard<std::mutex> lck(storage_mtx);
    ongoing_ops++;
    pending_ops.emplace([=,
        cb = std::move(cb),
@@ -828,11 +857,12 @@ DhtRunner::put(InfoHash hash, Value&& value, DoneCallback cb, time_point created
 void
 DhtRunner::put(InfoHash hash, std::shared_ptr<Value> value, DoneCallback cb, time_point created, bool permanent)
 {
+    std::unique_lock<std::mutex> lck(storage_mtx);
    if (running != State::Running) {
+        lck.unlock();
        if (cb) cb(false, {});
        return;
    }
-    std::lock_guard<std::mutex> lck(storage_mtx);
    ongoing_ops++;
    pending_ops.emplace([=, cb = std::move(cb)](SecureDht& dht) mutable {
        dht.put(hash, value, bindOpDoneCallback(std::move(cb)), created, permanent);
@@ -869,11 +899,12 @@ DhtRunner::cancelPut(const InfoHash& h, const std::shared_ptr<Value>& value)
 void
 DhtRunner::putSigned(InfoHash hash, std::shared_ptr<Value> value, DoneCallback cb, bool permanent)
 {
+    std::unique_lock<std::mutex> lck(storage_mtx);
    if (running != State::Running) {
+        lck.unlock();
        if (cb) cb(false, {});
        return;
    }
-    std::lock_guard<std::mutex> lck(storage_mtx);
    ongoing_ops++;
    pending_ops.emplace([=,
        cb = std::move(cb),
@@ -899,11 +930,12 @@ DhtRunner::putSigned(const std::string& key, Value&& value, DoneCallbackSimple c
 void
 DhtRunner::putEncrypted(InfoHash hash, InfoHash to, std::shared_ptr<Value> value, DoneCallback cb, bool permanent)
 {
+    std::unique_lock<std::mutex> lck(storage_mtx);
    if (running != State::Running) {
+        lck.unlock();
        if (cb) cb(false, {});
        return;
    }
-    std::lock_guard<std::mutex> lck(storage_mtx);
    ongoing_ops++;
    pending_ops.emplace([=,
        cb = std::move(cb),
@@ -989,11 +1021,12 @@ DhtRunner::bootstrap(std::vector<SockAddr> nodes, DoneCallbackSimple&& cb)
 void
 DhtRunner::bootstrap(const SockAddr& addr, DoneCallbackSimple&& cb)
 {
+    std::unique_lock<std::mutex> lck(storage_mtx);
    if (running != State::Running) {
+        lck.unlock();
        if (cb) cb(false);
        return;
    }
-    std::lock_guard<std::mutex> lck(storage_mtx);
    ongoing_ops++;
    pending_ops_prio.emplace([addr, cb = bindOpDoneCallback(std::move(cb))](SecureDht& dht) mutable {
        dht.pingNode(std::move(addr), std::move(cb));
@@ -1004,9 +1037,9 @@ DhtRunner::bootstrap(const SockAddr& addr, DoneCallbackSimple&& cb)
 void
 DhtRunner::bootstrap(const InfoHash& id, const SockAddr& address)
 {
+    std::lock_guard<std::mutex> lck(storage_mtx);
    if (running != State::Running)
        return;
-    std::unique_lock<std::mutex> lck(storage_mtx);
    pending_ops_prio.emplace([id, address](SecureDht& dht) mutable {
        dht.insertNode(id, address);
    });
@@ -1016,9 +1049,9 @@ DhtRunner::bootstrap(const InfoHash& id, const SockAddr& address)
 void
 DhtRunner::bootstrap(const std::vector<NodeExport>& nodes)
 {
+    std::lock_guard<std::mutex> lck(storage_mtx);
    if (running != State::Running)
        return;
-    std::lock_guard<std::mutex> lck(storage_mtx);
    pending_ops_prio.emplace([=](SecureDht& dht) {
        for (auto& node : nodes)
            dht.insertNode(node);
@@ -1042,11 +1075,12 @@ DhtRunner::connectivityChanged()

 void
 DhtRunner::findCertificate(InfoHash hash, std::function<void(const Sp<crypto::Certificate>&)> cb) {
+    std::unique_lock<std::mutex> lck(storage_mtx);
    if (running != State::Running) {
+        lck.unlock();
        cb({});
        return;
    }
-    std::lock_guard<std::mutex> lck(storage_mtx);
    ongoing_ops++;
    pending_ops.emplace([this, hash, cb = std::move(cb)] (SecureDht& dht) {
        dht.findCertificate(hash, [this, cb = std::move(cb)](const Sp<crypto::Certificate>& crt){
@@ -1122,7 +1156,7 @@ DhtRunner::enableProxy(bool proxify)
        if (not config_.push_token.empty())
            dht_via_proxy->setPushNotificationToken(config_.push_token);
 #endif
-        dht_via_proxy_ = std::unique_ptr<SecureDht>(new SecureDht(std::move(dht_via_proxy), config_.dht_config));
+        dht_via_proxy_ = std::unique_ptr<SecureDht>(new SecureDht(std::move(dht_via_proxy), config_.dht_config, identityAnnouncedCb_, logger_));
        // add current listeners
        for (auto& l: listeners_)
            l.second.tokenProxyDht = dht_via_proxy_->listen(l.second.hash, l.second.gcb, l.second.f, l.second.w);
@@ -561,12 +561,63 @@ void
 Connection::async_connect(std::vector<asio::ip::tcp::endpoint>&& endpoints, ConnectHandlerCb cb)
 {
    std::lock_guard<std::mutex> lock(mutex_);
-    if (ssl_socket_)
-        asio::async_connect(ssl_socket_->lowest_layer(), std::move(endpoints), wrapCallabck(std::move(cb)));
-    else if (socket_)
-        asio::async_connect(*socket_, std::move(endpoints), wrapCallabck(std::move(cb)));
-    else if (cb)
+    if (!ssl_socket_ && !socket_) {
        cb(asio::error::operation_aborted, {});
+        return;
+    }
+    auto& base = ssl_socket_? ssl_socket_->lowest_layer() : *socket_;
+
+    ConnectHandlerCb wcb = [&base, cb=std::move(cb)](const asio::error_code& ec, const asio::ip::tcp::endpoint& endpoint) {
+        if (!ec) {
+            auto socket = base.native_handle();
+            // Once connected, set a keep alive on the TCP socket with 30 seconds delay
+            // This will generate broken pipes as soon as possible.
+            // Note this needs to be done once connected to have a valid native_handle()
+            uint32_t start = 30;
+            uint32_t interval = 30;
+            uint32_t cnt = 1;
+#ifdef _WIN32
+            std::string val = "1";
+            setsockopt(socket, SOL_SOCKET, SO_KEEPALIVE, val.c_str(), sizeof(val));
+
+            // TCP_KEEPIDLE and TCP_KEEPINTVL are available since Win 10 version 1709
+            // TCP_KEEPCNT since Win 10 version 1703
+#ifdef TCP_KEEPIDLE
+            std::string start_str = std::to_string(start);
+            setsockopt(socket, IPPROTO_TCP, TCP_KEEPIDLE,
+                    start_str.c_str(), sizeof(start_str));
+#endif
+#ifdef TCP_KEEPINTVL
+            std::string interval_str = std::to_string(interval);
+            setsockopt(socket, IPPROTO_TCP, TCP_KEEPINTVL,
+                    interval_str.c_str(), sizeof(interval_str));
+#endif
+#ifdef TCP_KEEPCNT
+            std::string cnt_str = std::to_string(cnt);
+            setsockopt(socket, IPPROTO_TCP, TCP_KEEPCNT,
+                    cnt_str.c_str(), sizeof(cnt_str));
+#endif
+#else
+            uint32_t val = 1;
+            setsockopt(socket, SOL_SOCKET, SO_KEEPALIVE, &val, sizeof(uint32_t));
+#ifdef __APPLE__
+            // Apple devices only have one parameter
+            setsockopt(socket, IPPROTO_TCP, TCP_KEEPALIVE, &start, sizeof(uint32_t));
+#else
+            // Linux based systems
+            setsockopt(socket, SOL_TCP, TCP_KEEPIDLE, &start, sizeof(uint32_t));
+            setsockopt(socket, SOL_TCP, TCP_KEEPINTVL, &interval, sizeof(uint32_t));
+            setsockopt(socket, SOL_TCP, TCP_KEEPCNT, &cnt, sizeof(uint32_t));
+#endif
+#endif
+        }
+        if (cb)
+            cb(ec, endpoint);
+    };
+    if (ssl_socket_)
+        asio::async_connect(ssl_socket_->lowest_layer(), std::move(endpoints), wrapCallabck(std::move(wcb)));
+    else
+        asio::async_connect(*socket_, std::move(endpoints), wrapCallabck(std::move(wcb)));
 }

 void
@@ -1254,9 +1305,11 @@ Request::terminate(const asio::error_code& ec)
        return;

    response_.aborted = ec == asio::error::operation_aborted;
+    if (ec == asio::error::basic_errors::broken_pipe)
+        response_.status_code = 0U; // Avoid to give a successful answer (happen with a broken pipe, takes the last status)

    if (logger_) {
-        if (ec and ec != asio::error::eof and ec != asio::error::operation_aborted)
+        if (ec and ec != asio::error::eof and !response_.aborted)
            logger_->e("[http:request:%i] end with error: %s", id_, ec.message().c_str());
        else
            logger_->d("[http:request:%i] done with status code %u", id_, response_.status_code);
@@ -1415,7 +1468,7 @@ Request::await()
    std::unique_lock<std::mutex> lock(mtx);
    std::condition_variable cv;
    bool ok {false};
-    add_on_done_callback([&](const Response& resp){
+    add_on_done_callback([&](const Response&){
        std::lock_guard<std::mutex> lk(mtx);
        ok = true;
        cv.notify_all();
@@ -82,10 +82,6 @@ RequestAnswer::RequestAnswer(ParsedMessage&& msg)
   nodes6(std::move(msg.nodes6))
 {}

-NetworkEngine::NetworkEngine(const Sp<Logger>& log, std::mt19937_64& rand, Scheduler& scheduler, std::unique_ptr<DatagramSocket>&& sock)
-    : myid(zeroes), dht_socket(std::move(sock)), logger_(log), rd(rand), cache(rd), rate_limiter((size_t)-1), scheduler(scheduler)
-{}
-
 NetworkEngine::NetworkEngine(InfoHash& myid, NetworkConfig c,
        std::unique_ptr<DatagramSocket>&& sock,
        const Sp<Logger>& log,
@@ -380,7 +376,7 @@ NetworkEngine::isMartian(const SockAddr& addr)
        const uint8_t* address = (const uint8_t*)&sin6.sin6_addr;
        return address[0] == 0xFF ||
              (address[0] == 0xFE && (address[1] & 0xC0) == 0x80) ||
-               memcmp(address, zeroes.data(), 16) == 0 ||
+               memcmp(address, InfoHash::zero().data(), 16) == 0 ||
               memcmp(address, v4prefix,      12) == 0;
    }
    default:
@@ -460,9 +456,13 @@ NetworkEngine::processMessage(const uint8_t *buf, size_t buflen, SockAddr f)
            pmsg_it->second.last_part = now;
            // check data completion
            if (pmsg_it->second.msg->complete()) {
-                // process the full message
-                process(std::move(pmsg_it->second.msg), from);
-                partial_messages.erase(pmsg_it);
+                try {
+                    // process the full message
+                    process(std::move(pmsg_it->second.msg), from);
+                    partial_messages.erase(pmsg_it);
+		} catch (...) {
+                    return;
+		}  
            } else
                scheduler.add(now + RX_TIMEOUT, std::bind(&NetworkEngine::maintainRxBuffer, this, msg->tid));
        }
@@ -485,7 +485,11 @@ NetworkEngine::processMessage(const uint8_t *buf, size_t buflen, SockAddr f)
    }

    if (msg->value_parts.empty()) {
-        process(std::move(msg), from);
+        try {	    
+            process(std::move(msg), from);
+	} catch(...) {
+            return;
+	}
    } else {
        // starting partial message session
        auto k = msg->tid;
@@ -51,8 +51,8 @@ public:
    }

    bool onValue(const std::vector<Sp<Value>>& vals, bool expired, const system_clock::time_point& t = system_clock::time_point::min()) {
-        return expired 
-            ? onValuesExpired(vals, t) 
+        return expired
+            ? onValuesExpired(vals, t)
            : onValuesAdded(vals, t);
    }

@@ -74,6 +74,7 @@ public:
    std::vector<Sp<Value>> get(const Value::Filter& filter) const;
    Sp<Value> get(Value::Id id) const;
    std::vector<Sp<Value>> getValues() const;
+    size_t size() const { return values.size(); }

 private:
    OpValueCache(const OpValueCache&) = delete;
@@ -145,6 +146,10 @@ public:
    }
    time_point getExpiration() const;

+    size_t size() const {
+        return cache.size();
+    }
+
    size_t searchToken;
 private:
    constexpr static const std::chrono::seconds EXPIRATION {60};
@@ -176,6 +181,13 @@ public:
    std::vector<Sp<Value>> get(const Value::Filter& filter) const;
    Sp<Value> get(Value::Id id) const;

+    std::pair<size_t, size_t> size() const {
+        size_t tot = 0;
+        for (const auto& c : ops)
+            tot += c.second->size();
+        return {ops.size(), tot};
+    }
+
 private:
    SearchCache(const SearchCache&) = delete;
    SearchCache& operator=(const SearchCache&) = delete;
@@ -124,6 +124,14 @@ struct Dht::SearchNode {
        return last_get_reply + Node::NODE_EXPIRE_TIME;
    }

+    friend std::ostream& operator<< (std::ostream& s, const SearchNode& node) {
+        s << "getStatus:" << node.getStatus.size()
+            << " listenStatus:" << node.listenStatus.size()
+            << " acked:" << node.acked.size()
+            << " cache:" << (node.listenStatus.empty() ? 0 : node.listenStatus.begin()->second.cache.size()) << std::endl;
+        return s;
+    }
+
    /**
     * Could a particular "get" request be sent to this node now ?
     *
@@ -417,6 +425,18 @@ struct Dht::Search {
        }
    }

+    friend std::ostream& operator<< (std::ostream& s, const Search& sr) {
+        auto csize = sr.cache.size();
+        s << "announce:" << sr.announce.size()
+            << " gets:" << sr.callbacks.size()
+            << " listeners:" << sr.listeners.size()
+            << " cache:" << csize.first << ',' << csize.second << std::endl;
+        s << "nodes:" << std::endl;
+        for (const auto& n : sr.nodes)
+            s << *n;
+        return s;
+    }
+
    /**
     * @returns true if the node was not present and added to the search
     */
@@ -36,8 +36,8 @@ extern "C" {

 namespace dht {

-SecureDht::SecureDht(std::unique_ptr<DhtInterface> dht, SecureDht::Config conf)
-: dht_(std::move(dht)), key_(conf.id.first), certificate_(conf.id.second), enableCache_(conf.cert_cache_all)
+SecureDht::SecureDht(std::unique_ptr<DhtInterface> dht, SecureDht::Config conf, IdentityAnnouncedCb iacb, const std::shared_ptr<Logger>& l)
+: DhtInterface(l), dht_(std::move(dht)), key_(conf.id.first), certificate_(conf.id.second), enableCache_(conf.cert_cache_all)
 {
    if (!dht_) return;
    for (const auto& type : DEFAULT_TYPES)
@@ -52,16 +52,17 @@ SecureDht::SecureDht(std::unique_ptr<DhtInterface> dht, SecureDht::Config conf)
        auto certId = certificate_->getId();
        if (key_ and certId != key_->getPublicKey().getId())
            throw DhtException("SecureDht: provided certificate doesn't match private key.");
-
-        dht_->put(certId, Value {
-            CERTIFICATE_TYPE,
-            *certificate_,
-            1
-        }, [this, certId](bool ok) {
-            if (ok)
+        dht_->addOnConnectedCallback([this, certId, cb=std::move(iacb)]{
+            dht_->put(certId, Value {
+                CERTIFICATE_TYPE,
+                *certificate_,
+                1
+            }, [this, certId, cb=std::move(cb)](bool ok) {
+                if (cb) cb(ok);
                if (logger_)
-                    logger_->d(certId, "SecureDht: public key announced successfully");
-        }, {}, true);
+                    logger_->d(certId, "SecureDht: certificate announcement %s", ok ? "succeeded" : "failed");
+            }, {}, true);
+        });
    }
 }

@@ -496,10 +496,10 @@ Query::msgpack_unpack(const msgpack::object& o)
 }

 template <typename T>
-bool subset(std::vector<T> fds, std::vector<T> qfds)
+bool subset(const std::vector<T>& fds, const std::vector<T>& qfds)
 {
-    for (auto& fd : fds) {
-        if (std::find_if(qfds.begin(), qfds.end(), [&fd](T& _vfd) { return fd == _vfd; }) == qfds.end())
+    for (const auto& fd : fds) {
+        if (std::find_if(qfds.begin(), qfds.end(), [&fd](const T& _vfd) { return fd == _vfd; }) == qfds.end())
            return false;
    }
    return true;
@@ -142,6 +142,10 @@ public:
        }
    }

+    size_t size() const {
+        return values.size();
+    }
+
 private:
    // prevent copy
    ValueCache(const ValueCache&) = delete;
@@ -44,6 +44,10 @@ CryptoTester::testSignatureEncryption() {
    // check signature
    CPPUNIT_ASSERT(public_key.checkSignature(data1, signature1));
    CPPUNIT_ASSERT(public_key.checkSignature(data2, signature2));
+    signature1[7]++;
+    signature2[8]--;
+    CPPUNIT_ASSERT(!public_key.checkSignature(data1, signature1));
+    CPPUNIT_ASSERT(!public_key.checkSignature(data2, signature2));

    // encrypt data
    {
@@ -161,6 +165,14 @@ void CryptoTester::testCertificateSerialNumber()
    CPPUNIT_ASSERT(std::equal(SERIAL.begin(), SERIAL.end(), serial.begin(), serial.end()));
 }

+void CryptoTester::testOcsp() {
+    auto ca = dht::crypto::generateIdentity("Test CA");
+    auto device = dht::crypto::generateIdentity("Test Device", ca);
+    auto ocspRequest = device.second->generateOcspRequest(ca.second->cert);
+    auto req = dht::crypto::OcspRequest((const uint8_t*)ocspRequest.first.data(), ocspRequest.first.size());
+    CPPUNIT_ASSERT(ocspRequest.second == req.getNonce());
+}
+
 void
 CryptoTester::tearDown() {

@@ -32,6 +32,7 @@ class CryptoTester : public CppUnit::TestFixture {
    CPPUNIT_TEST(testCertificateRevocation);
    CPPUNIT_TEST(testCertificateRequest);
    CPPUNIT_TEST(testCertificateSerialNumber);
+    CPPUNIT_TEST(testOcsp);
    CPPUNIT_TEST_SUITE_END();

 public:
@@ -59,6 +60,10 @@ class CryptoTester : public CppUnit::TestFixture {
     * Test certificate serial number extraction
     */
    void testCertificateSerialNumber();
+    /**
+     * Test OCSP
+     */
+    void testOcsp();
 };

 }  // namespace test
@@ -71,7 +71,7 @@ DhtProxyTester::tearDown() {
        cv.notify_all();
    });
    std::unique_lock<std::mutex> lk(cv_m);
-    CPPUNIT_ASSERT(cv.wait_for(lk, 5s, [&]{ return done; }));
+    CPPUNIT_ASSERT(cv.wait_for(lk, 15s, [&]{ return done; }));
    serverProxy.reset();
    nodeProxy.reset();
 }
@@ -289,4 +289,47 @@ DhtProxyTester::testFuzzy()
        CPPUNIT_ASSERT(value->data == mtu);
 }

+void
+DhtProxyTester::testShutdownStop()
+{
+    constexpr size_t N = 40000;
+    constexpr unsigned C = 100;
+
+    // Arrange
+    auto key = dht::InfoHash::get("testShutdownStop");
+    std::vector<std::shared_ptr<dht::Value>> values;
+    std::vector<uint8_t> mtu;
+    mtu.reserve(N);
+    for (size_t i = 0; i < N; i++)
+        mtu.emplace_back((i % 2) ? 'T' : 'M');
+
+    std::atomic_uint callback_count {0};
+
+    // Act
+    for (size_t i = 0; i < C; i++) {
+        auto nodeTest = std::make_shared<dht::DhtRunner>();
+        nodeTest->run(0, clientConfig);
+        nodeTest->put(key, dht::Value(mtu), [&](bool ok) {
+            callback_count++;
+        });
+        nodeTest->get(key, [&](const std::vector<std::shared_ptr<dht::Value>>& vals){
+            values.insert(values.end(), vals.begin(), vals.end());
+            return true;
+        },[&](bool ok){
+            callback_count++;
+        });
+        bool done = false;
+        std::condition_variable cv;
+        std::mutex cv_m;
+        nodeTest->shutdown([&]{
+            std::lock_guard<std::mutex> lk(cv_m);
+            done = true;
+            cv.notify_all();
+        }, true);
+        std::unique_lock<std::mutex> lk(cv_m);
+        CPPUNIT_ASSERT(cv.wait_for(lk, 10s, [&]{ return done; }));
+    }
+    CPPUNIT_ASSERT_EQUAL(2*C, callback_count.load());
+}
+
 }  // namespace test
@@ -36,6 +36,7 @@ class DhtProxyTester : public CppUnit::TestFixture {
    CPPUNIT_TEST(testResubscribeGetValues);
    CPPUNIT_TEST(testPutGet40KChars);
    CPPUNIT_TEST(testFuzzy);
+    CPPUNIT_TEST(testShutdownStop);
    CPPUNIT_TEST_SUITE_END();

 public:
@@ -68,6 +69,8 @@ class DhtProxyTester : public CppUnit::TestFixture {

   void testFuzzy();

+   void testShutdownStop();
+
 private:
    dht::DhtRunner::Config clientConfig {};
    dht::DhtRunner nodePeer;
@@ -23,6 +23,7 @@
 #include <mutex>
 #include <condition_variable>
 using namespace std::chrono_literals;
+using namespace std::literals;

 namespace test {
 CPPUNIT_TEST_SUITE_REGISTRATION(DhtRunnerTester);
@@ -60,6 +61,12 @@ void
 DhtRunnerTester::testConstructors() {
    CPPUNIT_ASSERT(node1.getBoundPort() == 42222);
    CPPUNIT_ASSERT(node2.getBoundPort() == 42232);
+
+    dht::DhtRunner::Config config {};
+    dht::DhtRunner::Context context {};
+    dht::DhtRunner testNode;
+    testNode.run(config, std::move(context));
+    CPPUNIT_ASSERT(testNode.getBoundPort());
 }

 void
@@ -167,6 +174,76 @@ DhtRunnerTester::testListen() {
    node1.cancelListen(d, tokend);
 }

+void
+DhtRunnerTester::testIdOps() {
+    std::mutex mutex;
+    std::condition_variable cv;
+    unsigned valueCount(0);
+
+    dht::DhtRunner::Config config2;
+    config2.dht_config.node_config.max_peer_req_per_sec = -1;
+    config2.dht_config.node_config.max_req_per_sec = -1;
+    config2.dht_config.id = dht::crypto::generateIdentity();
+
+    dht::DhtRunner::Context context2;
+    context2.identityAnnouncedCb = [&](bool ok) {
+        CPPUNIT_ASSERT(ok);
+        std::lock_guard<std::mutex> lk(mutex);
+        valueCount++;
+        cv.notify_all();
+    };
+
+    node2.join();
+    node2.run(42232, config2, std::move(context2));
+    node2.bootstrap(node1.getBound());
+
+    node1.findCertificate(node2.getId(), [&](const std::shared_ptr<dht::crypto::Certificate>& crt){
+        CPPUNIT_ASSERT(crt);
+        std::lock_guard<std::mutex> lk(mutex);
+        valueCount++;
+        cv.notify_all();
+    });
+
+    {
+        std::unique_lock<std::mutex> lk(mutex);
+        CPPUNIT_ASSERT(cv.wait_for(lk, 20s, [&]{ return valueCount == 2; }));
+    }
+
+    dht::DhtRunner::Context context1;
+    context1.identityAnnouncedCb = [&](bool ok) {
+        CPPUNIT_ASSERT(ok);
+        std::lock_guard<std::mutex> lk(mutex);
+        valueCount++;
+        cv.notify_all();
+    };
+
+    config2.dht_config.id = dht::crypto::generateIdentity();
+    node1.join();
+    node1.run(42222, config2, std::move(context1));
+    node1.bootstrap(node2.getBound());
+
+    auto key = dht::InfoHash::get("key");
+    node1.putEncrypted(key, node2.getId(), dht::Value("yo"), [&](bool ok){
+        CPPUNIT_ASSERT(ok);
+        std::lock_guard<std::mutex> lk(mutex);
+        valueCount++;
+        cv.notify_all();
+    });
+
+    node2.listen<std::string>(key, [&](std::string&& value){
+        CPPUNIT_ASSERT_EQUAL("yo"s, value);
+        std::lock_guard<std::mutex> lk(mutex);
+        valueCount++;
+        cv.notify_all();
+        return false;
+    });
+
+    {
+        std::unique_lock<std::mutex> lk(mutex);
+        CPPUNIT_ASSERT(cv.wait_for(lk, 20s, [&]{ return valueCount == 4; }));
+    }
+}
+
 void
 DhtRunnerTester::testListenLotOfBytes() {
    std::mutex mutex;
@@ -33,6 +33,7 @@ class DhtRunnerTester : public CppUnit::TestFixture {
    CPPUNIT_TEST(testGetPut);
    CPPUNIT_TEST(testListen);
    CPPUNIT_TEST(testListenLotOfBytes);
+    CPPUNIT_TEST(testIdOps);
    CPPUNIT_TEST_SUITE_END();

    dht::DhtRunner node1 {};
@@ -58,6 +59,10 @@ class DhtRunnerTester : public CppUnit::TestFixture {
     * Test listen method
     */
    void testListen();
+    /**
+     * Test methods requiring a node identity
+     */
+    void testIdOps();
    /**
     * Test listen method with lot of datas
     */
@@ -59,7 +59,7 @@ ThreadPoolTester::testExecutor()
    auto executor8 = std::make_shared<dht::Executor>(pool, 8);

    constexpr unsigned N = 64 * 1024;
-    std::atomic_uint count1 {0};
+    unsigned count1 {0};
    std::atomic_uint count4 {0};
    std::atomic_uint count8 {0};
    for (unsigned i=0; i<N; i++) {
@@ -69,7 +69,7 @@ ThreadPoolTester::testExecutor()
    }

    auto start = clock::now();
-    while ((count1.load() != N ||
+    while ((count1 != N ||
            count4.load() != N ||
            count8.load() != N) && clock::now() - start < std::chrono::seconds(20))
    {
@@ -78,11 +78,28 @@ ThreadPoolTester::testExecutor()
    executor1.reset();
    executor4.reset();
    executor8.reset();
-    CPPUNIT_ASSERT_EQUAL(N, count1.load());
+    CPPUNIT_ASSERT_EQUAL(N, count1);
    CPPUNIT_ASSERT_EQUAL(N, count4.load());
    CPPUNIT_ASSERT_EQUAL(N, count8.load());
 }

+void
+ThreadPoolTester::testContext()
+{
+    std::atomic_uint count {0};
+    constexpr unsigned N = 64 * 1024;
+
+    {
+        dht::ExecutionContext ctx(dht::ThreadPool::computation());
+        for (unsigned i=0; i<N; i++) {
+            ctx.run([&] { count++; });
+        }
+    }
+
+    CPPUNIT_ASSERT_EQUAL(N, count.load());
+
+}
+
 void
 ThreadPoolTester::tearDown() {
 }
@@ -29,6 +29,7 @@ class ThreadPoolTester : public CppUnit::TestFixture {
    CPPUNIT_TEST_SUITE(ThreadPoolTester);
    CPPUNIT_TEST(testThreadPool);
    CPPUNIT_TEST(testExecutor);
+    CPPUNIT_TEST(testContext);
    CPPUNIT_TEST_SUITE_END();

 public:
@@ -43,6 +44,7 @@ class ThreadPoolTester : public CppUnit::TestFixture {

    void testThreadPool();
    void testExecutor();
+    void testContext();
 };

 }  // namespace test
Author	SHA1	Message	Date
Adrien Béraud	694f421a4f	thread pool: add ExecutionContext	2021-05-31 15:46:41 -04:00
Adrien Béraud	a178a8f820	dht, dhtrunner: add shutdown stop argument	2021-05-31 14:43:57 -04:00
Adrien Béraud	ec359d50fb	proxy tester: increase shutdown timeout	2021-05-31 14:43:57 -04:00
Adrien Béraud	3c5aeccd53	dhtrunner: don't read port config if already configured	2021-05-28 13:17:25 -04:00
Adrien Béraud	f99fe16a46	subset: use const ref	2021-05-28 13:17:25 -04:00
Adrien Béraud	43dc7cf222	sockaddr: fix warning	2021-05-28 13:17:25 -04:00
Sébastien Blin	ad549098f8	proxyServer: use lockListener_ to protect pushListeners_	2021-05-25 16:05:03 -04:00
Adrien Béraud	6795ba3838	cmake: fix building with peer discovery and without http	2021-05-20 15:49:11 -04:00
Adrien Béraud	efb5f8e36a	crypto: avoid generating 0 or negative certificate serial numbers	2021-05-15 01:08:07 -04:00
JUAN MÉNDEZ	8aa24f1289	Add python bindings for TypeValue Signed-off-by: JUAN MÉNDEZ <vpsink@gmail.com>	2021-05-10 14:54:16 -04:00
Sébastien Blin	14d882d220	http: add support for TCP keep alive and handle broken pipes	2021-05-03 10:09:31 -04:00
Adrien Béraud	f4c6c54a08	unit tests: add test for identityAnnouncedCb, putEncrypted	2021-04-16 16:11:02 -04:00
Adrien Béraud	4dc7a5ca0a	dhtrunner: log warning if calling run() while already running	2021-04-16 15:54:45 -04:00
Adrien Béraud	40fdd04cb3	dht: cleanup handling of onConnectCallback	2021-04-16 15:13:07 -04:00
Adrien Béraud	a24dedd225	proxy client: move callbacks to lambda	2021-04-16 15:04:19 -04:00
Adrien Béraud	ae2445fd07	proxy client: schedule connectCallbacks in callback queue	2021-04-16 15:01:23 -04:00
Adrien Béraud	7ba53cb4b3	proxy client: use unique_ptr for timers	2021-04-16 15:00:48 -04:00
Adrien Béraud	fbff8389c5	dhtrunner: build proxy with identity callback, logger	2021-04-16 14:45:33 -04:00
Adrien Béraud	827fdacc4b	securedht: build with logger	2021-04-16 14:45:14 -04:00
Adrien Béraud	3efc27e9f3	securedht: avoid use-after-free	2021-04-16 14:42:45 -04:00
Adrien Béraud	f7104199e3	securedht: fix typo	2021-04-15 12:50:31 -04:00
Adrien Béraud	5c0fbf7e68	securedht: pass IdentityAnnouncedCb by value, don't store	2021-04-15 12:47:18 -04:00
Adrien Béraud	55c595330a	securedht: use onConnect callback	2021-04-15 12:36:57 -04:00
Adrien Béraud	d3fc75589f	dht: add onConnect callback	2021-04-15 12:36:03 -04:00
Sébastien Blin	f8605f45f7	securedht: add IdentityAnnouncedCb to detect when identity is announced	2021-04-14 18:55:22 -04:00
Adrien Béraud	d541b17fe1	sockaddr: cleanup	2021-04-13 21:35:31 -04:00
Adrien Béraud	29eb1d97f8	tests: add unit test for Dhtrunner::run(config, context)	2021-04-13 21:35:31 -04:00
Adrien Béraud	623b1c4585	dhtrunner: add async version of getPublicAddress	2021-04-13 21:35:31 -04:00
Adrien Béraud	74f7d7811e	dhtrunner: use default bound address if not set	2021-04-13 21:35:31 -04:00
Adrien Béraud	7774331a20	dhtrunner: allow to provide address to bind in configuration	2021-04-13 21:35:31 -04:00
Sébastien Blin	aae30c5891	docker: update restinio version	2021-04-12 11:55:45 -04:00
Adrien Béraud	41760e4b65	ocsp response: improve exception message	2021-03-25 00:02:51 -04:00
Adrien Béraud	9ca0de250b	tests: add testOcsp	2021-03-24 23:29:53 -04:00
Adrien Béraud	d0156b4637	crypto: add OcspRequest	2021-03-24 23:29:18 -04:00
Adrien Béraud	f6cca88593	ocsp response: verifyDirect now returns revocation status	2021-03-24 23:11:59 -04:00
Adrien Béraud	c6da80573e	ocsp: fix leak	2021-03-24 23:09:24 -04:00
Adrien Béraud	5527869d13	Revert "dht: don't send unnecessary findNode" This reverts commit `0b4caeda73`.	2021-03-22 13:55:04 -04:00
Adrien Béraud	5c599b2feb	ocsp: check the appropriate field for status	2021-03-22 13:54:39 -04:00
Adrien Béraud	122d814a33	TrustRequest: add default empty parameter for conversationId	2021-03-18 17:08:54 -04:00
Adrien Béraud	a1a8adbe4b	update to version 2.2.0	2021-03-18 10:39:59 -04:00
Adrien Béraud	5a370cabb5	crypto: disable legacy encryption Require OpenDHT 2.x to decrypt encrypted data	2021-03-18 10:37:28 -04:00
Adrien Béraud	639ae26ac1	PEMCache: cleanup	2021-03-18 10:35:35 -04:00
Adrien Béraud	896b6de60c	dht: remove default constructor	2021-03-18 10:35:35 -04:00
Adrien Béraud	d025f9d1b5	crypto: add methods to sign/decrypt raw data	2021-03-18 10:35:35 -04:00
Sébastien Blin	0b029bc0bf	types: add conversationId to TrustRequest	2021-03-18 09:13:43 -04:00
Adrien Béraud	2256434ec7	dht: add inspection methods	2021-03-09 17:43:57 -05:00
Adrien Béraud	0b4caeda73	dht: don't send unnecessary findNode	2021-03-09 17:43:35 -05:00
Adrien Béraud	18a1a1c47f	callbacks: move function argument to lambda	2021-03-09 17:43:35 -05:00
Sébastien Blin	6702956521	infohash: add xor operator	2021-03-09 17:27:04 -05:00
Lucas Dias	c996220685	network_engine::process could throw an exception in message type MessageType::ValueUpdate if rsocket does not exist. processMessage called process out of try-catch	2021-01-19 12:55:02 -05:00