link ssl

dht: add override
autotools: build versioned shared library
2026-11-08 12:28:42 +00:00 · 2019-04-17 10:00:15 -04:00 · 2019-04-17 10:00:15 -04:00 · 2019-04-16 12:23:32 -04:00 · 2019-04-16 12:23:32 -04:00 · 2019-04-14 11:30:10 -04:00
32 changed files with 1138 additions and 278 deletions
@@ -46,6 +46,15 @@ Makefile.in
 .kdev4/
 *.kdev4

+# VSCode
+.vscode/*
+
+# Eclipse CDT
+.cproject
+.project
+.autotools
+language.settings.xml
+
 # Python
 *.pyc
 *.stamp
@@ -1,7 +1,7 @@
 cmake_minimum_required (VERSION 3.1)
 project (opendht)
 set (opendht_VERSION_MAJOR 1)
-set (opendht_VERSION_MINOR 8.2)
+set (opendht_VERSION_MINOR 9.1)
 set (opendht_VERSION ${opendht_VERSION_MAJOR}.${opendht_VERSION_MINOR})
 set (PACKAGE_VERSION ${opendht_VERSION})
 set (VERSION "${opendht_VERSION}")
@@ -62,6 +62,9 @@ if (OPENDHT_PROXY_SERVER OR OPENDHT_PROXY_CLIENT)
    if (NOT Jsoncpp_FOUND)
        message(SEND_ERROR "Jsoncpp is required for DHT proxy support")
    endif()
+    pkg_search_module(SSL libssl)
+    pkg_search_module(Crypto libcrypto)
+    pkg_search_module(Tls libtls)
 endif()

 # Build flags
@@ -75,6 +78,7 @@ set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -DMSGPACK_DISABLE_LEGACY_NIL -DMSGPACK_
 if (NOT CMAKE_BUILD_TYPE)
    set(CMAKE_BUILD_TYPE Release)
 endif ()
+add_definitions(-DPACKAGE_VERSION="${opendht_VERSION}")
 if (OPENDHT_LOG)
    add_definitions(-DOPENDHT_LOG=true)
 else ()
@@ -105,8 +109,20 @@ endif ()
 if (Jsoncpp_INCLUDE_DIRS)
    include_directories (SYSTEM "${Jsoncpp_INCLUDE_DIRS}")
 endif ()
+if (SSL_INCLUDE_DIRS)
+    include_directories (SYSTEM "${SSL_INCLUDE_DIRS}")
+endif ()
+if (Crypto_INCLUDE_DIRS)
+    include_directories (SYSTEM "${Crypto_INCLUDE_DIRS}")
+endif ()
+if (Tls_INCLUDE_DIRS)
+    include_directories (SYSTEM "${Tls_INCLUDE_DIRS}")
+endif ()
 link_directories (${Nettle_LIBRARY_DIRS})
 link_directories (${Jsoncpp_LIBRARY_DIRS})
+link_directories (${SSL_LIBRARY_DIRS})
+link_directories (${Crypto_LIBRARY_DIRS})
+link_directories (${Tls_LIBRARY_DIRS})
 include_directories (
    ./
    include/
@@ -149,6 +165,9 @@ list (APPEND opendht_SOURCES
    src/securedht.cpp
    src/dhtrunner.cpp
    src/log.cpp
+    src/peer_discovery.cpp
+    src/network_utils.h
+    src/network_utils.cpp
 )

 list (APPEND opendht_HEADERS
@@ -172,9 +191,14 @@ list (APPEND opendht_HEADERS
    include/opendht/securedht.h
    include/opendht/log.h
    include/opendht/log_enable.h
+    include/opendht/peer_discovery.h
    include/opendht.h
 )

+if (OPENDHT_PYTHON)
+    message("Indexation enabled since it is required for Python support")
+    set(OPENDHT_INDEX ON)
+endif()
 if (OPENDHT_INDEX)
    list (APPEND opendht_SOURCES src/indexation/pht.cpp)
    list (APPEND opendht_HEADERS include/opendht/indexation/pht.h)
@@ -246,7 +270,7 @@ if (OPENDHT_STATIC)
        target_include_directories(opendht-static SYSTEM PRIVATE ${argon2_INCLUDE_DIRS})
    endif ()
    target_link_libraries(opendht-static
-        PRIVATE ${Restbed_LIBRARY} ${argon2_LIBRARIES}
+        PRIVATE ${Restbed_LIBRARY} ${argon2_LIBRARIES} ${SSL_LIBRARIES} ${Crypto_LIBRARIES} ${Tls_LIBRARIES}
        PUBLIC ${CMAKE_THREAD_LIBS_INIT} ${GNUTLS_LIBRARIES} ${Nettle_LIBRARIES} ${Jsoncpp_LIBRARIES})
    install (TARGETS opendht-static DESTINATION ${CMAKE_INSTALL_LIBDIR} EXPORT opendht)
 endif ()
@@ -265,7 +289,7 @@ if (OPENDHT_SHARED)
        target_link_libraries(opendht PRIVATE ${argon2_LIBRARIES})
        target_include_directories(opendht SYSTEM PRIVATE ${argon2_INCLUDE_DIRS})
    endif ()
-    target_link_libraries(opendht PRIVATE ${CMAKE_THREAD_LIBS_INIT} ${GNUTLS_LIBRARIES} ${Nettle_LIBRARIES} ${Restbed_LIBRARY} ${Jsoncpp_LIBRARIES})
+    target_link_libraries(opendht PRIVATE ${CMAKE_THREAD_LIBS_INIT} ${GNUTLS_LIBRARIES} ${Nettle_LIBRARIES} ${Restbed_LIBRARY} ${Jsoncpp_LIBRARIES} ${SSL_LIBRARIES} ${Crypto_LIBRARIES} ${Tls_LIBRARIES})
    install (TARGETS opendht DESTINATION ${CMAKE_INSTALL_LIBDIR} EXPORT opendht)
 endif ()

@@ -310,6 +334,8 @@ if (OPENDHT_TESTS)
      tests/cryptotester.cpp
      tests/dhtrunnertester.h
      tests/dhtrunnertester.cpp
+      tests/peerdiscoverytester.h
+      tests/peerdiscoverytester.cpp
    )
    if (OPENDHT_PROXY_SERVER AND OPENDHT_PROXY_CLIENT)
      list (APPEND test_FILES
@@ -1,7 +1,7 @@
 dnl define macros
 m4_define([opendht_major_version], 1)
-m4_define([opendht_minor_version], 8)
-m4_define([opendht_patch_version], 2)
+m4_define([opendht_minor_version], 9)
+m4_define([opendht_patch_version], 1)
 m4_define([opendht_version],
 		  [opendht_major_version.opendht_minor_version.opendht_patch_version])

@@ -74,21 +74,21 @@ public:
    /**
     * Get the ID of the node.
     */
-    inline const InfoHash& getNodeId() const { return myid; }
+    inline const InfoHash& getNodeId() const override { return myid; }

    /**
     * Get the current status of the node for the given family.
     */
-    NodeStatus getStatus(sa_family_t af) const;
+    NodeStatus getStatus(sa_family_t af) const override;

-    NodeStatus getStatus() const {
+    NodeStatus getStatus() const override {
        return std::max(getStatus(AF_INET), getStatus(AF_INET6));
    }

    /**
     * Performs final operations before quitting.
     */
-    void shutdown(ShutdownCallback cb);
+    void shutdown(ShutdownCallback cb) override;

    /**
     * Returns true if the node is running (have access to an open socket).
@@ -96,12 +96,12 @@ public:
     *  af: address family. If non-zero, will return true if the node
     *      is running for the provided family.
     */
-    bool isRunning(sa_family_t af = 0) const;
+    bool isRunning(sa_family_t af = 0) const override;

-    virtual void registerType(const ValueType& type) {
+    virtual void registerType(const ValueType& type) override {
        types.registerType(type);
    }
-    const ValueType& getType(ValueType::Id type_id) const {
+    const ValueType& getType(ValueType::Id type_id) const override {
        return types.getType(type_id);
    }

@@ -110,18 +110,18 @@ public:
     * The node is not pinged, so this should be
     * used to bootstrap efficiently from previously known nodes.
     */
-    void insertNode(const InfoHash& id, const SockAddr&);
-    void insertNode(const InfoHash& id, const sockaddr* sa, socklen_t salen) {
+    void insertNode(const InfoHash& id, const SockAddr&) override;
+    void insertNode(const InfoHash& id, const sockaddr* sa, socklen_t salen) override {
        insertNode(id, SockAddr(sa, salen));
    }
-    void insertNode(const NodeExport& n) {
+    void insertNode(const NodeExport& n) override {
        insertNode(n.id, SockAddr(n.ss, n.sslen));
    }

-    void pingNode(const sockaddr*, socklen_t, DoneCallbackSimple&& cb={});
+    void pingNode(const sockaddr*, socklen_t, DoneCallbackSimple&& cb={}) override;

-    time_point periodic(const uint8_t *buf, size_t buflen, const SockAddr&);
-    time_point periodic(const uint8_t *buf, size_t buflen, const sockaddr* from, socklen_t fromlen) {
+    time_point periodic(const uint8_t *buf, size_t buflen, const SockAddr&) override;
+    time_point periodic(const uint8_t *buf, size_t buflen, const sockaddr* from, socklen_t fromlen) override {
        return periodic(buf, buflen, SockAddr(from, fromlen));
    }

@@ -135,14 +135,14 @@ public:
                     cb and donecb won't be called again afterward.
     * @param f a filter function used to prefilter values.
     */
-    virtual void get(const InfoHash& key, GetCallback cb, DoneCallback donecb={}, Value::Filter&& f={}, Where&& w = {});
-    virtual void get(const InfoHash& key, GetCallback cb, DoneCallbackSimple donecb={}, Value::Filter&& f={}, Where&& w = {}) {
+    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));
    }
-    virtual void get(const InfoHash& key, GetCallbackSimple cb, DoneCallback donecb={}, Value::Filter&& f={}, 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));
    }
-    virtual void get(const InfoHash& key, GetCallbackSimple cb, DoneCallbackSimple donecb, Value::Filter&& f={}, 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));
    }
    /**
@@ -155,20 +155,20 @@ public:
     * @param q a query used to filter values on the remotes before they send a
     *          response.
     */
-    virtual void query(const InfoHash& key, QueryCallback cb, DoneCallback done_cb = {}, Query&& q = {});
-    virtual void query(const InfoHash& key, QueryCallback cb, DoneCallbackSimple done_cb = {}, Query&& q = {}) {
+    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));
    }

    /**
     * Get locally stored data for the given hash.
     */
-    std::vector<Sp<Value>> getLocal(const InfoHash& key, Value::Filter f = {}) const;
+    std::vector<Sp<Value>> getLocal(const InfoHash& key, const Value::Filter& f = {}) const override;

    /**
     * Get locally stored data for the given key and value id.
     */
-    Sp<Value> getLocalById(const InfoHash& key, Value::Id vid) const;
+    Sp<Value> getLocalById(const InfoHash& key, Value::Id vid) const override;

    /**
     * Announce a value on all available protocols (IPv4, IPv6).
@@ -180,12 +180,12 @@ public:
            Sp<Value>,
            DoneCallback cb=nullptr,
            time_point created=time_point::max(),
-            bool permanent = false);
+            bool permanent = false) override;
    void put(const InfoHash& key,
            const Sp<Value>& v,
            DoneCallbackSimple cb,
            time_point created=time_point::max(),
-            bool permanent = false)
+            bool permanent = false) override
    {
        put(key, v, bindDoneCb(cb), created, permanent);
    }
@@ -194,7 +194,7 @@ public:
            Value&& v,
            DoneCallback cb=nullptr,
            time_point created=time_point::max(),
-            bool permanent = false)
+            bool permanent = false) override
    {
        put(key, std::make_shared<Value>(std::move(v)), cb, created, permanent);
    }
@@ -202,7 +202,7 @@ public:
            Value&& v,
            DoneCallbackSimple cb,
            time_point created=time_point::max(),
-            bool permanent = false)
+            bool permanent = false) override
    {
        put(key, std::forward<Value>(v), bindDoneCb(cb), created, permanent);
    }
@@ -210,18 +210,18 @@ public:
    /**
     * Get data currently being put at the given hash.
     */
-    std::vector<Sp<Value>> getPut(const InfoHash&);
+    std::vector<Sp<Value>> getPut(const InfoHash&) const override;

    /**
     * Get data currently being put at the given hash with the given id.
     */
-    Sp<Value> getPut(const InfoHash&, const Value::Id&);
+    Sp<Value> getPut(const InfoHash&, const Value::Id&) const override;

    /**
     * Stop any put/announce operation at the given location,
     * for the value with the given id.
     */
-    bool cancelPut(const InfoHash&, const Value::Id&);
+    bool cancelPut(const InfoHash&, const Value::Id&) override;

    /**
     * Listen on the network for any changes involving a specified hash.
@@ -230,28 +230,28 @@ public:
     *
     * @return a token to cancel the listener later.
     */
-    virtual size_t listen(const InfoHash&, ValueCallback, Value::Filter={}, Where={});
+    size_t listen(const InfoHash&, ValueCallback, Value::Filter={}, Where={}) override;

-    virtual size_t listen(const InfoHash& key, GetCallback cb, Value::Filter f={}, Where w={}) {
+    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){
            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={}) {
+    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));
    }

-    virtual bool cancelListen(const InfoHash&, size_t token);
+    bool cancelListen(const InfoHash&, size_t token) override;

    /**
     * Inform the DHT of lower-layer connectivity changes.
     * This will cause the DHT to assume a public IP address change.
     * The DHT will recontact neighbor nodes, re-register for listen ops etc.
     */
-    void connectivityChanged(sa_family_t);
-    void connectivityChanged() {
+    void connectivityChanged(sa_family_t) override;
+    void connectivityChanged() override {
        reported_addr.clear();
        connectivityChanged(AF_INET);
        connectivityChanged(AF_INET6);
@@ -261,18 +261,18 @@ public:
     * Get the list of good nodes for local storage saving purposes
     * The list is ordered to minimize the back-to-work delay.
     */
-    std::vector<NodeExport> exportNodes() const;
+    std::vector<NodeExport> exportNodes() const override;

-    std::vector<ValuesExport> exportValues() const;
-    void importValues(const std::vector<ValuesExport>&);
+    std::vector<ValuesExport> exportValues() const override;
+    void importValues(const std::vector<ValuesExport>&) override;

    void saveState(const std::string& path) const;
    void loadState(const std::string& path);

-    NodeStats getNodesStats(sa_family_t af) const;
+    NodeStats getNodesStats(sa_family_t af) const override;

-    std::string getStorageLog() const;
-    std::string getStorageLog(const InfoHash&) const;
+    std::string getStorageLog() const override;
+    std::string getStorageLog(const InfoHash&) const override;

    std::string getRoutingTablesLog(sa_family_t) const;
    std::string getSearchesLog(sa_family_t) const;
@@ -298,9 +298,9 @@ public:
        return {total_store_size, total_values};
    }

-    std::vector<SockAddr> getPublicAddress(sa_family_t family = 0);
+    std::vector<SockAddr> getPublicAddress(sa_family_t family = 0) override;

-    void pushNotificationReceived(const std::map<std::string, std::string>&) {}
+    void pushNotificationReceived(const std::map<std::string, std::string>&) override {}
    void resubscribe(unsigned) {}

 private:
@@ -106,7 +106,7 @@ public:
    /**
     * Get locally stored data for the given hash.
     */
-    virtual std::vector<Sp<Value>> getLocal(const InfoHash& key, Value::Filter f = {}) const = 0;
+    virtual std::vector<Sp<Value>> getLocal(const InfoHash& key, const Value::Filter& f = {}) const = 0;

    /**
     * Get locally stored data for the given key and value id.
@@ -143,12 +143,12 @@ public:
    /**
     * Get data currently being put at the given hash.
     */
-    virtual std::vector<Sp<Value>> getPut(const InfoHash&) = 0;
+    virtual std::vector<Sp<Value>> getPut(const InfoHash&) const = 0;

    /**
     * Get data currently being put at the given hash with the given id.
     */
-    virtual Sp<Value> getPut(const InfoHash&, const Value::Id&) = 0;
+    virtual Sp<Value> getPut(const InfoHash&, const Value::Id&) const = 0;

    /**
     * Stop any put/announce operation at the given location,
@@ -49,6 +49,8 @@ public:
    virtual void setPushNotificationToken(const std::string& token) {
 #ifdef OPENDHT_PUSH_NOTIFICATIONS
        deviceKey_ = token;
+#else
+        (void) token;
 #endif
    }

@@ -202,12 +204,12 @@ public:
    /**
     * Get data currently being put at the given hash.
     */
-    std::vector<Sp<Value>> getPut(const InfoHash&);
+    std::vector<Sp<Value>> getPut(const InfoHash&) const;

    /**
     * Get data currently being put at the given hash with the given id.
     */
-    Sp<Value> getPut(const InfoHash&, const Value::Id&);
+    Sp<Value> getPut(const InfoHash&, const Value::Id&) const;

    /**
     * Stop any put/announce operation at the given location,
@@ -224,7 +226,7 @@ public:
        return types.getType(type_id);
    }

-    std::vector<Sp<Value>> getLocal(const InfoHash& k, Value::Filter filter) const;
+    std::vector<Sp<Value>> getLocal(const InfoHash& k, const Value::Filter& filter) const;
    Sp<Value> getLocalById(const InfoHash& k, Value::Id id) const;

    /**
@@ -40,6 +40,7 @@ namespace dht {

 struct Node;
 class SecureDht;
+class PeerDiscovery;
 struct SecureDhtConfig;

 /**
@@ -58,6 +59,8 @@ public:
        bool threaded;
        std::string proxy_server;
        std::string push_node_id;
+        bool peer_discovery;
+        bool peer_publish;
    };

    DhtRunner();
@@ -105,7 +108,7 @@ public:
        getFilterSet<T>());
    }

-    std::future<std::vector<std::shared_ptr<dht::Value>>> get(InfoHash key, Value::Filter f = Value::AllFilter(), Where w = {}) {
+    std::future<std::vector<std::shared_ptr<dht::Value>>> get(InfoHash key, Value::Filter f = {}, Where w = {}) {
        auto p = std::make_shared<std::promise<std::vector<std::shared_ptr< dht::Value >>>>();
        auto values = std::make_shared<std::vector<std::shared_ptr< dht::Value >>>();
        get(key, [=](const std::vector<std::shared_ptr<dht::Value>>& vlist) {
@@ -136,7 +139,7 @@ public:
        query(hash, cb, bindDoneCb(done_cb), q);
    }

-    std::future<size_t> listen(InfoHash key, ValueCallback vcb, Value::Filter f = Value::AllFilter(), Where w = {});
+    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){
@@ -145,8 +148,8 @@ public:
            return true;
        }, std::forward<Value::Filter>(f), std::forward<Where>(w));
    }
-    std::future<size_t> listen(const std::string& key, GetCallback vcb, Value::Filter f = Value::AllFilter(), Where w = {});
-    std::future<size_t> listen(InfoHash key, GetCallbackSimple cb, Value::Filter f = Value::AllFilter(), Where w = {}) {
+    std::future<size_t> listen(const std::string& key, GetCallback vcb, Value::Filter f = {}, Where w = {});
+    std::future<size_t> listen(InfoHash key, GetCallbackSimple cb, Value::Filter f = {}, Where w = {}) {
        return listen(key, bindGetCb(cb), f, w);
    }

@@ -168,7 +171,7 @@ public:
    }

    template <typename T>
-    std::future<size_t> listen(InfoHash hash, std::function<bool(T&&)> cb, Value::Filter f = Value::AllFilter(), Where w = {})
+    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) {
            for (const auto& v : vals) {
@@ -184,7 +187,7 @@ public:
        getFilterSet<T>(f), w);
    }
    template <typename T>
-    std::future<size_t> listen(InfoHash hash, std::function<bool(T&&, bool)> cb, Value::Filter f = Value::AllFilter(), Where w = {})
+    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) {
            for (const auto& v : vals) {
@@ -259,6 +262,12 @@ public:
     */
    void bootstrap(const std::string& host, const std::string& service);

+    /**
+     * Insert known nodes to the routing table, without necessarly ping them.
+     * Usefull to restart a node and get things running fast without putting load on the network.
+     */
+    void bootstrap(const InfoHash& id, const SockAddr& address);
+
    /**
     * Clear the list of bootstrap added using bootstrap(const std::string&, const std::string&).
     */
@@ -361,7 +370,9 @@ public:
            },
            /*.threaded = */threaded,
            /*.proxy_server = */"",
-            /*.push_node_id = */""
+            /*.push_node_id = */"",
+            /*.peer_discovery = */true,
+            /*.peer_publish = */true,
        });
    }
    void run(in_port_t port, const Config& config);
@@ -526,6 +537,10 @@ private:

    /** Push notification token */
    std::string pushToken_;
+
+    /** PeerDiscovery Parameters */
+    std::unique_ptr<PeerDiscovery> peerDiscovery4_;
+    std::unique_ptr<PeerDiscovery> peerDiscovery6_;
 };

 }
@@ -65,7 +65,7 @@ public:
    typedef typename T::iterator iterator;
    typedef typename T::const_iterator const_iterator;

-    Hash () {
+    Hash () noexcept {
        data_.fill(0);
    }
    Hash (const uint8_t* h, size_t data_len) {
@@ -85,7 +85,7 @@ public:
        msgpack_unpack(o);
    }

-    size_t size() const { return data_.size(); }
+    static constexpr size_t size() noexcept { return N; }
    const uint8_t* data() const { return data_.data(); }
    uint8_t* data() { return data_.data(); }
    iterator begin() { return data_.begin(); }
@@ -27,6 +27,7 @@
 #include "utils.h"
 #include "rng.h"
 #include "rate_limiter.h"
+#include "log_enable.h"

 #include <vector>
 #include <string>
@@ -68,14 +69,14 @@ public:
    DhtProtocolException(uint16_t code, const std::string& msg="", InfoHash failing_node_id={})
        : DhtException(msg), msg(msg), code(code), failing_node_id(failing_node_id) {}

-    std::string getMsg() const { return msg; }
+    const std::string& getMsg() const { return msg; }
    uint16_t getCode() const { return code; }
-    const InfoHash getNodeId() const { return failing_node_id; }
+    const InfoHash& getNodeId() const { return failing_node_id; }

 private:
    std::string msg;
    uint16_t code;
-    const InfoHash failing_node_id;
+    InfoHash failing_node_id;
 };

 struct ParsedMessage;
@@ -0,0 +1,121 @@
+/*
+ *  Copyright (C) 2014-2019 Savoir-faire Linux Inc.
+ *  Author(s) : Mingrui Zhang <mingrui.zhang@savoirfairelinux.com>
+ *
+ *  This program is free software; you can redistribute it and/or modify
+ *  it under the terms of the GNU General Public License as published by
+ *  the Free Software Foundation; either version 3 of the License, or
+ *  (at your option) any later version.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+
+#pragma once
+
+#include "def.h"
+#include "sockaddr.h"
+#include "infohash.h"
+
+#include <thread>
+#include <mutex>
+#include <condition_variable>
+
+namespace dht {
+
+class OPENDHT_PUBLIC PeerDiscovery
+{
+public:
+
+    using PeerDiscoveredCallback = std::function<void(const InfoHash&, const SockAddr&)>;
+
+    PeerDiscovery(sa_family_t domain, in_port_t port);
+    ~PeerDiscovery();
+
+    /**
+     * startDiscovery - Keep Listening data from the sender until node is joinned or stop is called
+    */
+    void startDiscovery(PeerDiscoveredCallback callback);
+
+    /**
+     * startPublish - Keeping sending data until node is joinned or stop is called
+    */
+    void startPublish(const dht::InfoHash &nodeId, in_port_t port_to_send);
+
+    /**
+     * Thread Stopper
+    */
+    void stop();
+
+    /**
+     * Configure the sockopt to be able to listen multicast group
+    */
+    static void socketJoinMulticast(int sockfd, sa_family_t family);
+
+    /**
+     * Join the threads
+    */
+    void join() {
+        if(running_listen.joinable()) running_listen.join();
+        if(running_send.joinable()) running_send.join();
+    }
+
+private:
+    std::mutex mtx_;
+    std::condition_variable cv_;
+    bool running_ {true};
+    sa_family_t domain_ {AF_UNSPEC};
+    int port_;
+    int sockfd_ {-1};
+    int stop_writefd_ {-1};
+
+    SockAddr sockAddrSend_;
+    std::array<uint8_t,dht::InfoHash::size() + sizeof(in_port_t)> data_send_;
+
+    //Thread export to be joined
+    std::thread running_listen;
+    std::thread running_send;
+    dht::InfoHash nodeId_;
+
+    /**
+     * Multicast Socket Initialization, accept IPV4, IPV6
+    */
+    static int initialize_socket(sa_family_t domain);
+
+    /**
+     * Send messages
+    */
+    void sendTo(uint8_t *buf,size_t buf_size);
+
+    /**
+     * Receive messages
+    */
+    SockAddr recvFrom(uint8_t *buf, size_t &buf_size);
+
+    /**
+     * Send thread loop
+    */
+    void sender_thread();
+
+    /**
+     * Listener thread loop
+    */
+    void listener_thread(PeerDiscoveredCallback callback);
+
+    /**
+     * Listener Parameters Setup
+    */
+    void listener_setup();
+
+    /**
+     * Sender Parameters Setup
+    */
+    void sender_setup(const dht::InfoHash& nodeId, in_port_t port_to_send);
+};
+
+}
@@ -21,7 +21,6 @@
 #pragma once

 #include "utils.h"
-#include "log_enable.h"

 #include <functional>
 #include <map>
@@ -231,7 +231,7 @@ public:
    void query(const InfoHash& key, QueryCallback cb, DoneCallbackSimple done_cb = {}, Query&& q = {}) override {
        dht_->query(key, cb, done_cb, std::move(q));
    }
-    std::vector<Sp<Value>> getLocal(const InfoHash& key, Value::Filter f = Value::AllFilter()) const override {
+    std::vector<Sp<Value>> getLocal(const InfoHash& key, const Value::Filter& f = {}) const override {
        return dht_->getLocal(key, f);
    }
    Sp<Value> getLocalById(const InfoHash& key, Value::Id vid) const override {
@@ -270,10 +270,10 @@ public:
    {
        dht_->put(key, std::move(v), cb, created, permanent);
    }
-    std::vector<Sp<Value>> getPut(const InfoHash& h) override {
+    std::vector<Sp<Value>> getPut(const InfoHash& h) const override {
        return dht_->getPut(h);
    }
-    Sp<Value> getPut(const InfoHash& h, const Value::Id& vid) override {
+    Sp<Value> getPut(const InfoHash& h, const Value::Id& vid) const override {
        return dht_->getPut(h, vid);
    }
    bool cancelPut(const InfoHash& h, const Value::Id& vid) override {
@@ -23,6 +23,7 @@
 #ifndef _WIN32
 #include <sys/socket.h>
 #include <netinet/in.h>
+#include <arpa/inet.h>
 #ifdef __ANDROID__
 typedef uint16_t in_port_t;
 #endif
@@ -146,6 +147,21 @@ public:
            addr->sa_family = af;
    }

+    /**
+     * Set Network Interface to any
+     */
+    void setAny() {
+        auto family = getFamily();
+        switch(family) {
+        case AF_INET:
+            getIPv4().sin_addr.s_addr = htonl(INADDR_ANY);
+            break;
+        case AF_INET6:
+            getIPv6().sin6_addr = in6addr_any;
+            break;
+        }
+    }
+
    /**
     * Retreive the port (in host byte order) or 0 if the address is not
     * of a supported family.
@@ -175,6 +191,12 @@ public:
        }
    }

+    /**
+     * Set the address part of the socket address from a numeric IP address (string representation).
+     * Family must be already set. Throws in case of parse failue.
+     */
+    void setAddress(const char* address);
+
    /**
     * Returns the accessible byte length at the pointer returned by #get().
     * If zero, #get() returns null.
@@ -40,6 +40,8 @@ namespace dht {
 using NetId = uint32_t;
 using want_t = int_fast8_t;

+OPENDHT_PUBLIC const char* version();
+
 // shortcut for std::shared_ptr
 template<class T>
 using Sp = std::shared_ptr<T>;
@@ -2,7 +2,7 @@ lib_LTLIBRARIES = libopendht.la

 libopendht_la_CPPFLAGS = @CPPFLAGS@ -I$(top_srcdir)/include/opendht @Argon2_CFLAGS@ @JsonCpp_CFLAGS@ @MsgPack_CFLAGS@
 libopendht_la_LIBADD   = @Argon2_LIBS@ @JsonCpp_LIBS@ @GnuTLS_LIBS@ @Nettle_LIBS@
-libopendht_la_LDFLAGS  = @LDFLAGS@ @Argon2_LDFLAGS@
+libopendht_la_LDFLAGS  = @LDFLAGS@ @Argon2_LDFLAGS@ -version-number @OPENDHT_MAJOR_VERSION@:@OPENDHT_MINOR_VERSION@:@OPENDHT_PATCH_VERSION@
 libopendht_la_SOURCES  = \
        dht.cpp \
        storage.h \
@@ -26,7 +26,10 @@ libopendht_la_SOURCES  = \
        securedht.cpp \
        dhtrunner.cpp \
        default_types.cpp \
-        log.cpp
+        log.cpp \
+        peer_discovery.cpp \
+        network_utils.h \
+        network_utils.cpp

 if WIN32
 libopendht_la_SOURCES += rng.cpp
@@ -53,6 +56,7 @@ nobase_include_HEADERS = \
        ../include/opendht/default_types.h \
        ../include/opendht/log.h \
        ../include/opendht/log_enable.h \
+        ../include/opendht/peer_discovery.h \
        ../include/opendht/rng.h

 if ENABLE_PROXY_SERVER
@@ -986,19 +986,15 @@ Dht::get(const InfoHash& id, GetCallback getcb, DoneCallback donecb, Value::Filt
 {
    scheduler.syncTime();

-    auto q = std::make_shared<Query>(Select {}, std::move(where));
    auto op = std::make_shared<GetStatus<std::map<Value::Id, Sp<Value>>>>();
-    auto f = filter.chain(q->where.getFilter());
-
-    auto gcb = [getcb, donecb, op, f](const std::vector<Sp<Value>>& vals) {
+    auto gcb = [getcb, donecb, op](const std::vector<Sp<Value>>& vals) {
        auto& o = *op;
-        return callbackWrapper(getcb, donecb, vals, [&o,&f](const std::vector<Sp<Value>>& values) {
+        return callbackWrapper(getcb, donecb, vals, [&o](const std::vector<Sp<Value>>& values) {
            std::vector<Sp<Value>> newvals {};
            for (const auto& v : values) {
                auto it = o.values.find(v->id);
                if (it == o.values.cend() or (it->second != v && !(*it->second == *v))) {
-                    if (not f or f(*v))
-                        newvals.push_back(v);
+                    newvals.push_back(v);
                }
            }
            return newvals;
@@ -1008,6 +1004,9 @@ Dht::get(const InfoHash& id, GetCallback getcb, DoneCallback donecb, Value::Filt
        }, o);
    };

+    auto q = std::make_shared<Query>(Select {}, std::move(where));
+    auto f = filter.chain(q->where.getFilter());
+
    /* Try to answer this search locally. */
    gcb(getLocal(id, f));

@@ -1075,7 +1074,7 @@ void Dht::query(const InfoHash& id, QueryCallback cb, DoneCallback done_cb, Quer
 }

 std::vector<Sp<Value>>
-Dht::getLocal(const InfoHash& id, Value::Filter f) const
+Dht::getLocal(const InfoHash& id, const Value::Filter& f) const
 {
    auto s = store.find(id);
    if (s == store.end()) return {};
@@ -1092,17 +1091,14 @@ Dht::getLocalById(const InfoHash& id, Value::Id vid) const
 }

 std::vector<Sp<Value>>
-Dht::getPut(const InfoHash& id)
+Dht::getPut(const InfoHash& id) const
 {
    std::vector<Sp<Value>> ret;
-    auto find_values = [&](std::map<InfoHash, Sp<Search>> srs) {
+    auto find_values = [&](const std::map<InfoHash, Sp<Search>>& srs) {
        auto srp = srs.find(id);
-        if (srp == srs.end())
-            return;
-        auto& search = srp->second;
-        ret.reserve(ret.size() + search->announce.size());
-        for (const auto& a : search->announce)
-            ret.push_back(a.value);
+        if (srp == srs.end()) return;
+        auto vals = srp->second->getPut();
+        ret.insert(ret.end(), vals.begin(), vals.end());
    };
    find_values(searches4);
    find_values(searches6);
@@ -1110,18 +1106,11 @@ Dht::getPut(const InfoHash& id)
 }

 Sp<Value>
-Dht::getPut(const InfoHash& id, const Value::Id& vid)
+Dht::getPut(const InfoHash& id, const Value::Id& vid) const
 {
-    auto find_value = [&](std::map<InfoHash, Sp<Search>> srs) {
+    auto find_value = [&](const std::map<InfoHash, Sp<Search>>& srs) {
        auto srp = srs.find(id);
-        if (srp == srs.end())
-            return Sp<Value> {};
-        auto& search = srp->second;
-        for (auto& a : search->announce) {
-            if (a.value->id == vid)
-                return a.value;
-        }
-        return Sp<Value> {};
+        return (srp != srs.end()) ? srp->second->getPut(vid) : Sp<Value> {};
    };
    if (auto v4 = find_value(searches4))
        return v4;
@@ -1134,29 +1123,17 @@ bool
 Dht::cancelPut(const InfoHash& id, const Value::Id& vid)
 {
    bool canceled {false};
-    if (storageErase(id, vid))
-        canceled = true;
-    auto sr_cancel_put = [&](std::map<InfoHash, Sp<Search>> srs) {
+    auto sr_cancel_put = [&](std::map<InfoHash, Sp<Search>>& srs) {
        auto srp = srs.find(id);
-        if (srp == srs.end())
-            return;
-
-        auto& sr = srp->second;
-        for (auto it = sr->announce.begin(); it != sr->announce.end();) {
-            if (it->value->id == vid) {
-                canceled = true;
-                it = sr->announce.erase(it);
-            }
-            else
-                ++it;
-        }
+        return (srp != srs.end()) ? srp->second->cancelPut(vid) : false;
    };
-    sr_cancel_put(searches4);
-    sr_cancel_put(searches6);
+    canceled |= sr_cancel_put(searches4);
+    canceled |= sr_cancel_put(searches6);
+    if (canceled)
+        storageErase(id, vid);
    return canceled;
 }

-
 // Storage

 void
@@ -2016,14 +1993,15 @@ Dht::exportValues() const
 void
 Dht::importValues(const std::vector<ValuesExport>& import)
 {
-    for (const auto& h : import) {
-        if (h.second.empty())
+    const auto& now = scheduler.time();
+
+    for (const auto& node : import) {
+        if (node.second.empty())
            continue;

-        const auto& now = scheduler.time();
        try {
            msgpack::unpacked msg;
-            msgpack::unpack(msg, (const char*)h.second.data(), h.second.size());
+            msgpack::unpack(msg, (const char*)node.second.data(), node.second.size());
            auto valarr = msg.get();
            if (valarr.type != msgpack::type::ARRAY)
                throw msgpack::type_error();
@@ -2037,14 +2015,14 @@ Dht::importValues(const std::vector<ValuesExport>& import)
                    val_time = time_point{time_point::duration{valel.via.array.ptr[0].as<time_point::duration::rep>()}};
                    tmp_val.msgpack_unpack(valel.via.array.ptr[1]);
                } catch (const std::exception&) {
-                    DHT_LOG.e(h.first, "Error reading value at %s", h.first.toString().c_str());
+                    DHT_LOG.e(node.first, "Error reading value at %s", node.first.toString().c_str());
                    continue;
                }
                val_time = std::min(val_time, now);
-                storageStore(h.first, std::make_shared<Value>(std::move(tmp_val)), val_time);
+                storageStore(node.first, std::make_shared<Value>(std::move(tmp_val)), val_time);
            }
        } catch (const std::exception&) {
-            DHT_LOG.e(h.first, "Error reading values at %s", h.first.toString().c_str());
+            DHT_LOG.e(node.first, "Error reading values at %s", node.first.toString().c_str());
            continue;
        }
    }
@@ -110,7 +110,7 @@ DhtProxyClient::~DhtProxyClient()
 }

 std::vector<Sp<Value>>
-DhtProxyClient::getLocal(const InfoHash& k, Value::Filter filter) const {
+DhtProxyClient::getLocal(const InfoHash& k, const Value::Filter& filter) const {
    std::lock_guard<std::mutex> lock(searchLock_);
    auto s = searches_.find(k);
    if (s == searches_.end())
@@ -445,7 +445,7 @@ DhtProxyClient::doPut(const InfoHash& key, Sp<Value> val, DoneCallback cb, time_
 * Get data currently being put at the given hash.
 */
 std::vector<Sp<Value>>
-DhtProxyClient::getPut(const InfoHash& key) {
+DhtProxyClient::getPut(const InfoHash& key) const {
    std::vector<Sp<Value>> ret;
    auto search = searches_.find(key);
    if (search != searches_.end()) {
@@ -460,7 +460,7 @@ DhtProxyClient::getPut(const InfoHash& key) {
 * Get data currently being put at the given hash with the given id.
 */
 Sp<Value>
-DhtProxyClient::getPut(const InfoHash& key, const Value::Id& id) {
+DhtProxyClient::getPut(const InfoHash& key, const Value::Id& id) const {
    auto search = searches_.find(key);
    if (search == searches_.end())
        return {};
@@ -1037,6 +1037,8 @@ DhtProxyClient::pushNotificationReceived(const std::map<std::string, std::string
    } catch (const std::exception& e) {
        DHT_LOG.e("Error handling push notification: %s", e.what());
    }
+#else
+    (void) notification;
 #endif
 }

@@ -1066,6 +1068,9 @@ DhtProxyClient::resubscribe(const InfoHash& key, Listener& listener)
    listener.thread = std::thread([this, req, vcb, filter, state]() {
        sendListen(req, vcb, filter, state, ListenMethod::RESUBSCRIBE);
    });
+#else
+    (void) key;
+    (void) listener;
 #endif
 }

@@ -20,22 +20,15 @@

 #include "dhtrunner.h"
 #include "securedht.h"
+#include "peer_discovery.h"
+#include "network_utils.h"

 #ifdef OPENDHT_PROXY_CLIENT
 #include "dht_proxy_client.h"
 #endif

-#ifndef _WIN32
-#include <unistd.h>
-#else
-#include <io.h>
-#endif
-
-#ifndef _WIN32
-#include <sys/socket.h>
-#else
-#include <winsock2.h>
-#include <ws2tcpip.h>
+#ifdef _WIN32
+#include <cstring>
 #define close(x) closesocket(x)
 #define write(s, b, f) send(s, b, (int)strlen(b), 0)
 #endif
@@ -44,12 +37,13 @@ namespace dht {

 constexpr std::chrono::seconds DhtRunner::BOOTSTRAP_PERIOD;
 static constexpr size_t RX_QUEUE_MAX_SIZE = 1024 * 16;
+static constexpr in_port_t PEER_DISCOVERY_PORT = 8888;

 struct DhtRunner::Listener {
-    size_t tokenClassicDht;
-    size_t tokenProxyDht;
+    size_t tokenClassicDht {0};
+    size_t tokenProxyDht {0};
    ValueCallback gcb;
-    InfoHash hash;
+    InfoHash hash {};
    Value::Filter f;
    Where w;
 };
@@ -147,6 +141,35 @@ DhtRunner::run(const SockAddr& local4, const SockAddr& local6, const DhtRunner::
                cv.wait_until(lk, wakeup, hasJobToDo);
        }
    });
+
+    if (config.peer_discovery or config.peer_publish) {
+        try {
+            peerDiscovery4_.reset(new PeerDiscovery(AF_INET, PEER_DISCOVERY_PORT));
+        } catch(const std::exception& e){
+            std::cerr << "Can't start peer discovery (IPv4): " << e.what() << std::endl;
+        }
+        try {
+            peerDiscovery6_.reset(new PeerDiscovery(AF_INET6, PEER_DISCOVERY_PORT));
+        } catch(const std::exception& e) {
+            std::cerr << "Can't start peer discovery (IPv6): " << e.what() << std::endl;
+        }
+    }
+    if (config.peer_discovery) {
+        using sig = void (DhtRunner::*)(const InfoHash&, const SockAddr&);
+        if (peerDiscovery4_)
+            peerDiscovery4_->startDiscovery(std::bind(static_cast<sig>(&DhtRunner::bootstrap), this,
+                                                   std::placeholders::_1,std::placeholders::_2));
+
+        if (peerDiscovery6_)
+            peerDiscovery6_->startDiscovery(std::bind(static_cast<sig>(&DhtRunner::bootstrap), this,
+                                                    std::placeholders::_1,std::placeholders::_2));
+    }
+    if (config.peer_publish) {
+        if (peerDiscovery4_)
+            peerDiscovery4_->startPublish(dht_->getNodeId(), getBoundPort(AF_INET));
+        if (peerDiscovery6_)
+            peerDiscovery6_->startPublish(dht_->getNodeId(), getBoundPort(AF_INET6));
+    }
 }

 void
@@ -169,6 +192,9 @@ DhtRunner::join()
    running = false;
    cv.notify_all();
    bootstrap_cv.notify_all();
+    if (peerDiscovery4_) peerDiscovery4_->stop();
+    if (peerDiscovery6_) peerDiscovery6_->stop();
+
    if (dht_thread.joinable())
        dht_thread.join();
    if (bootstrap_thread.joinable())
@@ -176,6 +202,9 @@ DhtRunner::join()
    if (rcv_thread.joinable())
        rcv_thread.join();

+    if (peerDiscovery4_) peerDiscovery4_->join();
+    if (peerDiscovery6_) peerDiscovery6_->join();
+
    {
        std::lock_guard<std::mutex> lck(storage_mtx);
        pending_ops = decltype(pending_ops)();
@@ -199,17 +228,17 @@ DhtRunner::dumpTables() const
 InfoHash
 DhtRunner::getId() const
 {
-    if (!activeDht())
-        return {};
-    return activeDht()->getId();
+    if (auto dht = activeDht())
+        return dht->getId();
+    return {};
 }

 InfoHash
 DhtRunner::getNodeId() const
 {
-    if (!activeDht())
-        return {};
-    return activeDht()->getNodeId();
+    if (auto dht = activeDht())
+        return dht->getNodeId();
+    return {};
 }


@@ -458,6 +487,7 @@ int bindSocket(const SockAddr& addr, SockAddr& bound)
 #endif
    if (is_ipv6)
        setsockopt(sock, IPPROTO_IPV6, IPV6_V6ONLY, (const char*)&set, sizeof(set));
+    net::set_nonblocking(sock);
    int rc = bind(sock, addr.get(), addr.getLength());
    if(rc < 0) {
        rc = errno;
@@ -471,34 +501,6 @@ int bindSocket(const SockAddr& addr, SockAddr& bound)
    return sock;
 }

-#ifdef _WIN32
-inline void udpPipe(int fds[2])
-{
-    int lst = socket(AF_INET, SOCK_DGRAM, 0);
-    if (lst < 0)
-        throw DhtException(std::string("Can't open socket: ") + strerror(lst));
-    sockaddr_in inaddr;
-    sockaddr addr;
-    memset(&inaddr, 0, sizeof(inaddr));
-    memset(&addr, 0, sizeof(addr));
-    inaddr.sin_family = AF_INET;
-    inaddr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
-    inaddr.sin_port = 0;
-    int yes=1;
-    setsockopt(lst, SOL_SOCKET, SO_REUSEADDR, (char*)&yes, sizeof(yes));
-    int rc = bind(lst, (sockaddr*)&inaddr, sizeof(inaddr));
-    if (rc < 0) {
-        close(lst);
-        throw DhtException("Can't bind socket on " + print_addr((sockaddr*)&addr, sizeof(inaddr)) + " " + strerror(rc));
-    }
-    socklen_t len = sizeof(inaddr);
-    getsockname(lst, &addr, &len);
-    fds[0] = lst;
-    fds[1] = socket(AF_INET, SOCK_DGRAM, 0);
-    connect(fds[1], &addr, len);
-}
-#endif
-
 void
 DhtRunner::stopNetwork()
 {
@@ -524,7 +526,7 @@ DhtRunner::startNetwork(const SockAddr sin4, const SockAddr sin6)
        throw DhtException(std::string("Can't open pipe: ") + strerror(errno));
    }
 #else
-    udpPipe(stopfds);
+    net::udpPipe(stopfds);
 #endif
    int stop_readfd = stopfds[0];
    stop_writefd = stopfds[1];
@@ -569,7 +571,8 @@ DhtRunner::startNetwork(const SockAddr sin4, const SockAddr sin6)
                if(s6 >= 0)
                    FD_SET(s6, &readfds);

-                int rc = select(s4 > s6 ? s4 + 1 : s6 + 1, &readfds, nullptr, nullptr, nullptr);
+                int selectFd = std::max({s4, s6, stop_readfd}) + 1;
+                int rc = select(selectFd, &readfds, nullptr, nullptr, nullptr);
                if(rc < 0) {
                    if(errno != EINTR) {
                        perror("select");
@@ -591,6 +594,7 @@ DhtRunner::startNetwork(const SockAddr sin4, const SockAddr sin6)
                        rc = recvfrom(s6, (char*)buf.data(), buf.size(), 0, (sockaddr*)&from, &from_len);
                    else
                        continue;
+
                    if (rc > 0) {
                        {
                            std::lock_guard<std::mutex> lck(sock_mtx);
@@ -660,22 +664,21 @@ DhtRunner::listen(InfoHash hash, ValueCallback vcb, Value::Filter f, Where w)
        pending_ops.emplace([=](SecureDht& dht) mutable {
 #ifdef OPENDHT_PROXY_CLIENT
            auto tokenbGlobal = listener_token_++;
-            Listener listener {};
+            auto& listener = listeners_[tokenbGlobal];
            listener.hash = hash;
            listener.f = std::move(f);
            listener.w = std::move(w);
-            listener.gcb = [hash,vcb,tokenbGlobal,this](const std::vector<Sp<Value>>& vals, bool expired){
+            listener.gcb = [hash,vcb,tokenbGlobal,this](const std::vector<Sp<Value>>& vals, bool expired) {
                if (not vcb(vals, expired)) {
                    cancelListen(hash, tokenbGlobal);
                    return false;
                }
                return true;
            };
-            if (use_proxy)
-                listener.tokenProxyDht = dht.listen(hash, listener.gcb, listener.f, listener.w);
-            else
-                listener.tokenClassicDht = dht.listen(hash, listener.gcb, listener.f, listener.w);
-            listeners_.emplace(tokenbGlobal, std::move(listener));
+            if (auto token = dht.listen(hash, listener.gcb, listener.f, listener.w)) {
+                if (use_proxy)  listener.tokenProxyDht = token;
+                else            listener.tokenClassicDht = token;
+            }
            ret_token->set_value(tokenbGlobal);
 #else
            ret_token->set_value(dht.listen(hash, std::move(vcb), std::move(f), std::move(w)));
@@ -934,6 +937,18 @@ DhtRunner::bootstrap(const SockAddr& addr, DoneCallbackSimple&& cb)
    cv.notify_all();
 }

+void
+DhtRunner::bootstrap(const InfoHash& id, const SockAddr& address)
+{
+    {
+        std::unique_lock<std::mutex> lck(storage_mtx);
+        pending_ops_prio.emplace([id, address](SecureDht& dht) mutable {
+            dht.insertNode(id, address);
+        });
+    }
+    cv.notify_all();
+}
+
 void
 DhtRunner::bootstrap(const std::vector<NodeExport>& nodes)
 {
@@ -973,6 +988,8 @@ DhtRunner::findCertificate(InfoHash hash, std::function<void(const std::shared_p
 void
 DhtRunner::resetDht()
 {
+    peerDiscovery4_.reset();
+    peerDiscovery6_.reset();
 #ifdef OPENDHT_PROXY_CLIENT
    listeners_.clear();
    dht_via_proxy_.reset();
@@ -1066,6 +1083,8 @@ DhtRunner::forwardAllMessages(bool forward)
 #endif // OPENDHT_PROXY_CLIENT
    if (dht_)
        dht_->forwardAllMessages(forward);
+#else
+    (void) forward;
 #endif // OPENDHT_PROXY_SERVER
 }

@@ -1078,6 +1097,8 @@ DhtRunner::setPushNotificationToken(const std::string& token) {
    pushToken_ = token;
    if (dht_via_proxy_)
        dht_via_proxy_->setPushNotificationToken(token);
+#else
+    (void) token;
 #endif
 }

@@ -1093,6 +1114,8 @@ DhtRunner::pushNotificationReceived(const std::map<std::string, std::string>& da
        });
    }
    cv.notify_all();
+#else
+    (void) data;
 #endif
 }

@@ -17,7 +17,6 @@
 *  along with this program. If not, see <https://www.gnu.org/licenses/>.
 */

-
 #include "network_engine.h"
 #include "request.h"
 #include "default_types.h"
@@ -26,17 +25,6 @@

 #include <msgpack.hpp>

-#ifndef _WIN32
-#include <arpa/inet.h>
-#include <unistd.h>
-#else
-#include <ws2tcpip.h>
-#include <io.h>
-#endif
-#include <fcntl.h>
-
-#include <cstring>
-
 namespace dht {
 namespace net {

@@ -55,38 +43,12 @@ constexpr std::chrono::seconds NetworkEngine::RX_TIMEOUT;
 const std::string NetworkEngine::my_v {"RNG1"};
 constexpr size_t NetworkEngine::MAX_REQUESTS_PER_SEC;

-static const uint8_t v4prefix[16] = {
+static constexpr uint8_t v4prefix[16] = {
    0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xFF, 0xFF, 0, 0, 0, 0
 };

 constexpr unsigned SEND_NODES {8};

-#ifdef _WIN32
-
-static bool
-set_nonblocking(int fd, int nonblocking)
-{
-    unsigned long mode = !!nonblocking;
-    int rc = ioctlsocket(fd, FIONBIO, &mode);
-    return rc == 0;
-}
-
-extern const char *inet_ntop(int, const void *, char *, socklen_t);
-
-#else
-
-static bool
-set_nonblocking(int fd, int nonblocking)
-{
-    int rc = fcntl(fd, F_GETFL, 0);
-    if (rc < 0)
-        return false;
-    rc = fcntl(fd, F_SETFL, nonblocking?(rc | O_NONBLOCK):(rc & ~O_NONBLOCK));
-    return rc >= 0;
-}
-
-#endif
-

 /* Transaction-ids are 4-bytes long, with the first two bytes identifying
 * the kind of request, and the remaining two a sequence number in
@@ -163,16 +125,7 @@ NetworkEngine::NetworkEngine(InfoHash& myid, NetId net, const int& s, const int&
    onAnnounce(std::move(onAnnounce)),
    onRefresh(std::move(onRefresh)),
    myid(myid), network(net), dht_socket(s), dht_socket6(s6), DHT_LOG(log), scheduler(scheduler)
-{
-    if (dht_socket >= 0) {
-        if (!set_nonblocking(dht_socket, 1))
-            throw DhtException("Can't set socket to non-blocking mode");
-    }
-    if (dht_socket6 >= 0) {
-        if (!set_nonblocking(dht_socket6, 1))
-            throw DhtException("Can't set socket to non-blocking mode");
-    }
-}
+{}

 NetworkEngine::~NetworkEngine() {
    clear();
@@ -0,0 +1,81 @@
+/*
+ *  Copyright (C) 2019 Savoir-faire Linux Inc.
+ *  Author(s) : Adrien Béraud <adrien.beraud@savoirfairelinux.com>
+ *
+ *  This program is free software; you can redistribute it and/or modify
+ *  it under the terms of the GNU General Public License as published by
+ *  the Free Software Foundation; either version 3 of the License, or
+ *  (at your option) any later version.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+
+#include "network_utils.h"
+
+#ifdef _WIN32
+#include "utils.h"
+#include "sockaddr.h"
+#include <io.h>
+#include <string>
+#include <cstring>
+#define close(x) closesocket(x)
+#define write(s, b, f) send(s, b, (int)strlen(b), 0)
+#else
+#include <fcntl.h>
+#endif
+
+namespace dht {
+namespace net {
+
+bool
+set_nonblocking(int fd, bool nonblocking)
+{
+#ifdef _WIN32
+    unsigned long mode = !!nonblocking;
+    int rc = ioctlsocket(fd, FIONBIO, &mode);
+    return rc == 0;
+#else
+    int rc = fcntl(fd, F_GETFL, 0);
+    if (rc < 0)
+        return false;
+    rc = fcntl(fd, F_SETFL, nonblocking ? (rc | O_NONBLOCK) : (rc & ~O_NONBLOCK));
+    return rc >= 0;
+#endif
+}
+
+#ifdef _WIN32
+void udpPipe(int fds[2])
+{
+    int lst = socket(AF_INET, SOCK_DGRAM, 0);
+    if (lst < 0)
+        throw DhtException(std::string("Can't open socket: ") + strerror(WSAGetLastError()));
+    sockaddr_in inaddr;
+    sockaddr addr;
+    memset(&inaddr, 0, sizeof(inaddr));
+    memset(&addr, 0, sizeof(addr));
+    inaddr.sin_family = AF_INET;
+    inaddr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+    inaddr.sin_port = 0;
+    int yes = 1;
+    setsockopt(lst, SOL_SOCKET, SO_REUSEADDR, (char*)&yes, sizeof(yes));
+    int rc = bind(lst, (sockaddr*)&inaddr, sizeof(inaddr));
+    if (rc < 0) {
+        close(lst);
+        throw DhtException("Can't bind socket on " + print_addr((sockaddr*)&inaddr, sizeof(inaddr)) + " " + strerror(rc));
+    }
+    socklen_t len = sizeof(addr);
+    getsockname(lst, &addr, &len);
+    fds[0] = lst;
+    fds[1] = socket(AF_INET, SOCK_DGRAM, 0);
+    connect(fds[1], &addr, len);
+}
+#endif
+
+}
+}
@@ -0,0 +1,43 @@
+/*
+ *  Copyright (C) 2019 Savoir-faire Linux Inc.
+ *  Author(s) : Adrien Béraud <adrien.beraud@savoirfairelinux.com>
+ *
+ *  This program is free software; you can redistribute it and/or modify
+ *  it under the terms of the GNU General Public License as published by
+ *  the Free Software Foundation; either version 3 of the License, or
+ *  (at your option) any later version.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+#pragma once
+
+#ifdef _WIN32
+#include <ws2tcpip.h>
+#include <winsock2.h>
+#else
+#include <sys/socket.h>
+#include <netinet/in.h>
+#include <unistd.h>
+#endif
+
+#ifndef IPV6_JOIN_GROUP
+#define IPV6_JOIN_GROUP IPV6_ADD_MEMBERSHIP
+#endif
+
+namespace dht {
+namespace net {
+
+bool set_nonblocking(int fd, bool nonblocking = true);
+
+#ifdef _WIN32
+void udpPipe(int fds[2]);
+#endif
+
+}
+}
@@ -84,13 +84,13 @@ struct ParsedMessage {
    /* target id around which to find nodes */
    InfoHash target;
    /* transaction id */
-    Tid tid;
+    Tid tid {0};
    /* tid for packets going through request socket */
-    Tid socket_id;
+    Tid socket_id {0};
    /* security token */
    Blob token;
    /* the value id (announce confirmation) */
-    Value::Id value_id;
+    Value::Id value_id {0};
    /* time when value was first created */
    time_point created { time_point::max() };
    /* IPv4 nodes in response to a 'find' request */
@@ -0,0 +1,317 @@
+/*
+ *  Copyright (C) 2014-2019 Savoir-faire Linux Inc.
+ *  Author(s) : Mingrui Zhang <mingrui.zhang@savoirfairelinux.com>
+ *
+ *  This program is free software; you can redistribute it and/or modify
+ *  it under the terms of the GNU General Public License as published by
+ *  the Free Software Foundation; either version 3 of the License, or
+ *  (at your option) any later version.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+
+#include "peer_discovery.h"
+#include "network_utils.h"
+
+#ifdef _WIN32
+#include <Ws2tcpip.h> // needed for ip_mreq definition for multicast
+#include <Windows.h>
+#include <cstring>
+#define close(x) closesocket(x)
+#define write(s, b, f) send(s, b, (int)strlen(b), 0)
+#else
+#include <sys/types.h>
+#include <unistd.h>
+#endif
+#include <fcntl.h>
+
+namespace dht {
+
+constexpr char MULTICAST_ADDRESS_IPV4[10] = "224.0.0.1";
+constexpr char MULTICAST_ADDRESS_IPV6[8] = "ff05::2"; // Site-local multicast
+
+PeerDiscovery::PeerDiscovery(sa_family_t domain, in_port_t port)
+    : domain_(domain), port_(port), sockfd_(initialize_socket(domain))
+{
+    socketJoinMulticast(sockfd_, domain);
+}
+
+PeerDiscovery::~PeerDiscovery()
+{
+    if (sockfd_ != -1)
+        close(sockfd_);
+
+#ifdef _WIN32
+    WSACleanup();
+#endif
+}
+
+int
+PeerDiscovery::initialize_socket(sa_family_t domain)
+{
+#ifdef _WIN32
+    WSADATA wsaData;
+    if (WSAStartup(0x0101, &wsaData)) {
+        throw std::runtime_error(std::string("Can't initialize Winsock2 ") + strerror(errno));
+    }
+#endif
+
+    int sockfd = socket(domain, SOCK_DGRAM, 0);
+    if (sockfd < 0) {
+        throw std::runtime_error(std::string("Socket Creation Error: ") + strerror(errno));
+    }
+    net::set_nonblocking(sockfd);
+    return sockfd;
+}
+
+void
+PeerDiscovery::listener_setup()
+{
+    SockAddr sockAddrListen_;
+    sockAddrListen_.setFamily(domain_);
+    sockAddrListen_.setPort(port_);
+    sockAddrListen_.setAny();
+
+    unsigned int opt = 1;
+    if (setsockopt(sockfd_, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) < 0) {
+       std::cerr << "setsockopt SO_REUSEADDR failed: " << strerror(errno) << std::endl;
+    }
+    if (setsockopt(sockfd_, SOL_SOCKET, SO_REUSEPORT, &opt, sizeof(opt)) < 0) {
+       std::cerr << "setsockopt SO_REUSEPORT failed: " << strerror(errno) << std::endl;
+    }
+
+    // bind to receive address
+    if (bind(sockfd_, sockAddrListen_.get(), sockAddrListen_.getLength()) < 0){
+        throw std::runtime_error(std::string("Error binding socket: ") + strerror(errno));
+    }
+}
+
+void
+PeerDiscovery::socketJoinMulticast(int sockfd, sa_family_t family)
+{
+    switch (family)
+    {
+    case AF_INET:{
+        ip_mreq config_ipv4;
+
+        //This option can be used to set the interface for sending outbound
+        //multicast datagrams from the sockets application.
+        config_ipv4.imr_interface.s_addr = htonl(INADDR_ANY);
+        if( setsockopt(sockfd, IPPROTO_IP, IP_MULTICAST_IF, &config_ipv4.imr_interface, sizeof( struct in_addr )) < 0 ) {
+            throw std::runtime_error(std::string("Bound Network Interface IPv4 Error: ") + strerror(errno));
+        }
+
+        //The IP_MULTICAST_TTL socket option allows the application to primarily
+        //limit the lifetime of the packet in the Internet and prevent it from circulating indefinitely
+        unsigned char ttl4 = 20;
+        if( setsockopt(sockfd, IPPROTO_IP, IP_MULTICAST_TTL, &ttl4, sizeof( ttl4 )) < 0 ) {
+            throw std::runtime_error(std::string("TTL Sockopt Error: ") + strerror(errno));
+        }
+
+        // config the listener to be interested in joining in the multicast group
+        config_ipv4.imr_multiaddr.s_addr = inet_addr(MULTICAST_ADDRESS_IPV4);
+        config_ipv4.imr_interface.s_addr = htonl(INADDR_ANY);
+        if (setsockopt(sockfd, IPPROTO_IP, IP_ADD_MEMBERSHIP, (char*)&config_ipv4, sizeof(config_ipv4)) < 0){
+            throw std::runtime_error(std::string(" Member Addition IPv4 Error: ") + strerror(errno));
+        }
+        break;
+    }
+    case AF_INET6: {
+        ipv6_mreq config_ipv6;
+
+        /* unsigned int outif = 0;
+        if (setsockopt(sockfd, IPPROTO_IPV6, IPV6_MULTICAST_IF, &outif, sizeof(outif)) < 0) {
+            std::cerr << "Can't assign multicast interface: " << strerror(errno) << std::endl;
+        } */
+
+        unsigned int ttl6 = 20;
+        if( setsockopt(sockfd, IPPROTO_IPV6, IPV6_MULTICAST_HOPS, &ttl6, sizeof( ttl6 )) < 0 ) {
+            throw std::runtime_error(std::string("Hop Count Set Error: ") + strerror(errno));
+        }
+
+        config_ipv6.ipv6mr_interface = 0;
+        inet_pton(AF_INET6, MULTICAST_ADDRESS_IPV6, &config_ipv6.ipv6mr_multiaddr);
+        if (setsockopt(sockfd, IPPROTO_IPV6, IPV6_JOIN_GROUP, &config_ipv6, sizeof(config_ipv6)) < 0){
+            throw std::runtime_error(std::string("Member Addition IPv6 Error: ") + strerror(errno));
+        }
+        break;
+    }
+    }
+}
+
+void
+PeerDiscovery::sendTo(uint8_t *buf, size_t buf_size)
+{
+    ssize_t nbytes = sendto(
+        sockfd_,
+        buf,
+        buf_size,
+        0,
+        sockAddrSend_.get(),
+        sockAddrSend_.getLength()
+    );
+    if (nbytes < 0) {
+        throw std::runtime_error(std::string("Error sending packet: ") + strerror(errno));
+    }
+}
+
+SockAddr
+PeerDiscovery::recvFrom(uint8_t *buf, size_t& buf_size)
+{
+    sockaddr_storage storeage_recv;
+    socklen_t sa_len = sizeof(storeage_recv);
+
+    ssize_t nbytes = recvfrom(
+        sockfd_,
+        buf,
+        buf_size,
+        0,
+        (sockaddr*)&storeage_recv,
+        &sa_len
+    );
+    if (nbytes < 0) {
+        throw std::runtime_error(std::string("Error receiving packet: ") + strerror(errno));
+    }
+
+    buf_size = nbytes;
+    SockAddr ret {storeage_recv, sa_len};
+    return ret;
+}
+
+void
+PeerDiscovery::sender_setup(const dht::InfoHash& nodeId, in_port_t port_to_send)
+{
+    nodeId_ = nodeId;
+    // Setup sender address
+    sockAddrSend_.setFamily(domain_);
+    sockAddrSend_.setAddress(domain_ == AF_INET ? MULTICAST_ADDRESS_IPV4 : MULTICAST_ADDRESS_IPV6);
+    sockAddrSend_.setPort(port_);
+
+    // Setup sent data
+    std::copy_n(nodeId.cbegin(), nodeId.size(), data_send_.begin());
+    auto portAddr = reinterpret_cast<in_port_t*>(data_send_.data() + dht::InfoHash::size());
+    *portAddr = htons(port_to_send);
+}
+
+void
+PeerDiscovery::sender_thread()
+{
+    while(true) {
+        try {
+            sendTo(data_send_.data(), data_send_.size());
+        } catch (const std::exception& e) {
+            std::cerr << e.what() << std::endl;
+        }
+        {
+            std::unique_lock<std::mutex> lck(mtx_);
+            if (cv_.wait_for(lck,std::chrono::seconds(3),[&]{ return !running_; }))
+                break;
+        }
+    }
+}
+
+void
+PeerDiscovery::listener_thread(PeerDiscoveredCallback callback)
+{
+    int stopfds_pipe[2];
+#ifndef _WIN32
+    if (pipe(stopfds_pipe) == -1)
+        throw std::runtime_error(std::string("Can't open pipe: ") + strerror(errno));
+#else
+    net::udpPipe(stopfds_pipe);
+#endif
+    int stop_readfd = stopfds_pipe[0];
+    stop_writefd_ = stopfds_pipe[1];
+
+    while(true) {
+        fd_set readfds;
+
+        FD_ZERO(&readfds);
+        FD_SET(stop_readfd, &readfds);
+        FD_SET(sockfd_, &readfds);
+
+        int data_coming = select(sockfd_ > stop_readfd ? sockfd_ + 1 : stop_readfd + 1, &readfds, nullptr, nullptr, nullptr);
+
+        {
+            std::unique_lock<std::mutex> lck(mtx_);
+            if (not running_)
+                break;
+        }
+
+        if (data_coming < 0) {
+            if(errno != EINTR) {
+                perror("Select Error");
+                std::this_thread::sleep_for( std::chrono::seconds(1) );
+            }
+        }
+
+        if (data_coming > 0) {
+            if (FD_ISSET(stop_readfd, &readfds)) {
+                std::array<uint8_t, 64 * 1024> buf;
+                recv(stop_readfd, buf.data(), buf.size(), 0);
+            }
+
+            std::array<uint8_t,dht::InfoHash::size() + sizeof(in_port_t)> data_receive;
+            size_t data_receive_size = data_receive.size();
+            auto from = recvFrom(data_receive.data(), data_receive_size);
+
+            // Data_receive_size as a value-result member will hlep to filter packs
+            if(data_receive_size != data_receive.size()){
+                // std::cerr << "Received invalid peer discovery packet" << std::endl;
+                continue;
+            }
+
+            dht::InfoHash nodeId;
+            std::copy_n(data_receive.begin(), dht::InfoHash::size(), nodeId.begin());
+            auto portAddr = reinterpret_cast<in_port_t*>(data_receive.data() + dht::InfoHash::size());
+            auto port = ntohs(*portAddr);
+            if (nodeId != nodeId_){
+                from.setPort(port);
+                callback(nodeId, from);
+            }
+        }
+    }
+    if (stop_readfd != -1)
+        close(stop_readfd);
+    if (stop_writefd_ != -1) {
+        close(stop_writefd_);
+        stop_writefd_ = -1;
+    }
+}
+
+void
+PeerDiscovery::startDiscovery(PeerDiscoveredCallback callback)
+{
+    listener_setup();
+    running_listen = std::thread(&PeerDiscovery::listener_thread, this, callback);
+}
+
+void
+PeerDiscovery::startPublish(const dht::InfoHash& nodeId, in_port_t port_to_send)
+{
+    sender_setup(nodeId, port_to_send);
+    running_send = std::thread(&PeerDiscovery::sender_thread, this);
+}
+
+void
+PeerDiscovery::stop()
+{
+    {
+        std::unique_lock<std::mutex> lck(mtx_);
+        running_ = false;
+    }
+    cv_.notify_all();
+    if (stop_writefd_ != -1) {
+        if (write(stop_writefd_, "\0", 1) == -1) {
+            perror("write");
+        }
+    }
+}
+
+}
@@ -532,6 +532,35 @@ struct Dht::Search {
        scheduler.edit(opExpirationJob, cache.getExpiration());
    }

+    std::vector<Sp<Value>> getPut() const {
+        std::vector<Sp<Value>> ret;
+        ret.reserve(announce.size());
+        for (const auto& a : announce)
+            ret.push_back(a.value);
+        return ret;
+    }
+
+    Sp<Value> getPut(Value::Id vid) const {
+        for (auto& a : announce) {
+            if (a.value->id == vid)
+                return a.value;
+        }
+        return {};
+    }
+
+    bool cancelPut(Value::Id vid) {
+        bool canceled {false};
+        for (auto it = announce.begin(); it != announce.end();) {
+            if (it->value->id == vid) {
+                canceled = true;
+                it = announce.erase(it);
+            }
+            else
+                ++it;
+        }
+        return canceled;
+    }
+
    /**
     * @return The number of non-good search nodes.
     */
@@ -115,7 +115,7 @@ struct Storage {
        return {};
    }

-    std::vector<Sp<Value>> get(Value::Filter f = {}) const {
+    std::vector<Sp<Value>> get(const Value::Filter& f = {}) const {
        std::vector<Sp<Value>> newvals {};
        if (not f) newvals.reserve(values.size());
        for (auto& v : values) {
@@ -25,10 +25,18 @@
 #define IN_IS_ADDR_UNSPECIFIED(a) (((long int) (a)->s_addr) == 0x00000000)
 #endif /* IN_IS_ADDR_UNSPECIFIED */

+#ifndef PACKAGE_VERSION
+#define PACKAGE_VERSION "(unknown version)"
+#endif
+
 namespace dht {

 static constexpr std::array<uint8_t, 12> MAPPED_IPV4_PREFIX {{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xff, 0xff}};

+const char* version() {
+    return PACKAGE_VERSION;
+}
+
 std::pair<std::string, std::string>
 splitPort(const std::string& s) {
    if (s.empty())
@@ -73,6 +81,22 @@ SockAddr::resolve(const std::string& host, const std::string& service)
    return ips;
 }

+void
+SockAddr::setAddress(const char* address)
+{
+    auto family = getFamily();
+    void* addr = nullptr;
+    switch (family) {
+    case AF_INET:
+        addr = &getIPv4().sin_addr;
+        break;
+    case AF_INET6:
+        addr = &getIPv6().sin6_addr;
+        break;
+    }
+    if (not addr or inet_pton(family, address, addr) <= 0)
+        throw std::runtime_error("Can't parse IP address");
+}

 std::string
 print_addr(const sockaddr* sa, socklen_t slen)
@@ -160,20 +160,25 @@ DhtProxyTester::testResubscribeGetValues() {

    // For the second subscribe, the proxy will return the value in the body
    auto values = std::vector<dht::Blob>();
-    nodeClient->listen(key, [&](const std::vector<std::shared_ptr<dht::Value>>& v, bool) {
-        for (const auto& value : v)
-            values.emplace_back(value->data);
-        done = true;
+    auto ftoken = nodeClient->listen(key, [&](const std::vector<std::shared_ptr<dht::Value>>& v, bool expired) {
+        if (not expired) {
+            std::lock_guard<std::mutex> lk(cv_m);
+            for (const auto& value : v)
+                values.emplace_back(value->data);
+            done = true;
+        }
        cv.notify_all();
        return true;
    });

    cv.wait_for(lk, std::chrono::seconds(10), [&]{ return done; });
+    auto token = ftoken.get();
+    CPPUNIT_ASSERT(token);
+    nodeClient->cancelListen(key, token);
    done = false;
    // Here values should still contains 1 values
-    CPPUNIT_ASSERT_EQUAL(static_cast<int>(values.size()), 1);
-    CPPUNIT_ASSERT(values.front() == firstVal_data);
-
+    CPPUNIT_ASSERT_EQUAL((size_t)1u, values.size());
+    CPPUNIT_ASSERT(firstVal_data == values.front());
 }

 }  // namespace test
@@ -19,9 +19,9 @@

 #include "dhtrunnertester.h"

-// std
-#include <iostream>
-#include <string>
+#include <chrono>
+#include <mutex>
+#include <condition_variable>

 namespace test {
 CPPUNIT_TEST_SUITE_REGISTRATION(DhtRunnerTester);
@@ -62,44 +62,73 @@ DhtRunnerTester::testGetPut() {

 void
 DhtRunnerTester::testListen() {
+    std::mutex mutex;
+    std::condition_variable cv;
    std::atomic_uint valueCount(0);
-    std::atomic_uint putCount(0);
+    unsigned putCount(0);
+    unsigned putOkCount(0);

    auto a = dht::InfoHash::get("234");
    auto b = dht::InfoHash::get("2345");
    auto c = dht::InfoHash::get("23456");
    constexpr unsigned N = 32;

-    auto ftokena = node1.listen(a, [&](const std::shared_ptr<dht::Value>&){
+    auto ftokena = node1.listen(a, [&](const std::shared_ptr<dht::Value>&) {
        valueCount++;
        return true;
    });

-    auto ftokenb = node1.listen(b, [&](const std::shared_ptr<dht::Value>&){
+    auto ftokenb = node1.listen(b, [&](const std::shared_ptr<dht::Value>&) {
        valueCount++;
        return false;
    });

-    auto ftokenc = node1.listen(c, [&](const std::shared_ptr<dht::Value>&){
+    auto ftokenc = node1.listen(c, [&](const std::shared_ptr<dht::Value>&) {
        valueCount++;
        return true;
    });

    for (unsigned i=0; i<N; i++) {
-        node2.put(a, dht::Value("v1"), [&](bool ok) { if (ok) putCount++; });
-        node2.put(b, dht::Value("v2"), [&](bool ok) { if (ok) putCount++; });
+        node2.put(a, dht::Value("v1"), [&](bool ok) {
+            {
+                std::lock_guard<std::mutex> lock(mutex);
+                putCount++;
+                if (ok) putOkCount++;
+            }
+            cv.notify_all();
+        });
+        node2.put(b, dht::Value("v2"), [&](bool ok) {
+            {
+                std::lock_guard<std::mutex> lock(mutex);
+                putCount++;
+                if (ok) putOkCount++;
+            }
+            cv.notify_all();
+        });
    }

-    std::this_thread::sleep_for(std::chrono::milliseconds(100));
+    {
+        std::unique_lock<std::mutex> lk(mutex);
+        cv.wait_for(lk, std::chrono::seconds(30), [&]{ return putCount == N * 2u; });
+        CPPUNIT_ASSERT_EQUAL(N * 2u, putCount);
+        CPPUNIT_ASSERT_EQUAL(N * 2u, putOkCount);
+    }
+
+    CPPUNIT_ASSERT(ftokena.valid());
+    CPPUNIT_ASSERT(ftokenb.valid());
+    CPPUNIT_ASSERT(ftokenc.valid());
+
    auto tokena = ftokena.get();
-    auto tokenb = ftokenb.get();
    auto tokenc = ftokenc.get();
+    // tokenb might be 0 since the callback returns false.

    CPPUNIT_ASSERT(tokena);
-    CPPUNIT_ASSERT(tokenb);
    CPPUNIT_ASSERT(tokenc);
-    CPPUNIT_ASSERT_EQUAL(N * 2u, putCount.load());
    CPPUNIT_ASSERT_EQUAL(N + 1u, valueCount.load());
+
+    node1.cancelListen(a, tokena);
+    node1.cancelListen(b, std::move(ftokenb));
+    node1.cancelListen(c, tokenc);
 }

 }  // namespace test
@@ -0,0 +1,91 @@
+/*
+ *  Copyright (C) 2019 Savoir-faire Linux Inc.
+ *
+ *  Author: Mingrui Zhang <mingrui.zhang@savoirfairelinux.com>
+ *
+ *  This program is free software; you can redistribute it and/or modify
+ *  it under the terms of the GNU General Public License as published by
+ *  the Free Software Foundation; either version 3 of the License, or
+ *  (at your option) any later version.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+
+#include "peerdiscoverytester.h"
+
+namespace test {
+
+CPPUNIT_TEST_SUITE_REGISTRATION(PeerDiscoveryTester);
+
+void PeerDiscoveryTester::setUp(){}
+
+void PeerDiscoveryTester::testTransmission_ipv4(){
+
+    // Node for getnode id
+    dht::InfoHash data_n = dht::InfoHash::get("applepin");
+    int port = 2222;
+    in_port_t port_n = 50000;
+    try{
+        dht::PeerDiscovery test_n(AF_INET, port);
+        dht::PeerDiscovery test_s(AF_INET, port);
+        try{
+            test_s.startDiscovery([&](const dht::InfoHash& node, const dht::SockAddr& addr){
+                CPPUNIT_ASSERT_EQUAL(data_n, node);
+                CPPUNIT_ASSERT_EQUAL(port_n, addr.getPort());
+            });
+
+            test_n.startPublish(data_n,port_n);
+
+            std::this_thread::sleep_for(std::chrono::seconds(5));
+            test_n.stop();
+            test_s.stop();
+            test_n.join();
+            test_s.join();
+        } catch(std::exception &exception){
+            perror(exception.what());
+            CPPUNIT_ASSERT(false);
+        }
+    } catch(std::exception &exception){
+            perror(exception.what());
+    }
+}
+
+void PeerDiscoveryTester::testTransmission_ipv6(){
+
+    // Node for getnode id
+    dht::InfoHash data_n = dht::InfoHash::get("applepin");
+    int port = 3333;
+    in_port_t port_n = 50001;
+    try{
+        dht::PeerDiscovery test_n(AF_INET6, port);
+        dht::PeerDiscovery test_s(AF_INET6, port);
+
+        try{
+            test_s.startDiscovery([&](const dht::InfoHash& node, const dht::SockAddr& addr){
+                CPPUNIT_ASSERT_EQUAL(data_n, node);
+                CPPUNIT_ASSERT_EQUAL(port_n, addr.getPort());
+            });
+
+            test_n.startPublish(data_n,port_n);
+
+            std::this_thread::sleep_for(std::chrono::seconds(5));
+            test_n.stop();
+            test_s.stop();
+            test_n.join();
+            test_s.join();
+        } catch(std::exception &exception){
+            CPPUNIT_ASSERT(false);
+        }
+    } catch(std::exception &exception) {
+    }
+}
+
+void PeerDiscoveryTester::tearDown(){}
+
+}  // namespace test
@@ -0,0 +1,56 @@
+/*
+ *  Copyright (C) 2019 Savoir-faire Linux Inc.
+ *
+ *  Author: Mingrui Zhang <mingrui.zhang@savoirfairelinux.com>
+ *
+ *  This program is free software; you can redistribute it and/or modify
+ *  it under the terms of the GNU General Public License as published by
+ *  the Free Software Foundation; either version 3 of the License, or
+ *  (at your option) any later version.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+
+#pragma once
+
+#include "opendht/peer_discovery.h"
+
+#include <cppunit/TestFixture.h>
+#include <cppunit/extensions/HelperMacros.h>
+
+namespace test {
+
+class PeerDiscoveryTester : public CppUnit::TestFixture {
+
+    CPPUNIT_TEST_SUITE(PeerDiscoveryTester);
+    CPPUNIT_TEST(testTransmission_ipv4);
+    CPPUNIT_TEST(testTransmission_ipv6);
+    CPPUNIT_TEST_SUITE_END();
+
+ public:
+    /**
+     * Method automatically called before each test by CppUnit
+     */
+    void setUp();
+    /**
+     * Method automatically called after each test CppUnit
+     */
+    void tearDown();
+    /**
+     * Test Multicast Transmission Ipv4
+     */
+    void testTransmission_ipv4();
+    /**
+     * Test Multicast Transmission Ipv6
+     */
+    void testTransmission_ipv6();
+
+};
+
+}  // namespace test
@@ -29,24 +29,33 @@ extern "C" {

 using namespace dht;

-void print_usage() {
-    std::cout << "Usage: dhtnode [-v [-l logfile]] [-i] [-d] [-n network_id] [-p local_port] [-b bootstrap_host[:port]] [--proxyserver local_port]" << std::endl << std::endl;
+void print_info() {
    std::cout << "dhtnode, a simple OpenDHT command line node runner." << std::endl;
    std::cout << "Report bugs to: https://opendht.net" << std::endl;
 }

+void print_version() {
+    std::cout << "OpenDHT version " << dht::version() << std::endl;
+    print_info();
+}
+
+void print_usage() {
+    std::cout << "Usage: dhtnode [-v [-l logfile]] [-i] [-d] [-n network_id] [-p local_port] [-b bootstrap_host[:port]] [--proxyserver local_port]" << std::endl << std::endl;
+    print_info();
+}
+
 void print_id_req() {
    std::cout << "An identity is required to perform this operation (run with -i)" << std::endl;
 }

 void print_node_info(const std::shared_ptr<DhtRunner>& dht, const dht_params& params) {
-    std::cout << "OpenDht node " << dht->getNodeId() << " running on port " <<  dht->getBoundPort() << std::endl;
+    std::cout << "OpenDHT node " << dht->getNodeId() << " running on port " <<  dht->getBoundPort() << std::endl;
    if (params.generate_identity)
        std::cout << "Public key ID " << dht->getId() << std::endl;
 }

 void print_help() {
-    std::cout << "OpenDht command line interface (CLI)" << std::endl;
+    std::cout << "OpenDHT command line interface (CLI)" << std::endl;
    std::cout << "Possible commands:" << std::endl
              << "  h, help    Print this help message." << std::endl
              << "  x, quit    Quit the program." << std::endl
@@ -484,6 +493,10 @@ main(int argc, char **argv)
            print_usage();
            return 0;
        }
+        if (params.version) {
+            print_version();
+            return 0;
+        }

        if (params.daemonize) {
            daemonize();
@@ -505,6 +518,8 @@ main(int argc, char **argv)
        config.threaded = true;
        config.proxy_server = params.proxyclient;
        config.push_node_id = "dhtnode";
+        config.peer_discovery = params.peer_discovery;
+        config.peer_publish = params.peer_discovery;
        if (not params.proxyclient.empty())
            dht->setPushNotificationToken(params.devicekey);

@@ -90,6 +90,7 @@ static const constexpr in_port_t DHT_DEFAULT_PORT = 4222;

 struct dht_params {
    bool help {false}; // print help and exit
+    bool version {false};
    bool log {false};
    std::string logfile {};
    bool syslog {false};
@@ -98,6 +99,7 @@ struct dht_params {
    bool generate_identity {false};
    bool daemonize {false};
    bool service {false};
+    bool peer_discovery {false};
    std::pair<std::string, std::string> bootstrap {};
    in_port_t proxyserver {0};
    std::string proxyclient {};
@@ -107,29 +109,31 @@ struct dht_params {
 };

 static const constexpr struct option long_options[] = {
-   {"help",       no_argument      , nullptr, 'h'},
-   {"port",       required_argument, nullptr, 'p'},
-   {"net",        required_argument, nullptr, 'n'},
-   {"bootstrap",  required_argument, nullptr, 'b'},
-   {"identity",   no_argument      , nullptr, 'i'},
-   {"verbose",    no_argument      , nullptr, 'v'},
-   {"daemonize",  no_argument      , nullptr, 'd'},
-   {"service",    no_argument      , nullptr, 's'},
-   {"persist",    required_argument, nullptr, 'f'},
-   {"logfile",    required_argument, nullptr, 'l'},
-   {"syslog",     no_argument      , nullptr, 'L'},
-   {"proxyserver",required_argument, nullptr, 'S'},
-   {"proxyclient",required_argument, nullptr, 'C'},
-   {"pushserver", required_argument, nullptr, 'P'},
-   {"devicekey",  required_argument, nullptr, 'D'},
-   {nullptr,      0                , nullptr,  0}
+   {"help",             no_argument      , nullptr, 'h'},
+   {"port",             required_argument, nullptr, 'p'},
+   {"net",              required_argument, nullptr, 'n'},
+   {"bootstrap",        required_argument, nullptr, 'b'},
+   {"identity",         no_argument      , nullptr, 'i'},
+   {"verbose",          no_argument      , nullptr, 'v'},
+   {"daemonize",        no_argument      , nullptr, 'd'},
+   {"service",          no_argument      , nullptr, 's'},
+   {"peer-discovery",   no_argument      , nullptr, 'D'},
+   {"persist",          required_argument, nullptr, 'f'},
+   {"logfile",          required_argument, nullptr, 'l'},
+   {"syslog",           no_argument      , nullptr, 'L'},
+   {"proxyserver",      required_argument, nullptr, 'S'},
+   {"proxyclient",      required_argument, nullptr, 'C'},
+   {"pushserver",       required_argument, nullptr, 'y'},
+   {"devicekey",        required_argument, nullptr, 'z'},
+   {"version",          no_argument      , nullptr, 'V'},
+   {nullptr,            0                , nullptr,  0}
 };

 dht_params
 parseArgs(int argc, char **argv) {
    dht_params params;
    int opt;
-    while ((opt = getopt_long(argc, argv, "hidsvp:n:b:f:l:", long_options, nullptr)) != -1) {
+    while ((opt = getopt_long(argc, argv, "hidsvDp:n:b:f:l:", long_options, nullptr)) != -1) {
        switch (opt) {
        case 'p': {
                int port_arg = atoi(optarg);
@@ -147,13 +151,16 @@ parseArgs(int argc, char **argv) {
                    std::cout << "Invalid port: " << port_arg << std::endl;
            }
            break;
-        case 'P':
+        case 'D':
+            params.peer_discovery = true;
+            break;
+        case 'y':
            params.pushserver = optarg;
            break;
        case 'C':
            params.proxyclient = optarg;
            break;
-        case 'D':
+        case 'z':
            params.devicekey = optarg;
            break;
        case 'f':
@@ -168,6 +175,9 @@ parseArgs(int argc, char **argv) {
                params.bootstrap.second = std::to_string(DHT_DEFAULT_PORT);
            }
            break;
+        case 'V':
+            params.version = true;
+            break;
        case 'h':
            params.help = true;
            break;
Author	SHA1	Message	Date
Adrien Béraud	b6a58ce784	link ssl	2019-04-17 10:00:15 -04:00
Adrien Béraud	5ec72e213e	dht: add override	2019-04-17 10:00:15 -04:00
Adrien Béraud	683a111726	autotools: build versioned shared library	2019-04-16 12:23:32 -04:00
Adrien Béraud	0fd4fed537	search: add fallback return	2019-04-16 12:23:32 -04:00
Adrien Béraud	ef82652a1e	network: header cleanup	2019-04-14 11:30:10 -04:00
Adrien Béraud	4f03a5c64e	network: tid, socket_id, value_id: default to 0	2019-04-14 09:15:16 -04:00
Adrien Béraud	5257ac314f	add API to retrieve version string	2019-04-14 00:19:01 -04:00
Adrien Béraud	0580a4e7dd	update to version 1.9.1	2019-04-13 23:29:43 -04:00
Yanfei Han	fd1f2cd0e4	add peer_discovery.o object into auto configure lib	2019-04-13 23:04:18 -04:00
Adrien Béraud	af940c63b1	proxy: mute unused variable warning if disabled	2019-04-13 23:02:57 -04:00
Adrien Béraud	15da502354	dhtrunner: make peer discovery port static constexpr	2019-04-13 22:56:31 -04:00
Adrien Béraud	b7a9a52f97	network engine: DhtProtocolException cleanup	2019-04-13 22:56:31 -04:00
Adrien Béraud	6e10756e28	peerdiscovery: cleanup	2019-04-13 22:56:31 -04:00
Adrien Béraud	aad4ad03cf	network utils: add missing includes	2019-04-13 22:56:31 -04:00
Adrien Béraud	3d1e055c7b	peerdiscovery: always use local node id	2019-04-13 20:50:00 -04:00
Adrien Béraud	b02e9cfff7	peerdiscovery: flush stop socket if not stopping	2019-04-13 20:50:00 -04:00
Adrien Béraud	453db29058	dht: add override to getNodeId	2019-04-13 20:43:57 -04:00
Adrien Béraud	a8ba536d44	update to version 1.9.0	2019-04-09 16:25:39 -04:00
Adrien Béraud	a238183031	peerdiscovery: don't assign IPv6 multicast interface	2019-04-09 16:11:21 -04:00
Adrien Béraud	73827cd34d	peerdiscovery: allow IPV6_MULTICAST_IF failure	2019-04-09 15:12:37 -04:00
Adrien Béraud	876190a392	peerdiscovery: SO_REUSEADDR, SO_REUSEPORT in different calls	2019-04-09 14:58:22 -04:00
Adrien Béraud	7fe321e4fc	peerdiscovery: cleanup data serialization	2019-04-09 14:57:48 -04:00
Adrien Béraud	7f7075ca9a	peerdiscovery: support IPV6_JOIN_GROUP	2019-04-09 12:31:23 -04:00
Adrien Béraud	9b9f12d87e	tests: accept null token if callback returns false	2019-04-08 15:58:46 -04:00
Adrien Béraud	26ad6de3a4	dhtrunner: avoid race condition when setting token	2019-04-08 15:57:25 -04:00
Adrien Béraud	ae91f8bd55	dhtrunner: default-initialize tokens	2019-04-08 15:36:24 -04:00
Adrien Béraud	28eeee24ba	runnertester: reduce number of test values	2019-04-08 15:35:55 -04:00
Adrien Béraud	671c55a01d	dhtrunner: print peer discovery error message to cerr	2019-04-08 14:36:15 -04:00
Adrien Béraud	a4f2e6efdd	tests: avoid use-after-free in proxytester	2019-04-08 13:06:32 -04:00
Adrien Béraud	60831ee898	peerdiscovery: avoid crashing if message can't be sent	2019-04-08 13:02:39 -04:00
Adrien Béraud	98dbe1cba7	gitignore: ignore files from vscode, eclipse	2019-04-08 12:07:35 -04:00
Adrien Béraud	6c705e525f	net: factorize logic to network_utils	2019-04-08 12:07:11 -04:00
Adrien Béraud	96347d83e1	tools: add argument for peer discovery	2019-04-08 11:34:52 -04:00
Adrien Béraud	01d8e513a1	peerdiscovery: cleanup	2019-04-08 11:34:52 -04:00
Mingrui Zhang	b5921b0d9d	peerdiscovery: error correction for tester, distinguish creation error(ip version support) and listen/receive error	2019-04-08 11:32:08 -04:00
Mingrui Zhang	8ee7883c53	peerdiscovery: Exception added for supporting construction error	2019-04-08 11:32:08 -04:00
Mingrui Zhang	4240820463	peerdiscovery: add multicast node discovery mechanism	2019-04-08 11:32:08 -04:00
Adrien Béraud	7ebbf17f37	InfoHash: use static size	2019-04-05 11:21:23 -04:00
Adrien Béraud	6821e5bd96	dhtrunner: default to null function instead of AllFilter	2019-04-05 10:49:40 -04:00
Adrien Béraud	c4ec1e78e0	tests: fix occasional testListen failure	2019-04-05 10:49:35 -04:00
Adrien Béraud	e86d717e81	dht: make getPut() const	2019-04-05 10:49:35 -04:00
Adrien Béraud	db2b48d2af	dhtrunner: select on maxfd+1	2019-04-03 16:18:35 -04:00
Adrien Béraud	2b938110fd	cmake: force indexation if python is enabled	2019-04-03 14:52:35 -04:00
Adrien Béraud	c647b19e91	infohash: make some methods constexpr, noexcept	2019-04-03 14:52:35 -04:00
Sébastien Blin	8cd14646a6	Merge pull request #387 from savoirfairelinux/cert_request crypto: add certificate request support	2019-04-01 09:30:33 -04:00