build: add light mode

Inside the if-statement, the log message will always print zero.
Confusingly, its the same message as the one in the loop above it.
Make the message distinct.

.github/workflows/ccpp.yml

@@ -13,12 +13,19 @@ jobs:
       run: |
         sudo apt install libncurses5-dev libreadline-dev nettle-dev \
         libgnutls28-dev libuv1-dev cython3 python3-dev python3-setuptools libcppunit-dev libjsoncpp-dev \
-        autotools-dev autoconf libasio-dev libfmt-dev libhttp-parser-dev libmsgpack-dev
+        autotools-dev autoconf libfmt-dev libhttp-parser-dev libmsgpack-dev
     - name: autogen
       run: ./autogen.sh
     - name: argon2
       run: |
         cd argon2 && make && sudo make install && cd ..
+    - name: asio
+      run: |
+        wget https://github.com/aberaud/asio/archive/b2b7a1c166390459e1c169c8ae9ef3234b361e3f.tar.gz \
+        && tar -xvf b2b7a1c166390459e1c169c8ae9ef3234b361e3f.tar.gz && cd asio-b2b7a1c166390459e1c169c8ae9ef3234b361e3f/asio \
+        && ./autogen.sh && ./configure --prefix=/usr --without-boost --disable-examples --disable-tests  \
+        && sudo make install \
+        && cd ../../ && rm -rf asio*
     - name: configure
       run: ./configure
     - name: make

CMakeLists.txt

@@ -7,83 +7,105 @@ set (PACKAGE_VERSION ${opendht_VERSION})
 set (VERSION "${opendht_VERSION}")
 
 # Options
+option (OPENDHT_LIGHT "Build minimal version of the library" OFF)
+
 option (OPENDHT_STATIC "Build static library" ON)
+
 option (OPENDHT_SHARED "Build shared library" ON)
-option (OPENDHT_LOG "Build with logs" ON)
 option (OPENDHT_PYTHON "Build Python bindings" OFF)
 option (OPENDHT_TOOLS "Build DHT tools" ON)
 option (OPENDHT_SYSTEMD "Install systemd module" OFF)
 option (OPENDHT_ARGON2 "Use included argon2 sources" OFF)
-option (OPENDHT_LTO "Build with LTO" OFF)
-option (OPENDHT_SANITIZE "Build with address sanitizer and stack protector" OFF)
 option (OPENDHT_PROXY_SERVER "Enable DHT proxy server, use Restinio and jsoncpp" OFF)
 option (OPENDHT_PUSH_NOTIFICATIONS "Enable push notifications support" OFF)
 option (OPENDHT_PROXY_SERVER_IDENTITY "Allow clients to use the node identity" OFF)
 option (OPENDHT_PROXY_CLIENT "Enable DHT proxy client, use Restinio and jsoncpp" OFF)
 option (OPENDHT_PROXY_OPENSSL "Build DHT proxy with OpenSSL" ON)
 option (OPENDHT_PROXY_HTTP_PARSER_FORK "Build DHT proxy with custom http_parser to support old API" OFF)
+option (OPENDHT_PEER_DISCOVERY "Enable multicast peer discovery" ON)
 option (OPENDHT_INDEX "Build DHT indexation feature" OFF)
-option (OPENDHT_TESTS "Add unit tests executable" OFF)
 option (OPENDHT_C "Build C bindings" OFF)
+option (OPENDHT_TESTS "Add unit tests executable" OFF)
+
+if (OPENDHT_LIGHT)
+    set(OPENDHT_SHARED OFF)
+    set(OPENDHT_TOOLS OFF)
+    set(OPENDHT_PROXY_SERVER OFF)
+    set(OPENDHT_PROXY_CLIENT OFF)
+    set(OPENDHT_PEER_DISCOVERY OFF)
+    set(OPENDHT_INDEX OFF)
+    set(OPENDHT_PYTHON OFF)
+    add_definitions(-DOPENDHT_LIGHT)
+endif()
+
+option (OPENDHT_LOG "Build with logs" ON)
+option (OPENDHT_LTO "Build with LTO" OFF)
+option (OPENDHT_SANITIZE "Build with address sanitizer and stack protector" OFF)
 
 find_package(Doxygen)
 option (OPENDHT_DOCUMENTATION "Create and install the HTML based API documentation (requires Doxygen)" ${DOXYGEN_FOUND})
 
 # Dependencies
 list (APPEND CMAKE_MODULE_PATH "${PROJECT_SOURCE_DIR}/cmake")
-find_package (Threads)
 find_package (PkgConfig)
-find_package (GnuTLS 3.3 REQUIRED)
-pkg_search_module (Nettle nettle)
 find_package (Msgpack 1.2 REQUIRED)
-if (OPENDHT_TOOLS)
-    find_package (Readline 6 REQUIRED)
-endif ()
-if (NOT OPENDHT_ARGON2)
-    find_package(PkgConfig)
-    pkg_search_module(argon2 libargon2)
-    if (argon2_FOUND)
-        message("-- Found Argon2: " ${argon2_LIBRARY_DIRS} " (found version \"" ${argon2_VERSION} "\")")
-        link_directories (${argon2_LIBRARY_DIRS})
-    else ()
-        message("Argon2 not found, using included version.")
-        set(OPENDHT_ARGON2 ON)
-    endif()
-endif ()
-
-pkg_search_module(Jsoncpp jsoncpp)
-if (Jsoncpp_FOUND)
-    add_definitions(-DOPENDHT_JSONCPP)
-    list (APPEND opendht_SOURCES
-      src/base64.h
-      src/base64.cpp
-    )
-endif()
-
-if (OPENDHT_PROXY_SERVER OR OPENDHT_PROXY_CLIENT)
-    find_package(Restinio REQUIRED)
-    if (Restinio_FOUND)
-        add_definitions(-DASIO_STANDALONE)
-        find_library(FMT_LIBRARY fmt)
-        add_library(fmt SHARED IMPORTED)
-        find_library(HTTP_PARSER_LIBRARY http_parser)
-        add_library(http_parser SHARED IMPORTED)
-    endif()
-    if (NOT Jsoncpp_FOUND)
-        message(SEND_ERROR "Jsoncpp is required for DHT proxy support")
-    endif()
-    if (OPENDHT_PROXY_OPENSSL)
-        # https://cmake.org/cmake/help/latest/module/FindOpenSSL.html
-        pkg_search_module(OPENSSL REQUIRED openssl)
-        if (OPENSSL_FOUND)
-            message(STATUS "Found OpenSSL ${OPENSSL_VERSION}")
-            include_directories(${OPENSSL_INCLUDE_DIRS})
+if (NOT OPENDHT_LIGHT)
+    find_package (Threads)
+    find_package (GnuTLS 3.3 REQUIRED)
+    pkg_search_module (Nettle nettle)
+    if (OPENDHT_TOOLS)
+        find_package (Readline 6 REQUIRED)
+    endif ()
+    if (NOT OPENDHT_ARGON2)
+        find_package(PkgConfig)
+        pkg_search_module(argon2 libargon2)
+        if (argon2_FOUND)
+            message("-- Found Argon2: " ${argon2_LIBRARY_DIRS} " (found version \"" ${argon2_VERSION} "\")")
+            link_directories (${argon2_LIBRARY_DIRS})
         else ()
-            message(SEND_ERROR "OpenSSL is required for DHT proxy as specified")
+            message("Argon2 not found, using included version.")
+            set(OPENDHT_ARGON2 ON)
         endif()
+    endif ()
+
+    pkg_search_module(Jsoncpp jsoncpp)
+    if (Jsoncpp_FOUND)
+        add_definitions(-DOPENDHT_JSONCPP)
+        list (APPEND opendht_SOURCES
+        src/base64.h
+        src/base64.cpp
+        )
     endif()
-    if (OPENDHT_PROXY_HTTP_PARSER_FORK)
-        add_definitions(-DOPENDHT_PROXY_HTTP_PARSER_FORK)
+
+    if (OPENDHT_PROXY_SERVER OR OPENDHT_PROXY_CLIENT)
+        find_package(Restinio REQUIRED)
+        if (Restinio_FOUND)
+            find_library(FMT_LIBRARY fmt)
+            add_library(fmt SHARED IMPORTED)
+            find_library(HTTP_PARSER_LIBRARY http_parser)
+            add_library(http_parser SHARED IMPORTED)
+        endif()
+        if (NOT Jsoncpp_FOUND)
+            message(SEND_ERROR "Jsoncpp is required for DHT proxy support")
+        endif()
+        if (OPENDHT_PROXY_OPENSSL)
+            # https://cmake.org/cmake/help/latest/module/FindOpenSSL.html
+            pkg_search_module(OPENSSL REQUIRED openssl)
+            if (OPENSSL_FOUND)
+                message(STATUS "Found OpenSSL ${OPENSSL_VERSION}")
+                include_directories(${OPENSSL_INCLUDE_DIRS})
+            else ()
+                message(SEND_ERROR "OpenSSL is required for DHT proxy as specified")
+            endif()
+        endif()
+        if (OPENDHT_PROXY_HTTP_PARSER_FORK)
+            add_definitions(-DOPENDHT_PROXY_HTTP_PARSER_FORK)
+        endif()
+    else()
+        set(OPENDHT_PROXY_OPENSSL OFF)
+    endif()
+    if (OPENDHT_PROXY_SERVER OR OPENDHT_PROXY_CLIENT OR OPENDHT_PEER_DISCOVERY)
+        add_definitions(-DASIO_STANDALONE)
     endif()
 endif()
 
@@ -156,7 +178,6 @@ set (top_srcdir "${CMAKE_CURRENT_SOURCE_DIR}")
 list (APPEND opendht_SOURCES
     src/utils.cpp
     src/infohash.cpp
-    src/crypto.cpp
     src/default_types.cpp
     src/node.cpp
     src/value.cpp
@@ -174,12 +195,7 @@ list (APPEND opendht_SOURCES
     src/routing_table.cpp
     src/node_cache.cpp
     src/network_engine.cpp
-    src/securedht.cpp
-    src/dhtrunner.cpp
     src/log.cpp
-    src/peer_discovery.cpp
-    src/network_utils.cpp
-    src/thread_pool.cpp
 )
 
 list (APPEND opendht_HEADERS
@@ -187,7 +203,6 @@ list (APPEND opendht_HEADERS
     include/opendht/utils.h
     include/opendht/sockaddr.h
     include/opendht/rng.h
-    include/opendht/crypto.h
     include/opendht/infohash.h
     include/opendht/default_types.h
     include/opendht/node.h
@@ -203,12 +218,33 @@ list (APPEND opendht_HEADERS
     include/opendht/securedht.h
     include/opendht/log.h
     include/opendht/log_enable.h
-    include/opendht/peer_discovery.h
-    include/opendht/thread_pool.h
-    include/opendht/network_utils.h
     include/opendht.h
 )
 
+if (NOT OPENDHT_LIGHT)
+    list (APPEND opendht_SOURCES
+        src/crypto.cpp
+        src/securedht.cpp
+        src/dhtrunner.cpp
+        src/network_utils.cpp
+        src/thread_pool.cpp
+    )
+
+    list (APPEND opendht_HEADERS
+        include/opendht/crypto.h
+        include/opendht/securedht.h
+        include/opendht/dhtrunner.h
+        include/opendht/thread_pool.h
+        include/opendht/network_utils.h
+    )
+endif()
+
+if (OPENDHT_PEER_DISCOVERY)
+    list (APPEND opendht_SOURCES src/peer_discovery.cpp)
+    list (APPEND opendht_HEADERS include/opendht/peer_discovery.h)
+    add_definitions(-DOPENDHT_PEER_DISCOVERY)
+endif()
+
 if (OPENDHT_PYTHON)
     message("Indexation enabled since it is required for Python support")
     set(OPENDHT_INDEX ON)
@@ -322,7 +358,12 @@ if (OPENDHT_C)
         c/opendht_c.h
     )
     target_compile_definitions(opendht-c PRIVATE OPENDHT_C_BUILD)
-    target_link_libraries(opendht-c opendht)
+    if (OPENDHT_SHARED)
+        target_link_libraries(opendht-c opendht)
+    else ()
+        target_link_libraries(opendht-c opendht-static)
+    endif ()
+
     # PkgConfig module
     configure_file (
         opendht-c.pc.in
@@ -370,15 +411,17 @@ if (OPENDHT_TESTS)
         tests/infohashtester.cpp
         tests/valuetester.h
         tests/valuetester.cpp
-        tests/cryptotester.h
-        tests/cryptotester.cpp
-        tests/dhtrunnertester.h
-        tests/dhtrunnertester.cpp
-        tests/peerdiscoverytester.h
-        tests/peerdiscoverytester.cpp
-        tests/threadpooltester.h
-        tests/threadpooltester.cpp
     )
+    if (NOT OPENDHT_LIGHT)
+        list (APPEND test_FILES
+            tests/cryptotester.h
+            tests/cryptotester.cpp
+            tests/dhtrunnertester.h
+            tests/dhtrunnertester.cpp
+            tests/threadpooltester.h
+            tests/threadpooltester.cpp
+        )
+    endif()
     if (OPENDHT_PROXY_SERVER AND OPENDHT_PROXY_CLIENT)
         list (APPEND test_FILES
             tests/httptester.h
@@ -387,6 +430,12 @@ if (OPENDHT_TESTS)
             tests/dhtproxytester.cpp
         )
     endif()
+    if (OPENDHT_PEER_DISCOVERY)
+        list (APPEND test_FILES
+            tests/peerdiscoverytester.h
+            tests/peerdiscoverytester.cpp
+        )
+    endif()
     add_executable(opendht_unit_tests
         tests/tests_runner.cpp
         ${test_FILES}
@@ -401,8 +450,11 @@ if (OPENDHT_TESTS)
        ${CMAKE_THREAD_LIBS_INIT}
        ${CPPUNIT_LIBRARIES}
        ${GNUTLS_LIBRARIES}
-       ssl crypto jsoncpp
+       ${Jsoncpp_LIBRARIES}
     )
+    if (OPENDHT_PROXY_OPENSSL)
+        target_link_libraries(opendht_unit_tests ${OPENSSL_LIBRARIES})
+    endif()
     enable_testing()
     add_test(TEST opendht_unit_tests)
 endif()

c/opendht.cpp

@@ -122,7 +122,7 @@ void dht_publickey_delete(dht_publickey* pk) {
     delete reinterpret_cast<PubkeySp*>(pk);
 }
 
-int dht_publickey_pack(dht_publickey* pk, char* out, size_t* outlen) {
+int dht_publickey_export(const dht_publickey* pk, char* out, size_t* outlen) {
     const auto& pkey = *reinterpret_cast<const PubkeySp*>(pk);
     return gnutls_pubkey_export(pkey->pk, GNUTLS_X509_FMT_DER, out, outlen);
 }
@@ -164,6 +164,13 @@ dht_privatekey* dht_privatekey_import(const uint8_t* dat, size_t dat_size, const
     }
 }
 
+int dht_privatekey_export(const dht_privatekey* k, char* out, size_t* out_size, const char* password) {
+    if (!out or !out_size or !*out_size)
+        return -1;
+    const auto& key = *reinterpret_cast<const PrivkeySp*>(k);
+    return key->serialize((uint8_t*)out, out_size, password);
+}
+
 dht_publickey* dht_privatekey_get_publickey(const dht_privatekey* k) {
     const auto& key = *reinterpret_cast<const PrivkeySp*>(k);
     return reinterpret_cast<dht_publickey*>(new PubkeySp(std::make_shared<dht::crypto::PublicKey>(key->getPublicKey())));

c/opendht_c.h

@@ -45,7 +45,7 @@ struct OPENDHT_C_PUBLIC dht_publickey;
 typedef struct dht_publickey dht_publickey;
 OPENDHT_C_PUBLIC dht_publickey* dht_publickey_import(const uint8_t* dat, size_t dat_size);
 OPENDHT_C_PUBLIC void dht_publickey_delete(dht_publickey* pk);
-OPENDHT_C_PUBLIC int dht_publickey_pack(dht_publickey* pk, char* out, size_t* out_size);
+OPENDHT_C_PUBLIC int dht_publickey_export(const dht_publickey* pk, char* out, size_t* out_size);
 OPENDHT_C_PUBLIC dht_infohash dht_publickey_get_id(const dht_publickey* pk);
 OPENDHT_C_PUBLIC dht_pkid dht_publickey_get_long_id(const dht_publickey* pk);
 OPENDHT_C_PUBLIC bool dht_publickey_check_signature(const dht_publickey* pk, const char* data, size_t data_size, const char* signature, size_t signature_size);
@@ -56,6 +56,7 @@ struct OPENDHT_C_PUBLIC dht_privatekey;
 typedef struct dht_privatekey dht_privatekey;
 OPENDHT_C_PUBLIC dht_privatekey* dht_privatekey_generate(unsigned key_length_bits);
 OPENDHT_C_PUBLIC dht_privatekey* dht_privatekey_import(const uint8_t* dat, size_t dat_size, const char* password);
+OPENDHT_C_PUBLIC int dht_privatekey_export(const dht_privatekey*, char* out, size_t* out_size, const char* password);
 OPENDHT_C_PUBLIC dht_publickey* dht_privatekey_get_publickey(const dht_privatekey*);
 OPENDHT_C_PUBLIC void dht_privatekey_delete(dht_privatekey*);
 

configure.ac

@@ -35,10 +35,14 @@ AS_IF([test "x$enable_logs" != "xno"], [
 dnl Check for indexation
 AC_ARG_ENABLE([indexation], [AS_HELP_STRING([--disable-indexation], [Disable DHT indexation])])
 AM_CONDITIONAL(ENABLE_INDEXATION, test x$enable_indexation != "xno")
-AS_IF([test "x$enable_indexation" != "xno"], [
-	AC_DEFINE([OPENDHT_INDEXATION], [1], [Define if DHT indexation is enabled])
+AM_COND_IF(ENABLE_INDEXATION, [
+	AC_DEFINE([OPENDHT_INDEXATION], [], [Define if DHT indexation is enabled])
 ])
 
+AC_ARG_ENABLE([peer-discovery], [AS_HELP_STRING([--disable-peer-discovery], [Disable peer-discovery])])
+AM_CONDITIONAL(ENABLE_PEER_DISCOVERY, test x$enable_peer_discovery != "xno")
+AM_COND_IF(ENABLE_PEER_DISCOVERY, [AC_DEFINE([OPENDHT_PEER_DISCOVERY], [], [Define if peer discovery is enabled])])
+
 dnl Check for Doxygen
 AC_ARG_ENABLE([doc], AS_HELP_STRING([--enable-doc], [Enable documentation generation (doxygen)]))
 AS_IF([test "x$enable_doc" = "xyes"], [
@@ -135,7 +139,7 @@ AS_IF([test "x$with_jsoncpp" != "xno"],
       [have_jsoncpp=no])
 AS_IF([test "x$have_jsoncpp" = "xyes"], [
     AC_MSG_NOTICE([Using JsonCpp])
-    CPPFLAGS+=" -DOPENDHT_JSONCPP=1"
+    CPPFLAGS+=" -DOPENDHT_JSONCPP"
 ], [
     AC_MSG_NOTICE([Not using JsonCpp])
     AM_COND_IF(PROXY_CLIENT_OR_SERVER, AC_MSG_ERROR(["JsonCpp is required for proxy/push notification support"]))
@@ -154,7 +158,7 @@ AS_IF([test "x$have_openssl" = "xyes"], [
 AC_ARG_WITH([http_parser_fork], AS_HELP_STRING([--with-http-parser-fork], [Build with http_parser fork to support old API]))
 AS_IF([test "x$with_http_parser_fork" = "xyes"],[
     AC_MSG_NOTICE([Using http_parser fork])
-    CPPFLAGS+=" -DOPENDHT_PROXY_HTTP_PARSER_FORK"
+    AC_DEFINE([OPENDHT_PROXY_HTTP_PARSER_FORK], [], [Define if using http parser fork])
 ], [
     AC_MSG_NOTICE([Not using http_parser fork])
 ])
@@ -205,8 +209,8 @@ AM_COND_IF([ENABLE_TOOLS], [
   ])
 ])
 
-AM_COND_IF(ENABLE_PROXY_SERVER, [CPPFLAGS+=" -DOPENDHT_PROXY_SERVER"], [])
-AM_COND_IF(ENABLE_PROXY_CLIENT, [CPPFLAGS+=" -DOPENDHT_PROXY_CLIENT"], [])
+AM_COND_IF(ENABLE_PROXY_SERVER, AC_DEFINE([OPENDHT_PROXY_SERVER], [], [Building with proxy server]))
+AM_COND_IF(ENABLE_PROXY_CLIENT, AC_DEFINE([OPENDHT_PROXY_CLIENT], [], [Building with proxy client]))
 AM_COND_IF(ENABLE_PUSH_NOTIFICATIONS, [CPPFLAGS+=" -DOPENDHT_PUSH_NOTIFICATIONS"], [])
 AM_COND_IF(ENABLE_PROXY_SERVER_IDENTITY, [CPPFLAGS+=" -DOPENDHT_PROXY_SERVER_IDENTITY"], [])
 

docker/DockerfileDeps

@@ -3,13 +3,13 @@ MAINTAINER Adrien Béraud <adrien.beraud@savoirfairelinux.com>
 RUN apt-get update && apt-get install -y \
         build-essential cmake git wget libncurses5-dev libreadline-dev nettle-dev \
         libgnutls28-dev libuv1-dev cython3 python3-dev python3-setuptools libcppunit-dev libjsoncpp-dev \
-        autotools-dev autoconf libasio-dev \
+        autotools-dev autoconf libssl-dev \
     && apt-get clean
 
 RUN echo "*** Installing libfmt ***" \
     && wget https://github.com/fmtlib/fmt/archive/5.3.0.tar.gz \
     && tar -xzf 5.3.0.tar.gz && cd fmt-5.3.0/ \
-    && cmake -DCMAKE_INSTALL_PREFIX=/usr -DBUILD_SHARED_LIBS=On . \
+    && cmake -DCMAKE_INSTALL_PREFIX=/usr -DBUILD_SHARED_LIBS=On -DFMT_TEST=Off -DFMT_DOC=Off . \
     && make -j8 && make install \
     && cd ../ && rm -rf fmt*
 
@@ -17,22 +17,22 @@ RUN echo "*** Installing libfmt ***" \
 RUN echo "*** Building nodejs/http_parser dependency ***" \
     && wget https://github.com/nodejs/http-parser/archive/v2.9.2.tar.gz \
     && tar xvf v2.9.2.tar.gz && cd http-parser-2.9.2/ \
-    && PREFIX=/usr make -j8 && make install \
+    && PREFIX=/usr make -j8 install \
     && cd ../ && rm -rf http-parser*
 
 # libasio-dev (1.10.6-3) is too old
 RUN echo "** Building a recent version of asio ***" \
-    && wget https://github.com/chriskohlhoff/asio/archive/asio-1-12-2.tar.gz \
-    && tar -xvf asio-1-12-2.tar.gz && cd asio-asio-1-12-2/asio \
-    && ./autogen.sh && ./configure --prefix=/usr --without-boost \
+    && wget https://github.com/aberaud/asio/archive/b2b7a1c166390459e1c169c8ae9ef3234b361e3f.tar.gz \
+    && tar -xvf b2b7a1c166390459e1c169c8ae9ef3234b361e3f.tar.gz && cd asio-b2b7a1c166390459e1c169c8ae9ef3234b361e3f/asio \
+    && ./autogen.sh && ./configure --prefix=/usr --without-boost --disable-examples --disable-tests  \
     && make install \
     && cd ../../ && rm -rf asio*
 
 RUN echo "*** Downloading RESTinio ***" \
     && mkdir restinio && cd restinio \
-    && wget https://github.com/aberaud/restinio/archive/8d5d3e8237e0947adb9ba1ffc8281f4ad7cb2a59.tar.gz \
-    && ls -l && tar -xzf 8d5d3e8237e0947adb9ba1ffc8281f4ad7cb2a59.tar.gz \
-    && cd restinio-8d5d3e8237e0947adb9ba1ffc8281f4ad7cb2a59/dev \
+    && wget https://github.com/aberaud/restinio/archive/a7a10e419d9089c5b8ee63f5e3098c892f22fae4.tar.gz \
+    && ls -l && tar -xzf a7a10e419d9089c5b8ee63f5e3098c892f22fae4.tar.gz \
+    && cd restinio-a7a10e419d9089c5b8ee63f5e3098c892f22fae4/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 . \

docker/DockerfileDepsBionic

@@ -5,14 +5,14 @@ RUN echo "APT::Acquire::Retries \"3\";" > /etc/apt/apt.conf.d/80-retries
 RUN apt-get update && apt-get install -y \
         apt-transport-https build-essential cmake git wget libncurses5-dev libreadline-dev nettle-dev \
         libgnutls28-dev libuv1-dev cython3 python3-dev python3-setuptools libcppunit-dev libjsoncpp-dev \
-        autotools-dev autoconf libasio-dev libfmt-dev libhttp-parser-dev libmsgpack-dev \
+        autotools-dev autoconf libfmt-dev libhttp-parser-dev libmsgpack-dev \
     && apt-get clean
 
 # libasio-dev (1.10) is too old
 RUN echo "** Building a recent version of asio ***" \
-    && wget https://github.com/chriskohlhoff/asio/archive/asio-1-12-2.tar.gz \
-    && tar -xvf asio-1-12-2.tar.gz && cd asio-asio-1-12-2/asio \
-    && ./autogen.sh && ./configure --prefix=/usr --without-boost \
+    && wget https://github.com/aberaud/asio/archive/b2b7a1c166390459e1c169c8ae9ef3234b361e3f.tar.gz \
+    && tar -xvf b2b7a1c166390459e1c169c8ae9ef3234b361e3f.tar.gz && cd asio-b2b7a1c166390459e1c169c8ae9ef3234b361e3f/asio \
+    && ./autogen.sh && ./configure --prefix=/usr --without-boost --disable-examples --disable-tests  \
     && make install \
     && cd ../../ && rm -rf asio*
 

docker/DockerfileDepsLlvm

@@ -2,9 +2,9 @@ FROM ubuntu:16.04
 MAINTAINER Adrien Béraud <adrien.beraud@savoirfairelinux.com>
 RUN apt-get update \
     && apt-get install -y llvm llvm-dev clang make cmake git wget libncurses5-dev libreadline-dev \
-       nettle-dev libgnutls28-dev libuv1-dev libmsgpack-dev libjsoncpp-dev libasio-dev cython3 python3-dev \
+       nettle-dev libgnutls28-dev libuv1-dev libmsgpack-dev libjsoncpp-dev cython3 python3-dev \
        python3-setuptools libcppunit-dev python3-pip \
-       autotools-dev autoconf libasio-dev \
+       autotools-dev autoconf libssl-dev \
     && apt-get remove -y gcc g++ && apt-get autoremove -y && apt-get clean
 
 ENV CC cc
@@ -13,7 +13,7 @@ ENV CXX c++
 RUN echo "*** Installing libfmt ***" \
     && wget https://github.com/fmtlib/fmt/archive/5.3.0.tar.gz \
     && tar -xzf 5.3.0.tar.gz && cd fmt-5.3.0/ \
-    && cmake -DCMAKE_INSTALL_PREFIX=/usr -DBUILD_SHARED_LIBS=On . \
+    && cmake -DCMAKE_INSTALL_PREFIX=/usr -DBUILD_SHARED_LIBS=On -DFMT_TEST=Off -DFMT_DOC=Off . \
     && make -j8 && make install \
     && cd ../ && rm -rf fmt*
 
@@ -21,14 +21,14 @@ RUN echo "*** Installing libfmt ***" \
 RUN echo "*** Building nodejs/http_parser dependency ***" \
     && wget https://github.com/nodejs/http-parser/archive/v2.9.2.tar.gz \
     && tar xvf v2.9.2.tar.gz && cd http-parser-2.9.2/ \
-    && PREFIX=/usr make -j8 && make install \
+    && PREFIX=/usr make -j8 install \
     && cd ../ && rm -rf http-parser*
 
 # libasio-dev (1.10.6-3) is too old
 RUN echo "** Building a recent version of asio ***" \
-    && wget https://github.com/chriskohlhoff/asio/archive/asio-1-12-2.tar.gz \
-    && tar -xvf asio-1-12-2.tar.gz && cd asio-asio-1-12-2/asio \
-    && ./autogen.sh && ./configure --prefix=/usr --without-boost \
+    && wget https://github.com/aberaud/asio/archive/b2b7a1c166390459e1c169c8ae9ef3234b361e3f.tar.gz \
+    && tar -xvf b2b7a1c166390459e1c169c8ae9ef3234b361e3f.tar.gz && cd asio-b2b7a1c166390459e1c169c8ae9ef3234b361e3f/asio \
+    && ./autogen.sh && ./configure --prefix=/usr --without-boost --disable-examples --disable-tests  \
     && make install \
     && cd ../../ && rm -rf asio*
 
@@ -37,9 +37,9 @@ RUN apt-get update && apt-get install -y python3-pip && pip3 install --upgrade c
 
 RUN echo "*** Downloading RESTinio ***" \
     && mkdir restinio && cd restinio \
-    && wget https://github.com/Stiffstream/restinio/archive/v.0.5.1.2.tar.gz \
-    && ls -l && tar -xzf v.0.5.1.2.tar.gz \
-    && cd restinio-v.0.5.1.2/dev \
+    && wget https://github.com/aberaud/restinio/archive/a7a10e419d9089c5b8ee63f5e3098c892f22fae4.tar.gz \
+    && ls -l && tar -xzf a7a10e419d9089c5b8ee63f5e3098c892f22fae4.tar.gz \
+    && cd restinio-a7a10e419d9089c5b8ee63f5e3098c892f22fae4/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 . \

include/opendht.h

@@ -19,10 +19,4 @@
 #pragma once
 
 #include "opendht/dhtrunner.h"
-#ifdef OPENDHT_PROXY_SERVER
-#include "opendht/dht_proxy_server.h"
-#endif
 #include "opendht/default_types.h"
-#ifdef OPENDHT_INDEXATION
-#include "opendht/indexation/pht.h"
-#endif

include/opendht/callbacks.h

@@ -76,9 +76,9 @@ struct OPENDHT_PUBLIC NodeStats {
 struct OPENDHT_PUBLIC NodeInfo {
     InfoHash id;
     InfoHash node_id;
-    NodeStats ipv4;
-    NodeStats ipv6;
-    size_t ongoing_ops;
+    NodeStats ipv4 {};
+    NodeStats ipv6 {};
+    size_t ongoing_ops {0};
 
 #ifdef OPENDHT_JSONCPP
     /**
@@ -122,6 +122,7 @@ struct OPENDHT_PUBLIC Config {
     ssize_t max_peer_req_per_sec {0};
 };
 
+#ifndef OPENDHT_LIGHT
 /**
  * SecureDht configuration.
  */
@@ -130,6 +131,7 @@ struct OPENDHT_PUBLIC SecureDhtConfig
     Config node_config {};
     crypto::Identity id {};
 };
+#endif
 
 static constexpr size_t DEFAULT_STORAGE_LIMIT {1024 * 1024 * 64};
 

include/opendht/crypto.h

@@ -160,6 +160,7 @@ struct OPENDHT_PUBLIC PrivateKey
     explicit operator bool() const { return key; }
 
     PublicKey getPublicKey() const;
+    int serialize(uint8_t* out, size_t* out_len, const std::string& password = {}) const;
     Blob serialize(const std::string& password = {}) const;
 
     /**

include/opendht/default_types.h

@@ -58,6 +58,7 @@ public:
     MSGPACK_DEFINE(service, data)
 };
 
+#ifndef OPENDHT_LIGHT
 template <typename T>
 class OPENDHT_PUBLIC SignedValue : public Value::Serializable<T>
 {
@@ -102,7 +103,6 @@ public:
 
 
 
-
 class OPENDHT_PUBLIC ImMessage : public SignedValue<ImMessage>
 {
 private:
@@ -177,16 +177,8 @@ public:
     {
         pk.pack_array(2);
         pk.pack(id);
-#if 1
         pk.pack_bin(ice_data.size());
         pk.pack_bin_body((const char*)ice_data.data(), ice_data.size());
-#else
-        // hack for backward compatibility with old opendht compiled with msgpack 1.0
-        // remove when enough people have moved to new versions
-        pk.pack_array(ice_data.size());
-        for (uint8_t b : ice_data)
-            pk.pack(b);
-#endif
     }
 
     virtual void msgpack_unpack(msgpack::object o)
@@ -200,6 +192,7 @@ public:
     Value::Id id {0};
     Blob ice_data;
 };
+#endif
 
 /* "Peer" announcement
  */
@@ -261,8 +254,11 @@ private:
     SockAddr addr;
 };
 
-
+#ifndef OPENDHT_LIGHT
 OPENDHT_PUBLIC extern const std::array<std::reference_wrapper<const ValueType>, 5> DEFAULT_TYPES;
+#else
+OPENDHT_PUBLIC extern const std::array<std::reference_wrapper<const ValueType>, 1> DEFAULT_TYPES;
+#endif
 
 OPENDHT_PUBLIC extern const std::array<std::reference_wrapper<const ValueType>, 1> DEFAULT_INSECURE_TYPES;
 

include/opendht/dht_proxy_client.h

@@ -53,9 +53,9 @@ public:
     DhtProxyClient();
 
     explicit DhtProxyClient(
-        std::shared_ptr<dht::crypto::Certificate> serverCA, dht::crypto::Identity clientIdentity,
+        std::shared_ptr<crypto::Certificate> serverCA, crypto::Identity clientIdentity,
         std::function<void()> loopSignal, const std::string& serverHost,
-        const std::string& pushClientId = "", std::shared_ptr<dht::Logger> logger = {});
+        const std::string& pushClientId = "", std::shared_ptr<Logger> logger = {});
 
     void setHeaderFields(http::Request& request);
 
@@ -404,7 +404,7 @@ private:
 #endif // OPENDHT_PUSH_NOTIFICATIONS
 
     Json::StreamWriterBuilder jsonBuilder_;
-    Json::CharReaderBuilder jsonReaderBuilder_;
+    std::unique_ptr<Json::CharReader> jsonReader_;
     std::shared_ptr<dht::Logger> logger_;
 
     std::shared_ptr<http::Request> buildRequest(const std::string& target = {});

include/opendht/dht_proxy_server.h

@@ -31,10 +31,7 @@
 
 #include <restinio/all.hpp>
 #include <restinio/tls.hpp>
-
-#ifdef OPENDHT_JSONCPP
 #include <json/json.h>
-#endif
 
 #include <memory>
 #include <mutex>
@@ -54,8 +51,6 @@ class DhtRunner;
 
 using RestRouter = restinio::router::express_router_t<>;
 using RequestStatus = restinio::request_handling_status_t;
-using ResponseByParts = restinio::chunked_output_t;
-using ResponseByPartsBuilder = restinio::response_builder_t<ResponseByParts>;
 
 /**
  * Describes the REST API
@@ -87,15 +82,15 @@ public:
 
     struct ServerStats {
         /** Current number of listen operations */
-        size_t listenCount;
+        size_t listenCount {0};
         /** Current number of permanent put operations */
-        size_t putCount;
+        size_t putCount {0};
         /** Current number of push tokens with at least one listen operation */
-        size_t pushListenersCount;
+        size_t pushListenersCount {0};
         /** Average requests per second */
-        double requestRate;
+        double requestRate {0};
         /** Node Info **/
-        NodeInfo nodeInfo;
+        NodeInfo nodeInfo {};
 
         std::string toString() const {
             std::ostringstream ss;
@@ -113,7 +108,6 @@ public:
             return ss.str();
         }
 
-#ifdef OPENDHT_JSONCPP
         /**
          * Build a json object from a NodeStats
          */
@@ -126,7 +120,6 @@ public:
             result["nodeInfo"] = nodeInfo.toJson();
             return result;
         }
-#endif
     };
 
     ServerStats stats() const { return stats_; }

include/opendht/http.h

@@ -99,7 +99,7 @@ public:
     asio::streambuf& input();
     std::istream& data() { return istream_; }
 
-    std::string read_bytes(size_t bytes);
+    std::string read_bytes(size_t bytes = 0);
     std::string read_until(const char delim);
 
     void async_connect(std::vector<asio::ip::tcp::endpoint>&& endpoints, ConnectHandlerCb);
@@ -108,11 +108,20 @@ public:
     void async_read_until(const char* delim, BytesHandlerCb cb);
     void async_read_until(char delim, BytesHandlerCb cb);
     void async_read(size_t bytes, BytesHandlerCb cb);
+    void async_read_some(size_t bytes, BytesHandlerCb cb);
 
     void timeout(const std::chrono::seconds timeout, HandlerCb cb = {});
     void close();
 
 private:
+
+    template<typename T>
+    T wrapCallabck(T cb) const {
+        return [t=shared_from_this(),cb=std::move(cb)](auto ...params) {
+            return cb(params...);
+        };
+    }
+
     unsigned int id_;
     static std::atomic_uint ids_;
 
@@ -212,6 +221,7 @@ public:
     using OnDataCb = std::function<void(const char* at, size_t length)>;
     using OnStateChangeCb = std::function<void(State state, const Response& response)>;
     using OnJsonCb = std::function<void(Json::Value value, unsigned int status_code)>;
+    using OnDoneCb = std::function<void(const Response& response)>;
 
     // resolves implicitly
     Request(asio::io_context& ctx, const std::string& url, const Json::Value& json, OnJsonCb jsoncb,
@@ -219,6 +229,7 @@ public:
     Request(asio::io_context& ctx, const std::string& url, std::shared_ptr<dht::Logger> logger = {});
     Request(asio::io_context& ctx, const std::string& host, const std::string& service,
             const bool ssl = false, std::shared_ptr<dht::Logger> logger = {});
+    Request(asio::io_context& ctx, const std::string& url, OnDoneCb onDone, std::shared_ptr<dht::Logger> logger = {});
 
     // user defined resolver
     Request(asio::io_context& ctx, std::shared_ptr<Resolver> resolver, sa_family_t family = AF_UNSPEC);
@@ -230,7 +241,7 @@ public:
 
     ~Request();
 
-    unsigned int id() const;
+    inline unsigned int id() const { return  id_; };
     void set_connection(std::shared_ptr<Connection> connection);
     std::shared_ptr<Connection> get_connection() const;
     inline const Url& get_url() const {
@@ -258,6 +269,7 @@ public:
     void add_on_status_callback(OnStatusCb cb);
     void add_on_body_callback(OnDataCb cb);
     void add_on_state_change_callback(OnStateChangeCb cb);
+    void add_on_done_callback(OnDoneCb cb);
 
     void send();
 
@@ -270,19 +282,14 @@ private:
     using OnCompleteCb = std::function<void()>;
 
     struct Callbacks {
-        Callbacks(){}
-
         OnStatusCb on_status;
         OnDataCb on_header_field;
         OnDataCb on_header_value;
         OnDataCb on_body;
-        OnCompleteCb on_headers_complete;
-        OnCompleteCb on_message_complete;
-
         OnStateChangeCb on_state_change;
     };
 
-    void notify_state_change(const State state);
+    void notify_state_change(State state);
 
     void build();
 
@@ -299,17 +306,11 @@ private:
     void post();
 
     void handle_request(const asio::error_code& ec);
+    void handle_response(const asio::error_code& ec, size_t bytes);
 
-    void handle_response_header(const asio::error_code& ec);
-
-    void handle_response_body(const asio::error_code& ec, size_t bytes);
-
-    /**
-     * Parse the request with http_parser.
-     * Return how many bytes were parsed.
-     * Note: we pass requerst.size()==0 to signal that EOF has been received.
-     */
-    size_t parse_request(const std::string& request);
+    void onHeadersComplete();
+    void onBody(const char* at, size_t length);
+    void onComplete();
 
     std::shared_ptr<dht::Logger> logger_;
 
@@ -318,7 +319,6 @@ private:
     restinio::http_connection_header_t connection_type_ {restinio::http_connection_header_t::close};
     std::string body_;
 
-    std::mutex cbs_mutex_;
     Callbacks cbs_;
     State state_;
 
@@ -336,7 +336,6 @@ private:
 
     Response response_ {};
     std::string request_;
-    std::atomic<bool> message_complete_ {false};
     std::atomic<bool> finishing_ {false};
     std::unique_ptr<http_parser> parser_;
     std::unique_ptr<http_parser_settings> parser_s_;

include/opendht/infohash.h

@@ -50,7 +50,7 @@ namespace dht {
 using byte = uint8_t;
 
 namespace crypto {
-    OPENDHT_PUBLIC void hash(const uint8_t* data, size_t data_length, uint8_t* hash, size_t hash_length);
+OPENDHT_PUBLIC void hash(const uint8_t* data, size_t data_length, uint8_t* hash, size_t hash_length);
 }
 
 /**

include/opendht/log.h

@@ -71,13 +71,13 @@ OPENDHT_PUBLIC void
 printLog(std::ostream &s, char const *m, va_list args);
 
 OPENDHT_PUBLIC
-std::unique_ptr<Logger> getStdLogger();
+std::shared_ptr<Logger> getStdLogger();
 
 OPENDHT_PUBLIC
-std::unique_ptr<Logger> getFileLogger(const std::string &path);
+std::shared_ptr<Logger> getFileLogger(const std::string &path);
 
 OPENDHT_PUBLIC
-std::unique_ptr<Logger> getSyslogLogger(const char* name);
+std::shared_ptr<Logger> getSyslogLogger(const char* name);
 
 OPENDHT_PUBLIC void
 enableLogging(dht::DhtRunner &dht);

include/opendht/peer_discovery.h

@@ -22,10 +22,13 @@
 #include "def.h"
 #include "sockaddr.h"
 #include "infohash.h"
+#include "log_enable.h"
 
 #include <thread>
-#include <mutex>
-#include <condition_variable>
+
+namespace asio {
+class io_context;
+}
 
 namespace dht {
 
@@ -35,7 +38,7 @@ public:
     static constexpr in_port_t DEFAULT_PORT = 8888;
     using ServiceDiscoveredCallback = std::function<void(msgpack::object&&, SockAddr&&)>;
 
-    PeerDiscovery(in_port_t port = DEFAULT_PORT);
+    PeerDiscovery(in_port_t port = DEFAULT_PORT, std::shared_ptr<asio::io_context> ioContext = {}, std::shared_ptr<Logger> logger = {});
     ~PeerDiscovery();
 
     /**
@@ -51,7 +54,7 @@ public:
     }
 
     /**
-     * startPublish - Keeping sending data until node is joinned or stop is called - msgpack
+     * startPublish - Keeping sending data until node is joinned or stop is called
     */
     void startPublish(const std::string &type, const msgpack::sbuffer &pack_buf);
     void startPublish(sa_family_t domain, const std::string &type, const msgpack::sbuffer &pack_buf);
@@ -79,20 +82,12 @@ public:
     bool stopPublish(const std::string &type);
     bool stopPublish(sa_family_t domain, const std::string &type);
 
-    /**
-     * Configure the sockopt to be able to listen multicast group
-    */
-    static void joinMulticast(int sockfd, sa_family_t family);
-
-    /**
-     * Join the threads
-    */
-    void join();
-
 private:
     class DomainPeerDiscovery;
     std::unique_ptr<DomainPeerDiscovery> peerDiscovery4_;
     std::unique_ptr<DomainPeerDiscovery> peerDiscovery6_;
+    std::shared_ptr<asio::io_context> ioContext_;
+    std::thread ioRunnner_;
 };
 
 }

include/opendht/value.h

@@ -20,7 +20,7 @@
 #pragma once
 
 #include "infohash.h"
-#include "crypto.h"
+//#include "crypto.h"
 #include "utils.h"
 #include "sockaddr.h"
 
@@ -46,6 +46,12 @@ namespace dht {
 struct Value;
 struct Query;
 
+namespace crypto {
+class PublicKey;
+class PrivateKey;
+class Certificate;
+}
+
 /**
  * A storage policy is applied once to every incoming value storage requests.
  * If the policy returns false, the value is dropped.
@@ -241,15 +247,9 @@ struct OPENDHT_PUBLIC Value
         };
     }
 
-    static Filter OwnerFilter(const crypto::PublicKey& pk) {
-        return OwnerFilter(pk.getId());
-    }
+    static Filter OwnerFilter(const crypto::PublicKey& pk);
 
-    static Filter OwnerFilter(const InfoHash& pkh) {
-        return [pkh](const Value& v) {
-            return v.owner and v.owner->getId() == pkh;
-        };
-    }
+    static Filter OwnerFilter(const InfoHash& pkh);
 
     static Filter SeqNumFilter(uint16_t seq_no) {
         return [seq_no](const Value& v) {
@@ -341,20 +341,13 @@ struct OPENDHT_PUBLIC Value
      * Afterward, checkSignature() will return true and owner will
      * be set to the corresponding public key.
      */
-    void sign(const crypto::PrivateKey& key) {
-        if (isEncrypted())
-            throw DhtException("Can't sign encrypted data.");
-        owner = std::make_shared<const crypto::PublicKey>(key.getPublicKey());
-        signature = key.sign(getToSign());
-    }
+    void sign(const crypto::PrivateKey& key);
 
     /**
      * Check that the value is signed and that the signature matches.
      * If true, the owner field will contain the signer public key.
      */
-    bool checkSignature() const {
-        return isSigned() and owner->checkSignature(getToSign(), signature);
-    }
+    bool checkSignature() const;
 
     std::shared_ptr<const crypto::PublicKey> getOwner() const {
         return std::static_pointer_cast<const crypto::PublicKey>(owner);
@@ -363,15 +356,7 @@ struct OPENDHT_PUBLIC Value
     /**
      * Sign the value with from and returns the encrypted version for to.
      */
-    Value encrypt(const crypto::PrivateKey& from, const crypto::PublicKey& to) {
-        if (isEncrypted())
-            throw DhtException("Data is already encrypted.");
-        setRecipient(to.getId());
-        sign(from);
-        Value nv {id};
-        nv.setCypher(to.encrypt(getToEncrypt()));
-        return nv;
-    }
+    Value encrypt(const crypto::PrivateKey& from, const crypto::PublicKey& to);
 
     Value() {}
 
@@ -424,7 +409,11 @@ struct OPENDHT_PUBLIC Value
     inline bool operator== (const Value& o) {
         return id == o.id &&
         (isEncrypted() ? cypher == o.cypher :
-        ((owner == o.owner || *owner == *o.owner) && type == o.type && data == o.data && user_type == o.user_type && signature == o.signature));
+        (
+#ifndef OPENDHT_LIGHT
+            (owner == o.owner || *owner == *o.owner) &&
+#endif
+            type == o.type && data == o.data && user_type == o.user_type && signature == o.signature));
     }
 
     void setRecipient(const InfoHash& r) {
@@ -482,7 +471,8 @@ struct OPENDHT_PUBLIC Value
     template <typename Packer>
     void msgpack_pack_to_sign(Packer& pk) const
     {
-        bool has_owner = owner && *owner;
+#ifndef OPENDHT_LIGHT
+        bool has_owner = owner && *owner
         pk.pack_map((user_type.empty()?0:1) + (has_owner?(recipient ? 5 : 4):2));
         if (has_owner) { // isSigned
             pk.pack(std::string("seq"));   pk.pack(seq);
@@ -491,6 +481,9 @@ struct OPENDHT_PUBLIC Value
                 pk.pack(std::string("to")); pk.pack(recipient);
             }
         }
+#else
+        pk.pack_map((user_type.empty()?0:1) + 2);
+#endif
         pk.pack(std::string("type"));  pk.pack(type);
         pk.pack(std::string("data"));  pk.pack_bin(data.size());
                                        pk.pack_bin_body((const char*)data.data(), data.size());
@@ -535,9 +528,11 @@ struct OPENDHT_PUBLIC Value
                     pk.pack(static_cast<uint64_t>(type));
                     break;
                 case Value::Field::OwnerPk:
+#ifndef OPENDHT_LIGHT
                     if (owner)
                         owner->msgpack_pack(pk);
                     else
+#endif
                         InfoHash().msgpack_pack(pk);
                     break;
                 case Value::Field::SeqNum:

src/Makefile.am

@@ -27,7 +27,6 @@ libopendht_la_SOURCES  = \
         dhtrunner.cpp \
         default_types.cpp \
         log.cpp \
-        peer_discovery.cpp \
         network_utils.cpp \
         thread_pool.cpp
 
@@ -56,7 +55,6 @@ nobase_include_HEADERS = \
         ../include/opendht/default_types.h \
         ../include/opendht/log.h \
         ../include/opendht/log_enable.h \
-        ../include/opendht/peer_discovery.h \
         ../include/opendht/network_utils.h \
         ../include/opendht/rng.h \
         ../include/opendht/thread_pool.h
@@ -77,6 +75,11 @@ libopendht_la_SOURCES += http.cpp
 nobase_include_HEADERS += ../include/opendht/proxy.h ../include/opendht/http.h
 endif
 
+if ENABLE_PEER_DISCOVERY
+libopendht_la_SOURCES += peer_discovery.cpp
+nobase_include_HEADERS += ../include/opendht/peer_discovery.h
+endif
+
 if ENABLE_INDEXATION
 libopendht_la_SOURCES += indexation/pht.cpp
 nobase_include_HEADERS += ../include/opendht/indexation/pht.h

src/callbacks.cpp

@@ -1,3 +1,25 @@
+/*
+ *  Copyright (C) 2014-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/>.
+ */
+
+#ifdef HAVE_CONFIG_H
+#include "config.h"
+#endif
+
 #include "callbacks.h"
 
 namespace dht {

src/crypto.cpp

@@ -348,9 +348,7 @@ PrivateKey::serialize(const std::string& password) const
     size_t buf_sz = 8192;
     Blob buffer;
     buffer.resize(buf_sz);
-    int err = password.empty()
-        ? gnutls_x509_privkey_export_pkcs8(x509_key, GNUTLS_X509_FMT_PEM, nullptr,          GNUTLS_PKCS_PLAIN,         buffer.data(), &buf_sz)
-        : gnutls_x509_privkey_export_pkcs8(x509_key, GNUTLS_X509_FMT_PEM, password.c_str(), GNUTLS_PKCS_PBES2_AES_256, buffer.data(), &buf_sz);
+    auto err = serialize(buffer.data(), &buf_sz, password);
     if (err != GNUTLS_E_SUCCESS) {
         std::cerr << "Could not export private key - " << gnutls_strerror(err) << std::endl;
         return {};
@@ -359,6 +357,16 @@ PrivateKey::serialize(const std::string& password) const
     return buffer;
 }
 
+int
+PrivateKey::serialize(uint8_t* out, size_t* out_len, const std::string& password) const
+{
+    if (!x509_key)
+        return -1;
+    return password.empty()
+        ? gnutls_x509_privkey_export_pkcs8(x509_key, GNUTLS_X509_FMT_PEM, nullptr,          GNUTLS_PKCS_PLAIN,         out, out_len)
+        : gnutls_x509_privkey_export_pkcs8(x509_key, GNUTLS_X509_FMT_PEM, password.c_str(), GNUTLS_PKCS_PBES2_AES_256, out, out_len);
+}
+
 PublicKey
 PrivateKey::getPublicKey() const
 {

src/default_types.cpp

@@ -15,6 +15,9 @@
  *  You should have received a copy of the GNU General Public License
  *  along with this program. If not, see <https://www.gnu.org/licenses/>.
  */
+#ifdef HAVE_CONFIG_H
+#include "config.h"
+#endif
 
 #include "default_types.h"
 
@@ -83,6 +86,8 @@ IpServiceAnnouncement::storePolicy(InfoHash h, std::shared_ptr<Value>& v, const
 
 const ValueType DhtMessage::TYPE(1, "DHT message", std::chrono::minutes(5), DhtMessage::storePolicy);
 const ValueType IpServiceAnnouncement::TYPE(2, "Internet Service Announcement", std::chrono::minutes(15), IpServiceAnnouncement::storePolicy);
+
+#ifndef OPENDHT_LIGHT
 const ValueType ImMessage::TYPE = {3, "IM message", std::chrono::minutes(5)};
 const ValueType TrustRequest::TYPE = {4, "Certificate trust request", std::chrono::hours(24*7)};
 const ValueType IceCandidates::TYPE = {5, "ICE candidates", std::chrono::minutes(1)};
@@ -96,6 +101,13 @@ DEFAULT_TYPES
     IceCandidates::TYPE,
     TrustRequest::TYPE
 }};
+#else
+const std::array<std::reference_wrapper<const ValueType>, 1>
+DEFAULT_TYPES
+{{
+    ValueType::USER_DATA
+}};
+#endif
 
 const std::array<std::reference_wrapper<const ValueType>, 1>
 DEFAULT_INSECURE_TYPES

src/dht.cpp

@@ -76,8 +76,9 @@ Dht::shutdown(ShutdownCallback cb)
     for (auto& str : store)
         *remaining += maintainStorage(str, true, str_donecb);
 
+    DHT_LOG.w("shuting down node: after storage, %u ops", *remaining);
+
     if (!*remaining) {
-        DHT_LOG.w("shuting down node: %u ops remaining", *remaining);
         if (cb) cb();
     }
 }

src/dht_proxy_client.cpp

@@ -24,6 +24,7 @@
 #include "utils.h"
 
 #include <http_parser.h>
+#include <deque>
 
 namespace dht {
 
@@ -69,22 +70,44 @@ struct DhtProxyClient::ProxySearch {
     std::map<Value::Id, PermanentPut> puts {};
 };
 
+struct LineSplit {
+    void append(const char* d, size_t l) {
+        buf_.insert(buf_.end(), d, d+l);
+    }
+    bool getLine(char c) {
+        auto it = buf_.begin();
+        while (it != buf_.end()) {
+            if (*it == c) {
+                line_.clear();
+                line_.insert(line_.end(), buf_.begin(), ++it);
+                buf_.erase(buf_.begin(), it);
+                return true;
+            }
+            it++;
+        }
+        return false;
+    }
+    const std::string& line() const { return  line_; }
+private:
+    std::deque<char> buf_ {};
+    std::string line_ {};
+};
+
 DhtProxyClient::DhtProxyClient() {}
 
 DhtProxyClient::DhtProxyClient(
         std::shared_ptr<dht::crypto::Certificate> serverCA, dht::crypto::Identity clientIdentity,
         std::function<void()> signal, const std::string& serverHost,
         const std::string& pushClientId, std::shared_ptr<dht::Logger> logger)
-    :
-        clientIdentity_(clientIdentity), serverCertificate_(serverCA),
-        pushClientId_(pushClientId), loopSignal_(signal), logger_(logger)
+    : proxyUrl_(serverHost)
+    , clientIdentity_(clientIdentity), serverCertificate_(serverCA)
+    , pushClientId_(pushClientId), loopSignal_(signal)
+    , jsonReader_(Json::CharReaderBuilder{}.newCharReader())
+    , logger_(logger)
 {
-    // build http client
-    proxyUrl_ = serverHost;
-
     jsonBuilder_["commentStyle"] = "None";
     jsonBuilder_["indentation"] = "";
-    if (logger_){
+    if (logger_) {
         if (serverCertificate_)
             logger_->d("[proxy:client] using ca certificate for ssl:\n%s",
                        serverCertificate_->toString(false/*chain*/).c_str());
@@ -152,21 +175,22 @@ DhtProxyClient::~DhtProxyClient()
 void
 DhtProxyClient::stop()
 {
-    isDestroying_ = true;
-    resolver_.reset();
-    cancelAllListeners();
-    if (infoState_)
-        infoState_->cancel = true;
-    {
-        std::lock_guard<std::mutex> lock(requestLock_);
-        for (auto& request : requests_)
-            request.second->cancel();
+    if (not isDestroying_.exchange(true)) {
+        resolver_.reset();
+        cancelAllListeners();
+        if (infoState_)
+            infoState_->cancel = true;
+        {
+            std::lock_guard<std::mutex> lock(requestLock_);
+            for (auto& request : requests_)
+                request.second->cancel();
+        }
+        if (not httpContext_.stopped())
+            httpContext_.stop();
+        if (httpClientThread_.joinable())
+            httpClientThread_.join();
+        requests_.clear();
     }
-    if (not httpContext_.stopped())
-        httpContext_.stop();
-    if (httpClientThread_.joinable())
-        httpClientThread_.join();
-    requests_.clear();
 }
 
 std::vector<Sp<Value>>
@@ -271,28 +295,31 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Va
     if (logger_)
         logger_->d("[proxy:client] [get] [search %s]", key.to_c_str());
 
+    if (isDestroying_) {
+        if (donecb) donecb(false, {});
+        return;
+    }
     try {
         auto request = buildRequest("/" + key.toString());
         auto reqid = request->id();
-        request->set_connection_type(restinio::http_connection_header_t::keep_alive);
+        //request->set_connection_type(restinio::http_connection_header_t::keep_alive);
         request->set_method(restinio::http_method_get());
         setHeaderFields(*request);
 
         auto opstate = std::make_shared<OperationState>();
         Value::Filter filter = w.empty() ? f : f.chain(w.getFilter());
 
-        request->add_on_body_callback([this, key, opstate, filter, cb](const char* at, size_t length){
+        auto rxBuf = std::make_shared<LineSplit>();
+        request->add_on_body_callback([this, key, opstate, filter, rxBuf, cb](const char* at, size_t length){
             try {
-                auto body = std::string(at, length);
+                auto& b = *rxBuf;
+                b.append(at, length);
                 // one value per body line
-                std::string data_line;
-                std::stringstream body_stream(body);
-                while (std::getline(body_stream, data_line, '\n') and !(opstate->stop.load())){
+                while (b.getLine('\n') and !opstate->stop){
                     std::string err;
                     Json::Value json;
-                    auto* char_data = static_cast<const char*>(&data_line[0]);
-                    auto reader = std::unique_ptr<Json::CharReader>(jsonReaderBuilder_.newCharReader());
-                    if (!reader->parse(char_data, char_data + data_line.size(), &json, &err)){
+                    const auto& line = b.line();
+                    if (!jsonReader_->parse(line.data(), line.data() + line.size(), &json, &err)){
                         opstate->ok.store(false);
                         return;
                     }
@@ -315,30 +342,27 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Va
                 opstate->ok.store(false);
             }
         });
-        request->add_on_state_change_callback([this, reqid, opstate, donecb, key]
-                                              (http::Request::State state, const http::Response& response){
-            if (state == http::Request::State::DONE){
-                if (response.status_code != 200) {
-                    if (logger_)
-                        logger_->e("[proxy:client] [get %s] failed with code=%i", key.to_c_str(), response.status_code);
-                    opstate->ok.store(false);
-                    if (not response.aborted and response.status_code == 0)
-                        opFailed();
-                }
-                if (donecb) {
-                    {
-                        std::lock_guard<std::mutex> lock(lockCallbacks_);
-                        callbacks_.emplace_back([donecb, opstate](){
-                            donecb(opstate->ok, {});
-                            opstate->stop.store(true);
-                        });
-                    }
-                    loopSignal_();
-                }
-                if (not isDestroying_) {
-                    std::lock_guard<std::mutex> l(requestLock_);
-                    requests_.erase(reqid);
+        request->add_on_done_callback([this, reqid, opstate, donecb, key] (const http::Response& response){
+            if (response.status_code != 200) {
+                if (logger_)
+                    logger_->e("[proxy:client] [get %s] failed with code=%i", key.to_c_str(), response.status_code);
+                opstate->ok.store(false);
+                if (not response.aborted and response.status_code == 0)
+                    opFailed();
+            }
+            if (donecb) {
+                {
+                    std::lock_guard<std::mutex> lock(lockCallbacks_);
+                    callbacks_.emplace_back([donecb, opstate](){
+                        donecb(opstate->ok, {});
+                        opstate->stop.store(true);
+                    });
                 }
+                loopSignal_();
+            }
+            if (not isDestroying_) {
+                std::lock_guard<std::mutex> l(requestLock_);
+                requests_.erase(reqid);
             }
         });
         {
@@ -356,9 +380,8 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Va
 void
 DhtProxyClient::put(const InfoHash& key, Sp<Value> val, DoneCallback cb, time_point created, bool permanent)
 {
-    if (not val){
-        if (cb)
-            cb(false, {});
+    if (not val or isDestroying_) {
+        if (cb) cb(false, {});
         return;
     }
     if (logger_)
@@ -465,22 +488,19 @@ DhtProxyClient::doPut(const InfoHash& key, Sp<Value> val, DoneCallbackSimple cb,
             }
         }
         request->set_body(Json::writeString(jsonBuilder_, json));
-        request->add_on_state_change_callback([this, reqid, cb]
-                                              (http::Request::State state, const http::Response& response){
-            if (state == http::Request::State::DONE){
-                bool ok = response.status_code == 200;
-                if (not ok) {
-                    if (logger_)
-                        logger_->e("[proxy:client] [status] failed with code=%i", response.status_code);
-                    if (not response.aborted and response.status_code == 0)
-                        opFailed();
-                }
-                if (cb)
-                    cb(ok);
-                if (not isDestroying_) {
-                    std::lock_guard<std::mutex> l(requestLock_);
-                    requests_.erase(reqid);
-                }
+        request->add_on_done_callback([this, reqid, cb] (const http::Response& response){
+            bool ok = response.status_code == 200;
+            if (not ok) {
+                if (logger_)
+                    logger_->e("[proxy:client] [status] failed with code=%i", response.status_code);
+                if (not response.aborted and response.status_code == 0)
+                    opFailed();
+            }
+            if (cb)
+                cb(ok);
+            if (not isDestroying_) {
+                std::lock_guard<std::mutex> l(requestLock_);
+                requests_.erase(reqid);
             }
         });
         {
@@ -580,33 +600,29 @@ DhtProxyClient::queryProxyInfo(std::shared_ptr<InfoState> infoState, sa_family_t
         auto reqid = request->id();
         request->set_method(restinio::http_method_get()); 
         setHeaderFields(*request);
-        request->add_on_state_change_callback([this, reqid, family, infoState]
-                                              (http::Request::State state, const http::Response& response){
-            if (state == http::Request::State::DONE) {
-                if (infoState->cancel.load())
+        request->add_on_done_callback([this, reqid, family, infoState] (const http::Response& response){
+            if (infoState->cancel.load())
+                return;
+            if (response.status_code != 200) {
+                if (logger_)
+                    logger_->e("[proxy:client] [status] ipv%i failed with code=%i",
+                                family == AF_INET ? 4 : 6, response.status_code);
+                // pass along the failures
+                if ((family == AF_INET and infoState->ipv4 == 0) or (family == AF_INET6 and infoState->ipv6 == 0))
+                    onProxyInfos(Json::Value{}, family);
+            } else {
+                std::string err;
+                Json::Value proxyInfos;
+                if (!jsonReader_->parse(response.body.data(), response.body.data() + response.body.size(), &proxyInfos, &err)){
+                    onProxyInfos(Json::Value{}, family);
                     return;
-                if (response.status_code != 200) {
-                    if (logger_)
-                        logger_->e("[proxy:client] [status] ipv%i failed with code=%i",
-                                    family == AF_INET ? 4 : 6, response.status_code);
-                    // pass along the failures
-                    if ((family == AF_INET and infoState->ipv4 == 0) or (family == AF_INET6 and infoState->ipv6 == 0))
-                        onProxyInfos(Json::Value{}, family);
-                } else {
-                    std::string err;
-                    Json::Value proxyInfos;
-                    auto reader = std::unique_ptr<Json::CharReader>(jsonReaderBuilder_.newCharReader());
-                    if (!reader->parse(response.body.data(), response.body.data() + response.body.size(), &proxyInfos, &err)){
-                        onProxyInfos(Json::Value{}, family);
-                        return;
-                    }
-                    if (not infoState->cancel)
-                        onProxyInfos(proxyInfos, family);
-                }
-                if (not isDestroying_) {
-                    std::lock_guard<std::mutex> l(requestLock_);
-                    requests_.erase(reqid);
                 }
+                if (not infoState->cancel)
+                    onProxyInfos(proxyInfos, family);
+            }
+            if (not isDestroying_) {
+                std::lock_guard<std::mutex> l(requestLock_);
+                requests_.erase(reqid);
             }
         });
 
@@ -703,6 +719,8 @@ DhtProxyClient::listen(const InfoHash& key, ValueCallback cb, Value::Filter filt
 {
     if (logger_)
         logger_->d("[proxy:client] [listen] [search %s]", key.to_c_str());
+    if (isDestroying_)
+        return 0;
 
     std::lock_guard<std::mutex> lock(searchLock_);
     auto& search = searches_[key];
@@ -868,20 +886,17 @@ DhtProxyClient::handleExpireListener(const asio::error_code &ec, const InfoHash&
                 body["key"] = deviceKey_;
                 body["client_id"] = pushClientId_;
                 request->set_body(Json::writeString(jsonBuilder_, body));
-                request->add_on_state_change_callback([this, reqid, key]
-                                                      (http::Request::State state, const http::Response& response){
-                    if (state == http::Request::State::DONE){
-                        if (response.status_code != 200) {
-                            if (logger_)
-                                logger_->e("[proxy:client] [unsubscribe %s] failed with code=%i",
-                                           key.to_c_str(), response.status_code);
-                            if (not response.aborted and response.status_code == 0)
-                                opFailed();
-                        }
-                        if (not isDestroying_) {
-                            std::lock_guard<std::mutex> l(requestLock_);
-                            requests_.erase(reqid);
-                        }
+                request->add_on_done_callback([this, reqid, key] (const http::Response& response){
+                    if (response.status_code != 200) {
+                        if (logger_)
+                            logger_->e("[proxy:client] [unsubscribe %s] failed with code=%i",
+                                        key.to_c_str(), response.status_code);
+                        if (not response.aborted and response.status_code == 0)
+                            opFailed();
+                    }
+                    if (not isDestroying_) {
+                        std::lock_guard<std::mutex> l(requestLock_);
+                        requests_.erase(reqid);
                     }
                 });
                 {
@@ -935,19 +950,18 @@ DhtProxyClient::sendListen(const restinio::http_request_header_t header,
             body = fillBody(method == ListenMethod::RESUBSCRIBE);
         request->set_body(body);
 #endif
-
-        request->add_on_body_callback([this, reqid, opstate, cb](const char* at, size_t length){
+        auto rxBuf = std::make_shared<LineSplit>();
+        request->add_on_body_callback([this, reqid, opstate, rxBuf, cb](const char* at, size_t length){
             try {
-                auto body = std::string(at, length);
+                auto& b = *rxBuf;
+                b.append(at, length);
+
                 // one value per body line
-                std::string data_line;
-                std::stringstream body_stream(body);
-                while (std::getline(body_stream, data_line, '\n') and !opstate->stop){
+                while (b.getLine('\n') and !opstate->stop){
                     std::string err;
                     Json::Value json;
-                    auto* char_data = static_cast<const char*>(&data_line[0]);
-                    auto reader = std::unique_ptr<Json::CharReader>(jsonReaderBuilder_.newCharReader());
-                    if (!reader->parse(char_data, char_data + data_line.size(), &json, &err)){
+                    const auto& line = b.line();
+                    if (!jsonReader_->parse(line.data(), line.data() + line.size(), &json, &err)){
                         opstate->ok.store(false);
                         return;
                     }
@@ -974,21 +988,18 @@ DhtProxyClient::sendListen(const restinio::http_request_header_t header,
                 opstate->ok.store(false);
             }
         });
-        request->add_on_state_change_callback([this, opstate, reqid]
-                                              (http::Request::State state, const http::Response& response){
-            if (state == http::Request::State::DONE){
-                if (response.status_code != 200) {
-                    if (logger_)
-                        logger_->e("[proxy:client] [listen] send request #%i failed with code=%i",
-                                   reqid, response.status_code);
-                    opstate->ok.store(false);
-                    if (not response.aborted and response.status_code == 0)
-                        opFailed();
-                }
-                if (not isDestroying_) {
-                    std::lock_guard<std::mutex> l(requestLock_);
-                    requests_.erase(reqid);
-                }
+        request->add_on_done_callback([this, opstate, reqid] (const http::Response& response) {
+            if (response.status_code != 200) {
+                if (logger_)
+                    logger_->e("[proxy:client] [listen] send request #%i failed with code=%i",
+                                reqid, response.status_code);
+                opstate->ok.store(false);
+                if (not response.aborted and response.status_code == 0)
+                    opFailed();
+            }
+            if (not isDestroying_) {
+                std::lock_guard<std::mutex> l(requestLock_);
+                requests_.erase(reqid);
             }
         });
         {

src/dht_proxy_server.cpp

@@ -18,6 +18,10 @@
  *  along with this program. If not, see <https://www.gnu.org/licenses/>.
  */
 
+#ifdef HAVE_CONFIG_H
+#include "config.h"
+#endif
+
 #include "dht_proxy_server.h"
 
 #include "default_types.h"
@@ -68,6 +72,9 @@ constexpr char RESP_MSG_NO_TOKEN[] = "{\"err\":\"No token\"}";
 
 constexpr const std::chrono::minutes PRINT_STATS_PERIOD {2};
 
+using ResponseByParts = restinio::chunked_output_t;
+using ResponseByPartsBuilder = restinio::response_builder_t<ResponseByParts>;
+
 class opendht_logger_t
 {
 public:
@@ -386,7 +393,6 @@ HttpResponse DhtProxyServer::initHttpResponse(HttpResponse response) const
     response.append_header("Server", "RESTinio");
     response.append_header(restinio::http_field::content_type, "application/json");
     response.append_header(restinio::http_field::access_control_allow_origin, "*");
-    response.connection_keep_alive();
     return response;
 }
 
@@ -957,7 +963,7 @@ DhtProxyServer::put(restinio::request_handle_t request,
             if (logger_)
                 logger_->d("[proxy:server] [put %s] %s %s", infoHash.toString().c_str(),
                           value->toString().c_str(), (permanent ? "permanent" : ""));
-            if (permanent){
+            if (permanent) {
                 std::string pushToken, clientId, platform;
                 auto& pVal = root["permanent"];
                 if (pVal.isObject()){
@@ -1003,7 +1009,6 @@ DhtProxyServer::put(restinio::request_handle_t request,
                     if (pput.expireNotifyTimer)
                         pput.expireNotifyTimer->expires_at(timeout - proxy::OP_MARGIN);
                 }
-                lock.unlock();
             }
             dht_->put(infoHash, value, [this, request, value](bool ok){
                 if (ok){

src/dhtrunner.cpp

@@ -18,11 +18,16 @@
  *  along with this program. If not, see <https://www.gnu.org/licenses/>.
  */
 
+#ifdef HAVE_CONFIG_H
+#include "config.h"
+#endif
+
 #include "dhtrunner.h"
 #include "securedht.h"
-#include "peer_discovery.h"
 #include "network_utils.h"
-
+#ifdef OPENDHT_PEER_DISCOVERY
+#include "peer_discovery.h"
+#endif
 #ifdef OPENDHT_PROXY_CLIENT
 #include "dht_proxy_client.h"
 #endif
@@ -124,7 +129,8 @@ DhtRunner::run(const Config& config, Context&& context)
         {
             std::lock_guard<std::mutex> lck(sock_mtx);
             if (rcv.size() >= RX_QUEUE_MAX_SIZE) {
-                std::cerr << "Dropping packet: queue is full!" << std::endl;
+                if (logger_)
+                    logger_->e("Dropping packet: queue is full!");
                 rcv.pop();
             }
             rcv.emplace(std::move(pkt));
@@ -184,11 +190,16 @@ DhtRunner::run(const Config& config, Context&& context)
     });
 
     if (config.peer_discovery or config.peer_publish) {
+#ifdef OPENDHT_PEER_DISCOVERY
         peerDiscovery_ = context.peerDiscovery ?
             std::move(context.peerDiscovery) :
             std::make_shared<PeerDiscovery>();
+#else
+        std::cerr << "Peer discovery requested but OpenDHT built without peer discovery support." << std::endl;
+#endif
     }
 
+#ifdef OPENDHT_PEER_DISCOVERY
     auto netId = config.dht_config.node_config.network;
     if (config.peer_discovery) {
         peerDiscovery_->startDiscovery<NodeInsertionPack>(PEER_DISCOVERY_DHT_SERVICE, [this, netId](NodeInsertionPack&& v, SockAddr&& addr){
@@ -217,6 +228,7 @@ DhtRunner::run(const Config& config, Context&& context)
             peerDiscovery_->startPublish(AF_INET6, PEER_DISCOVERY_DHT_SERVICE, sbuf_node);
         }
     }
+#endif
 }
 
 void
@@ -292,8 +304,10 @@ DhtRunner::join()
             return;
         cv.notify_all();
         bootstrap_cv.notify_all();
+#ifdef OPENDHT_PEER_DISCOVERY
         if (peerDiscovery_)
             peerDiscovery_->stop();
+#endif
         if (dht_)
             if (auto sock = dht_->getSocket())
                 sock->stop();
@@ -307,9 +321,6 @@ DhtRunner::join()
     if (bootstrap_thread.joinable())
         bootstrap_thread.join();
 
-    if (peerDiscovery_)
-        peerDiscovery_->join();
-
     {
         std::lock_guard<std::mutex> lck(storage_mtx);
         pending_ops = decltype(pending_ops)();
@@ -465,12 +476,14 @@ DhtRunner::getNodesStats(sa_family_t af) const
 NodeInfo
 DhtRunner::getNodeInfo() const {
     std::lock_guard<std::mutex> lck(dht_mtx);
-    NodeInfo info;
-    info.id = getId();
-    info.node_id = getNodeId();
+    NodeInfo info {};
+    if (auto dht = activeDht()) {
+        info.id = dht->getId();
+        info.node_id = dht->getNodeId();
+        info.ipv4 = dht->getNodesStats(AF_INET);
+        info.ipv6 = dht->getNodesStats(AF_INET6);
+    }
     info.ongoing_ops = ongoing_ops;
-    info.ipv4 = dht_->getNodesStats(AF_INET);
-    info.ipv6 = dht_->getNodesStats(AF_INET6);
     return info;
 }
 

src/http.cpp

@@ -29,15 +29,8 @@
 namespace dht {
 namespace http {
 
-constexpr char HTTP_HEADER_CONNECTION[] = "Connection";
-constexpr char HTTP_HEADER_CONNECTION_KEEP_ALIVE[] = "keep-alive";
-constexpr char HTTP_HEADER_CONNECTION_CLOSE[] = "close";
-constexpr char HTTP_HEADER_CONTENT_LENGTH[] = "Content-Length";
 constexpr char HTTP_HEADER_CONTENT_TYPE_JSON[] = "application/json";
-constexpr char HTTP_HEADER_TRANSFER_ENCODING[] = "Transfer-Encoding";
-constexpr char HTTP_HEADER_TRANSFER_ENCODING_CHUNKED[] = "chunked";
 constexpr char HTTP_HEADER_DELIM[] = "\r\n\r\n";
-constexpr char BODY_VALUE_DELIM = '\n';
 
 Url::Url(const std::string& url): url(url)
 {
@@ -130,8 +123,7 @@ Connection::Connection(asio::io_context& ctx, std::shared_ptr<dht::crypto::Certi
     ssl_socket_ = std::make_unique<ssl_socket_t>(ctx_, ssl_ctx_);
 }
 
-Connection::~Connection()
-{
+Connection::~Connection() {
     close();
 }
 
@@ -200,6 +192,8 @@ Connection::input()
 std::string
 Connection::read_bytes(size_t bytes)
 {
+    if (bytes == 0)
+        bytes = read_buf_.in_avail();
     std::string content;
     content.resize(bytes);
     auto rb = istream_.readsome(&content[0], bytes);
@@ -219,9 +213,9 @@ void
 Connection::async_connect(std::vector<asio::ip::tcp::endpoint>&& endpoints, ConnectHandlerCb cb)
 {
     if (ssl_socket_)
-        asio::async_connect(ssl_socket_->lowest_layer(), std::move(endpoints), cb);
+        asio::async_connect(ssl_socket_->lowest_layer(), std::move(endpoints), wrapCallabck(std::move(cb)));
     else if (socket_)
-        asio::async_connect(*socket_, std::move(endpoints), cb);
+        asio::async_connect(*socket_, std::move(endpoints), wrapCallabck(std::move(cb)));
     else if (cb)
         cb(asio::error::operation_aborted, {});
 }
@@ -261,8 +255,8 @@ void
 Connection::async_write(BytesHandlerCb cb)
 {
     if (!is_open())   return;
-    if (ssl_socket_)  asio::async_write(*ssl_socket_, write_buf_, cb);
-    else if (socket_) asio::async_write(*socket_, write_buf_, cb);
+    if (ssl_socket_)  asio::async_write(*ssl_socket_, write_buf_, wrapCallabck(std::move(cb)));
+    else if (socket_) asio::async_write(*socket_, write_buf_, wrapCallabck(std::move(cb)));
     else if (cb)      cb(asio::error::operation_aborted, 0);
 }
 
@@ -270,8 +264,8 @@ void
 Connection::async_read_until(const char* delim, BytesHandlerCb cb)
 {
     if (!is_open())   return;
-    if (ssl_socket_)  asio::async_read_until(*ssl_socket_, read_buf_, delim, cb);
-    else if (socket_) asio::async_read_until(*socket_, read_buf_, delim, cb);
+    if (ssl_socket_)  asio::async_read_until(*ssl_socket_, read_buf_, delim, wrapCallabck(std::move(cb)));
+    else if (socket_) asio::async_read_until(*socket_, read_buf_, delim, wrapCallabck(std::move(cb)));
     else if (cb)      cb(asio::error::operation_aborted, 0);
 }
 
@@ -279,8 +273,8 @@ void
 Connection::async_read_until(char delim, BytesHandlerCb cb)
 {
     if (!is_open())   return;
-    if (ssl_socket_)  asio::async_read_until(*ssl_socket_, read_buf_, delim, cb);
-    else if (socket_) asio::async_read_until(*socket_, read_buf_, delim, cb);
+    if (ssl_socket_)  asio::async_read_until(*ssl_socket_, read_buf_, delim, wrapCallabck(std::move(cb)));
+    else if (socket_) asio::async_read_until(*socket_, read_buf_, delim, wrapCallabck(std::move(cb)));
     else if (cb)      cb(asio::error::operation_aborted, 0);
 }
 
@@ -288,11 +282,27 @@ void
 Connection::async_read(size_t bytes, BytesHandlerCb cb)
 {
     if (!is_open())   return;
-    if (ssl_socket_)  asio::async_read(*ssl_socket_, read_buf_, asio::transfer_exactly(bytes), cb);
-    else if (socket_) asio::async_read(*socket_, read_buf_, asio::transfer_exactly(bytes), cb);
+    if (ssl_socket_)  asio::async_read(*ssl_socket_, read_buf_, asio::transfer_exactly(bytes), wrapCallabck(std::move(cb)));
+    else if (socket_) asio::async_read(*socket_, read_buf_, asio::transfer_exactly(bytes), wrapCallabck(std::move(cb)));
     else if (cb)      cb(asio::error::operation_aborted, 0);
 }
 
+void
+Connection::async_read_some(size_t bytes, BytesHandlerCb cb)
+{
+    if (!is_open()) {
+        if (cb) cb(asio::error::operation_aborted, 0);
+        return;
+    }
+    auto buf = read_buf_.prepare(bytes);
+    auto onEnd = [this_=shared_from_this(), cb=std::move(cb)](const asio::error_code& ec, size_t t){
+        this_->read_buf_.commit(t);
+        cb(ec, t);
+    };
+    if (ssl_socket_)  ssl_socket_->async_read_some(buf, onEnd);
+    else              socket_->async_read_some(buf, onEnd);
+}
+
 void
 Connection::timeout(const std::chrono::seconds timeout, HandlerCb cb)
 {
@@ -438,18 +448,19 @@ Request::Request(asio::io_context& ctx, const std::string& url, const Json::Valu
     set_header_field(restinio::http_field_t::content_type, HTTP_HEADER_CONTENT_TYPE_JSON);
     set_header_field(restinio::http_field_t::accept, HTTP_HEADER_CONTENT_TYPE_JSON);
     Json::StreamWriterBuilder wbuilder;
+    set_method(restinio::http_method_post());
     set_body(Json::writeString(wbuilder, json));
-    add_on_state_change_callback([this, jsoncb](State state, const Response& response){
-        if (state != Request::State::DONE)
-            return;
+    add_on_done_callback([this, jsoncb](const Response& response){
         Json::Value json;
-        std::string err;
-        Json::CharReaderBuilder rbuilder;
-        auto reader = std::unique_ptr<Json::CharReader>(rbuilder.newCharReader());
-        if (!reader->parse(response.body.data(), response.body.data() + response.body.size(), &json, &err) and logger_)
-            logger_->e("[http:request:%i] can't parse response to json", id_, err.c_str());
+        if (response.status_code != 0) {
+            std::string err;
+            Json::CharReaderBuilder rbuilder;
+            auto reader = std::unique_ptr<Json::CharReader>(rbuilder.newCharReader());
+            if (!reader->parse(response.body.data(), response.body.data() + response.body.size(), &json, &err) and logger_)
+                logger_->e("[http:request:%i] can't parse response to json", id_, err.c_str());
+        }
         if (jsoncb)
-            jsoncb(json, response.status_code);
+            jsoncb(std::move(json), response.status_code);
     });
 }
 
@@ -460,6 +471,13 @@ Request::Request(asio::io_context& ctx, const std::string& url, std::shared_ptr<
     init_default_headers();
 }
 
+Request::Request(asio::io_context& ctx, const std::string& url, OnDoneCb onDone, std::shared_ptr<dht::Logger> logger)
+    : logger_(logger), id_(Request::ids_++), ctx_(ctx), resolver_(std::make_shared<Resolver>(ctx, url, logger))
+{
+    init_default_headers();
+    add_on_done_callback(std::move(onDone));
+}
+
 Request::Request(asio::io_context& ctx, const std::string& host, const std::string& service,
                  const bool ssl, std::shared_ptr<dht::Logger> logger)
     : logger_(logger), id_(Request::ids_++), ctx_(ctx),
@@ -499,7 +517,8 @@ void
 Request::init_default_headers()
 {
     const auto& url = resolver_->get_url();
-    set_header_field(restinio::http_field_t::host, url.host + ":" + url.service);
+    set_header_field(restinio::http_field_t::user_agent, "Mozilla/5.0");
+    set_header_field(restinio::http_field_t::accept, "text/html");
     set_target(url.target);
 }
 
@@ -510,39 +529,28 @@ Request::cancel()
         c->close();
 }
 
-unsigned int
-Request::id() const
-{
-    return id_;
-}
-
 void
-Request::set_connection(std::shared_ptr<Connection> connection)
-{
-    conn_ = connection;
+Request::set_connection(std::shared_ptr<Connection> connection) {
+    conn_ = std::move(connection);
 }
 
 std::shared_ptr<Connection>
-Request::get_connection() const
-{
+Request::get_connection() const {
     return conn_;
 }
 
 void
-Request::set_certificate_authority(std::shared_ptr<dht::crypto::Certificate> certificate)
-{
+Request::set_certificate_authority(std::shared_ptr<dht::crypto::Certificate> certificate) {
     server_ca_ = certificate;
 }
 
 void
-Request::set_identity(const dht::crypto::Identity& identity)
-{
+Request::set_identity(const dht::crypto::Identity& identity) {
     client_identity_ = identity;
 }
 
 void
-Request::set_logger(std::shared_ptr<dht::Logger> logger)
-{
+Request::set_logger(std::shared_ptr<dht::Logger> logger) {
     logger_ = logger;
 }
 
@@ -553,32 +561,27 @@ Request::set_header(restinio::http_request_header_t header)
 }
 
 void
-Request::set_method(restinio::http_method_id_t method)
-{
+Request::set_method(restinio::http_method_id_t method) {
     header_.method(method);
 }
 
 void
-Request::set_target(std::string target)
-{
+Request::set_target(std::string target) {
     header_.request_target(std::move(target));
 }
 
 void
-Request::set_header_field(restinio::http_field_t field, std::string value)
-{
+Request::set_header_field(restinio::http_field_t field, std::string value) {
     headers_[field] = std::move(value);
 }
 
 void
-Request::set_connection_type(restinio::http_connection_header_t connection)
-{
+Request::set_connection_type(restinio::http_connection_header_t connection) {
     connection_type_ = connection;
 }
 
 void
-Request::set_body(std::string body)
-{
+Request::set_body(std::string body) {
     body_ = std::move(body);
 }
 
@@ -597,7 +600,7 @@ void
 Request::build()
 {
     std::stringstream request;
-    bool append_body = true;
+    bool append_body = !body_.empty();
 
     // first header
     request << header_.method().c_str() << " " << header_.request_target() << " " <<
@@ -613,53 +616,54 @@ Request::build()
     // last connection header
     const char* conn_str = nullptr;
     switch (connection_type_){
+    case restinio::http_connection_header_t::keep_alive:
+        conn_str = "keep-alive";
+        break;
     case restinio::http_connection_header_t::upgrade:
         if (logger_)
             logger_->e("Unsupported connection type 'upgrade', fallback to 'close'");
     // fallthrough
     case restinio::http_connection_header_t::close:
-        conn_str = "close";
-        break;
-    case restinio::http_connection_header_t::keep_alive:
-        conn_str = "keep-alive";
+        conn_str = "close"; // default
         break;
     }
     if (conn_str)
         request << "Connection: " << conn_str << "\r\n";
 
     // body & content-length
-    if (!body_.empty())
-        request << "Content-Length: " << body_.size() << "\r\n\r\n";
-    // last delim
-    if (append_body)
-        request << body_ << "\r\n";
+    if (append_body) {
+        request << "Content-Length: " << body_.size() << "\r\n\r\n"
+                << body_;
+    } else
+        request << "\r\n";
     request_ = request.str();
 }
 
 void
-Request::add_on_status_callback(OnStatusCb cb)
-{
-    std::lock_guard<std::mutex> lock(cbs_mutex_);
+Request::add_on_status_callback(OnStatusCb cb) {
     cbs_.on_status = std::move(cb);
 }
 
 void
-Request::add_on_body_callback(OnDataCb cb)
-{
-    std::lock_guard<std::mutex> lock(cbs_mutex_);
+Request::add_on_body_callback(OnDataCb cb) {
     cbs_.on_body = std::move(cb);
 }
 
 void
-Request::add_on_state_change_callback(OnStateChangeCb cb)
-{
-    std::lock_guard<std::mutex> lock(cbs_mutex_);
+Request::add_on_state_change_callback(OnStateChangeCb cb) {
     cbs_.on_state_change = std::move(cb);
 }
 
 void
-Request::notify_state_change(const State state)
-{
+Request::add_on_done_callback(OnDoneCb cb) {
+    add_on_state_change_callback([onDone=std::move(cb)](State state, const Response& response){
+        if (state == Request::State::DONE)
+            onDone(response);
+    });
+}
+
+void
+Request::notify_state_change(State state) {
     state_ = state;
     if (cbs_.on_state_change)
         cbs_.on_state_change(state, response_);
@@ -676,38 +680,20 @@ Request::init_parser()
     if (!parser_s_)
         parser_s_ = std::make_unique<http_parser_settings>();
     http_parser_settings_init(parser_s_.get());
-    {
-        // user registered callbacks wrappers to store its data in the response
-        std::lock_guard<std::mutex> lock(cbs_mutex_);
-        cbs_.on_status = [this, statusCb = std::move(cbs_.on_status)](unsigned int status_code){
-            response_.status_code = status_code;
-            if (statusCb)
-                statusCb(status_code);
-        };
-        auto header_field = std::make_shared<std::string>();
-        cbs_.on_header_field = [header_field, headerFieldCb = std::move(cbs_.on_header_field)](const char* at, size_t length) {
-            *header_field = std::string(at, length);
-            if (headerFieldCb)
-                headerFieldCb(at, length);
-        };
-        cbs_.on_header_value = [this, header_field, headerValueCb = std::move(cbs_.on_header_value)](const char* at, size_t length) {
-            response_.headers[*header_field] = std::string(at, length);
-            if (headerValueCb)
-                headerValueCb(at, length);
-        };
-        cbs_.on_headers_complete = [this](){
-            notify_state_change(State::HEADER_RECEIVED);
-        };
-        cbs_.on_body = [bodyCb = std::move(cbs_.on_body)](const char* at, size_t length) {
-            if (bodyCb)
-                bodyCb(at, length);
-        };
-        cbs_.on_message_complete = [this](){
-            if (logger_)
-                logger_->d("[http:request:%i] response: message complete", id_);
-            message_complete_.store(true);
-        };
-    }
+
+    cbs_.on_status = [this, statusCb = std::move(cbs_.on_status)](unsigned int status_code){
+        response_.status_code = status_code;
+        if (statusCb)
+            statusCb(status_code);
+    };
+    auto header_field = std::make_shared<std::string>();
+    cbs_.on_header_field = [header_field](const char* at, size_t length) {
+        *header_field = std::string(at, length);
+    };
+    cbs_.on_header_value = [this, header_field](const char* at, size_t length) {
+        response_.headers[*header_field] = std::string(at, length);
+    };
+
     // http_parser raw c callback (note: no context can be passed into them)
     parser_s_->on_status = [](http_parser* parser, const char* /*at*/, size_t /*length*/) -> int {
         static_cast<Request*>(parser->data)->cbs_.on_status(parser->status_code);
@@ -722,15 +708,15 @@ Request::init_parser()
         return 0;
     };
     parser_s_->on_body = [](http_parser* parser, const char* at, size_t length) -> int {
-        static_cast<Request*>(parser->data)->cbs_.on_body(at, length);
+        static_cast<Request*>(parser->data)->onBody(at, length);
         return 0;
     };
     parser_s_->on_headers_complete = [](http_parser* parser) -> int {
-        static_cast<Request*>(parser->data)->cbs_.on_headers_complete();
+        static_cast<Request*>(parser->data)->onHeadersComplete();
         return 0;
     };
     parser_s_->on_message_complete = [](http_parser* parser) -> int {
-        static_cast<Request*>(parser->data)->cbs_.on_message_complete();
+        static_cast<Request*>(parser->data)->onComplete();
         return 0;
     };
 }
@@ -748,7 +734,7 @@ Request::connect(std::vector<asio::ip::tcp::endpoint>&& endpoints, HandlerCb cb)
     if (logger_){
         std::string eps = "";
         for (const auto& endpoint : endpoints)
-            eps.append(endpoint.address().to_string() + " ");
+            eps.append(endpoint.address().to_string() + ":" + std::to_string(endpoint.port()) + " ");
         logger_->d("[http:request:%i] connect begin: %s", id_, eps.c_str());
     }
     if (get_url().protocol == "https"){
@@ -772,13 +758,22 @@ Request::connect(std::vector<asio::ip::tcp::endpoint>&& endpoints, HandlerCb cb)
             this_.terminate(ec);
             return;
         }
-        else if (ec and this_.logger_)
-            this_.logger_->e("[http:request:%i] connect: failed with all endpoints", this_.id_);
-        else {
+        else if (ec) {
             if (this_.logger_)
-                this_.logger_->d("[http:request:%i] connect success", this_.id_);
+                this_.logger_->e("[http:request:%i] connect failed with all endpoints: %s", this_.id_, ec.message().c_str());
+        } else {
+            // if (this_.logger_)
+            //     this_.logger_->d("[http:request:%i] connect success", this_.id_);
 
-            if (this_.get_url().protocol == "https"){
+            const auto& url = this_.get_url();
+            auto port = endpoint.port();
+            if ((url.protocol == "http" && port == (in_port_t)80)
+             || (url.protocol == "https" && port == (in_port_t)443))
+                this_.set_header_field(restinio::http_field_t::host, url.host);
+            else
+                this_.set_header_field(restinio::http_field_t::host, url.host + ":" + std::to_string(port));
+
+            if (url.protocol == "https") {
                 if (this_.server_ca_)
                     this_.conn_->set_ssl_verification(endpoint, asio::ssl::verify_peer
                                                           | asio::ssl::verify_fail_if_no_peer_cert);
@@ -789,8 +784,8 @@ Request::connect(std::vector<asio::ip::tcp::endpoint>&& endpoints, HandlerCb cb)
                             return;
                         if (ec and logger)
                             logger->e("[http:request:%i] handshake error: %s", id, ec.message().c_str());
-                        else if (logger)
-                            logger->d("[http:request:%i] handshake success", id);
+                        //else if (logger)
+                        //    logger->d("[http:request:%i] handshake success", id);
                         if (cb)
                             cb(ec);
                     });
@@ -820,7 +815,7 @@ Request::send()
                     this_.logger_->e("[http:request:%i] resolve error: %s", this_.id_, ec.message().c_str());
                 this_.terminate(asio::error::connection_aborted);
             }
-            else if (!this_.conn_ or !this_.conn_->is_open()){
+            else if (!this_.conn_ or !this_.conn_->is_open()) {
                 this_.connect(std::move(endpoints), [wthis](const asio::error_code &ec) {
                     if (auto sthis = wthis.lock()) {
                         if (ec)
@@ -846,10 +841,9 @@ Request::post()
     build();
     init_parser();
 
-    if (logger_){
-        std::string header; std::getline(std::istringstream(request_), header);
-        logger_->d("[http:request:%i] send: %s", id_, header.c_str());
-    }
+    if (logger_)
+        logger_->d("[http:request:%i] sending %zu bytes", id_, request_.size());
+
     // write the request to buffer
     std::ostream request_stream(&conn_->input());
     request_stream << request_;
@@ -870,20 +864,16 @@ Request::terminate(const asio::error_code& ec)
     if (finishing_.exchange(true))
         return;
 
-    if (ec != asio::error::eof and ec != asio::error::operation_aborted and logger_)
-        logger_->e("[http:request:%i] end with error: %s", id_, ec.message().c_str());
-
-    // set response outcome, ignore end of file and abort
-    if ((!ec or ec == asio::error::eof) and parser_)
-        response_.status_code = parser_->status_code;
-    else
-        response_.status_code = 0;
-
     response_.aborted = ec == asio::error::operation_aborted;
 
-    if (logger_)
-        logger_->d("[http:request:%i] done with status code %u", id_, response_.status_code);
-    if (connection_type_ != restinio::http_connection_header_t::keep_alive)
+    if (logger_) {
+        if (ec and ec != asio::error::eof and ec != asio::error::operation_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);
+    }
+
+    if (!parser_ or !http_should_keep_alive(parser_.get()))
         if (auto c = conn_)
             c->close();
     notify_state_change(State::DONE);
@@ -900,41 +890,63 @@ Request::handle_request(const asio::error_code& ec)
         terminate(asio::error::not_connected);
         return;
     }
-    if (logger_)
-        logger_->d("[http:request:%i] send success", id_);
+    // if (logger_)
+    //    logger_->d("[http:request:%i] send success", id_);
     // read response
     notify_state_change(State::RECEIVING);
 
     std::weak_ptr<Request> wthis = shared_from_this();
-    conn_->async_read_until(HTTP_HEADER_DELIM, [wthis](const asio::error_code& ec, size_t){
+    conn_->async_read_until(HTTP_HEADER_DELIM, [wthis](const asio::error_code& ec, size_t n_bytes){
         if (auto sthis = wthis.lock())
-            sthis->handle_response_header(ec);
+            sthis->handle_response(ec, n_bytes);
     });
 }
 
 void
-Request::handle_response_header(const asio::error_code& ec)
+Request::handle_response(const asio::error_code& ec, size_t n_bytes)
 {
     if (ec && ec != asio::error::eof){
         terminate(ec);
         return;
     }
-    if (!conn_->is_open()){
-        terminate(asio::error::not_connected);
+    auto request = (ec == asio::error::eof) ? std::string{} : conn_->read_bytes();
+    size_t ret = http_parser_execute(parser_.get(), parser_s_.get(), request.c_str(), request.size());
+    if (ret != request.size()) {
+        if (logger_)
+            logger_->e("Error parsing HTTP: %zu %s %s", ret,
+                http_errno_name(HTTP_PARSER_ERRNO(parser_)),
+                http_errno_description(HTTP_PARSER_ERRNO(parser_)));
+        terminate(asio::error::basic_errors::broken_pipe);
         return;
     }
-    if (logger_)
-        logger_->d("[http:request:%i] response headers received", id_);
-    // read the header
-    std::string header;
-    std::string headers;
-    while (std::getline(conn_->data(), header) && header != "\r"){
-        headers.append(header + "\n");
-    }
-    headers.append("\n");
-    parse_request(headers);
 
-    std::weak_ptr<Request> wthis = shared_from_this();
+    if (state_ != State::DONE and parser_ and not http_body_is_final(parser_.get())) {
+        auto toRead = parser_->content_length ? std::min<uint64_t>(parser_->content_length, 64 * 1024) : 64 * 1024;
+        std::weak_ptr<Request> wthis = shared_from_this();
+        conn_->async_read_some(toRead, [wthis](const asio::error_code& ec, size_t bytes){
+            if (auto sthis = wthis.lock())
+                sthis->handle_response(ec, bytes);
+        });
+    }
+}
+
+void
+Request::onBody(const char* at, size_t length)
+{
+    if (cbs_.on_body)
+        cbs_.on_body(at, length);
+    else
+        response_.body.insert(response_.body.end(), at, at+length);
+}
+
+void
+Request::onComplete() {
+    terminate(asio::error::eof);
+}
+
+void
+Request::onHeadersComplete() {
+    notify_state_change(State::HEADER_RECEIVED);
 
     auto expect_it = headers_.find(restinio::http_field_t::expect);
     if (expect_it != headers_.end() and (expect_it->second == "100-continue") and response_.status_code != 200){
@@ -942,167 +954,13 @@ Request::handle_response_header(const asio::error_code& ec)
         request_.append(body_);
         std::ostream request_stream(&conn_->input());
         request_stream << body_ << "\r\n";
+        std::weak_ptr<Request> wthis = shared_from_this();
         conn_->async_write([wthis](const asio::error_code& ec, size_t) {
             if (auto sthis = wthis.lock())
                 sthis->handle_request(ec);
         });
         return;
     }
-
-    auto connection_it = response_.headers.find(HTTP_HEADER_CONNECTION);
-    auto content_length_it = response_.headers.find(HTTP_HEADER_CONTENT_LENGTH);
-    auto transfer_encoding_it = response_.headers.find(HTTP_HEADER_TRANSFER_ENCODING);
-
-    // has content-length
-    if (content_length_it != response_.headers.end())
-    {
-        unsigned int content_length = atoi(content_length_it->second.c_str());
-        response_.body.append(conn_->read_bytes(content_length));
-        // full body already in the header
-        if (response_.body.size() + 1 == content_length) {
-            response_.body.append("\n");
-            parse_request(response_.body);
-            if (message_complete_.load())
-                terminate(asio::error::eof);
-        }
-        else { // more chunks to come (don't add the missing \n from std::getline)
-            conn_->async_read(content_length - response_.body.size(), [wthis](const asio::error_code& ec, size_t bytes){
-                if (auto sthis = wthis.lock())
-                    sthis->handle_response_body(ec, bytes);
-            });
-        }
-    }
-    // server wants to keep sending or we have content-length defined
-    else if (connection_it != response_.headers.end() and connection_it->second == HTTP_HEADER_CONNECTION_KEEP_ALIVE)
-    {
-        conn_->async_read_until(BODY_VALUE_DELIM, [wthis](const asio::error_code& ec, size_t bytes){
-            if (auto sthis = wthis.lock())
-                sthis->handle_response_body(ec, bytes);
-        });
-    }
-    // server wants to close the connection
-    else if (connection_it != response_.headers.end() and connection_it->second == HTTP_HEADER_CONNECTION_CLOSE)
-    {
-        terminate(asio::error::eof);
-    }
-    // client wants to close the connection
-    else if (connection_type_ == restinio::http_connection_header_t::close)
-    {
-        terminate(asio::error::eof);
-    }
-    else if (transfer_encoding_it != response_.headers.end() and
-             transfer_encoding_it->second == HTTP_HEADER_TRANSFER_ENCODING_CHUNKED)
-    {
-        std::string chunk_size;
-        std::getline(conn_->data(), chunk_size);
-        unsigned int content_length = std::stoul(chunk_size, nullptr, 16);
-        conn_->async_read(content_length, [wthis](const asio::error_code& ec, size_t bytes){
-            if (auto sthis = wthis.lock())
-                sthis->handle_response_body(ec, bytes);
-        });
-    }
-}
-
-void
-Request::handle_response_body(const asio::error_code& ec, size_t bytes)
-{
-    if (ec && ec != asio::error::eof){
-        terminate(ec);
-        return;
-    }
-    if (!conn_->is_open()){
-        terminate(asio::error::not_connected);
-        return;
-    }
-    if (logger_)
-        logger_->d("[http:request:%i] response body: %i bytes received", id_, bytes);
-
-    if (bytes == 0){
-        terminate(asio::error::eof);
-        return;
-    }
-
-    // avoid creating non-existant headers by accessing the headers map without the presence of key
-    auto connection_it = response_.headers.find(HTTP_HEADER_CONNECTION);
-    auto content_length_it = response_.headers.find(HTTP_HEADER_CONTENT_LENGTH);
-    auto transfer_encoding_it = response_.headers.find(HTTP_HEADER_TRANSFER_ENCODING);
-
-    // read the content-length body
-    unsigned int content_length;
-    if (content_length_it != response_.headers.end() and !response_.body.empty()){
-        // extract the content-length
-        content_length = atoi(content_length_it->second.c_str());
-        response_.body.append(conn_->read_bytes(bytes));
-        // check if fully parsed
-        if (response_.body.size() == content_length)
-            parse_request(response_.body);
-    }
-    // read and parse the chunked encoding fragment
-    else {
-        auto body = conn_->read_until(BODY_VALUE_DELIM);
-        response_.body += body;
-        if (body == "0\r\n"){
-            parse_request(response_.body);
-            terminate(asio::error::eof);
-            return;
-        }
-        parse_request(body + '\n');
-    }
-
-    std::weak_ptr<Request> wthis = shared_from_this();
-
-    // should be executed after each parse_request who can trigger http_parser on_message_complete
-    if (message_complete_.load()){
-        terminate(asio::error::eof);
-    }
-    // has content-length
-    else if (content_length_it != response_.headers.end() and response_.body.size() != content_length)
-    {
-        conn_->async_read(content_length - response_.body.size(), [wthis](const asio::error_code& ec, size_t bytes){
-            if (auto sthis = wthis.lock())
-                sthis->handle_response_body(ec, bytes);
-        });
-    }
-    // server wants to keep sending
-    else if (connection_it != response_.headers.end() and connection_it->second == HTTP_HEADER_CONNECTION_KEEP_ALIVE)
-    {
-        conn_->async_read_until(BODY_VALUE_DELIM, [wthis](const asio::error_code& ec, size_t bytes){
-            if (auto sthis = wthis.lock())
-                sthis->handle_response_body(ec, bytes);
-        });
-    }
-    // server wants to close the connection
-    else if (connection_it != response_.headers.end() and connection_it->second == HTTP_HEADER_CONNECTION_CLOSE)
-    {
-        terminate(asio::error::eof);
-    }
-    // client wants to close the connection
-    else if (connection_type_ == restinio::http_connection_header_t::close)
-    {
-        terminate(asio::error::eof);
-    }
-    else if (transfer_encoding_it != response_.headers.end() and
-             transfer_encoding_it->second == HTTP_HEADER_TRANSFER_ENCODING_CHUNKED)
-    {
-        std::string chunk_size;
-        std::getline(conn_->data(), chunk_size);
-        if (chunk_size.size() == 0){
-            parse_request(response_.body);
-            terminate(asio::error::eof);
-        }
-        else
-            conn_->async_read_until(BODY_VALUE_DELIM, [wthis](const asio::error_code& ec, size_t bytes){
-            if (auto sthis = wthis.lock())
-                sthis->handle_response_body(ec, bytes);
-        });
-    }
-}
-
-size_t
-Request::parse_request(const std::string& request)
-{
-    std::lock_guard<std::mutex> lock(cbs_mutex_);
-    return http_parser_execute(parser_.get(), parser_s_.get(), request.c_str(), request.size());
 }
 
 } // namespace http

src/log.cpp

@@ -56,9 +56,9 @@ printLog(std::ostream& s, char const *m, va_list args) {
     s << std::endl;
 }
 
-std::unique_ptr<Logger>
+std::shared_ptr<Logger>
 getStdLogger() {
-    return std::unique_ptr<Logger>(new Logger(
+    return std::make_shared<Logger>(
         [](char const *m, va_list args) {
             std::cerr << red;
             printLog(std::cerr, m, args);
@@ -70,22 +70,22 @@ getStdLogger() {
             std::cout << def;
         },
         [](char const *m, va_list args) { printLog(std::cout, m, args); }
-    ));
+    );
 }
 
-std::unique_ptr<Logger>
+std::shared_ptr<Logger>
 getFileLogger(const std::string &path) {
     auto logfile = std::make_shared<std::ofstream>();
     logfile->open(path, std::ios::out);
 
-    return std::unique_ptr<Logger>(new Logger(
+    return std::make_shared<Logger>(
         [=](char const *m, va_list args) { printLog(*logfile, m, args); },
         [=](char const *m, va_list args) { printLog(*logfile, m, args); },
         [=](char const *m, va_list args) { printLog(*logfile, m, args); }
-    ));
+    );
 }
 
-std::unique_ptr<Logger>
+std::shared_ptr<Logger>
 getSyslogLogger(const char* name) {
 #ifndef _WIN32
     struct Syslog {
@@ -103,13 +103,13 @@ getSyslogLogger(const char* name) {
         logfile = std::make_shared<Syslog>(name);
         opened_logfile = logfile;
     }
-    return std::unique_ptr<Logger>(new Logger(
+    return std::make_shared<Logger>(
         [logfile](char const *m, va_list args) { vsyslog(LOG_ERR, m, args); },
         [logfile](char const *m, va_list args) { vsyslog(LOG_WARNING, m, args); },
         [logfile](char const *m, va_list args) { vsyslog(LOG_INFO, m, args); }
-    ));
+    );
 #else
-    return std::unique_ptr<Logger>(new Logger());
+    return std::make_shared<Logger>();
 #endif
 }
 

src/peer_discovery.cpp

@@ -2,6 +2,7 @@
  *  Copyright (C) 2014-2019 Savoir-faire Linux Inc.
  *  Author(s) : Mingrui Zhang <mingrui.zhang@savoirfairelinux.com>
  *              Vsevolod Ivanov <vsevolod.ivanov@savoirfairelinux.com>
+ *              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
@@ -21,402 +22,166 @@
 #include "network_utils.h"
 #include "utils.h"
 
-#ifdef _WIN32
-#include <Ws2tcpip.h> // needed for ip_mreq definition for multicast
-#include <Windows.h>
-#include <cstring>
-#if defined(_MSC_VER)
-#include <BaseTsd.h>
-typedef SSIZE_T ssize_t;
-#endif
-#define close(x) closesocket(x)
-#else
-#include <sys/types.h>
-#include <unistd.h>
-#endif
-#include <fcntl.h>
+#include <asio.hpp>
 
-#ifndef IPV6_JOIN_GROUP
-#define IPV6_JOIN_GROUP IPV6_ADD_MEMBERSHIP
-#endif
+using namespace std::chrono_literals;
 
 namespace dht {
 
 // Organization-local Scope multicast
-constexpr char MULTICAST_ADDRESS_IPV4[12] = "239.192.0.1";
-constexpr char MULTICAST_ADDRESS_IPV6[10] = "ff08::101";
+constexpr char MULTICAST_ADDRESS_IPV4[] = "239.192.0.1";
+constexpr char MULTICAST_ADDRESS_IPV6[] = "ff08::101";
 
 class PeerDiscovery::DomainPeerDiscovery
 {
 public:
-    DomainPeerDiscovery(sa_family_t domain, in_port_t port);
+    DomainPeerDiscovery(asio::ip::udp domain, in_port_t port, Sp<asio::io_context> ioContext = {}, Sp<Logger> logger = {});
     ~DomainPeerDiscovery();
 
-    /**
-     * startDiscovery - Keep Listening data from the sender until node is joinned or stop is called
-    */
     void startDiscovery(const std::string &type, ServiceDiscoveredCallback callback);
-
-    /**
-     * startPublish - Keeping sending data until node is joinned or stop is called - msgpack
-    */
     void startPublish(const std::string &type, const msgpack::sbuffer &pack_buf);
 
-    /**
-     * Thread Stopper
-    */
     void stop();
 
-    /**
-     * Remove possible callBack to discovery
-    */
     bool stopDiscovery(const std::string &type);
-
-    /**
-     * Remove different serivce message to send
-    */
     bool stopPublish(const std::string &type);
 
-    /**
-     * Configure the sockopt to be able to listen multicast group
-    */
-    static void joinMulticast(int sockfd, sa_family_t family);
-
-    /**
-     * Join the threads
-    */
-    void join() {
-        if(listening_.joinable()) listening_.join();
-        if(sending_.joinable()) sending_.join();
-    }
 private:
+    Sp<Logger> logger_;
     //dmtx_ for callbackmap_ and drunning_ (write)
     std::mutex dmtx_;
     //mtx_ for messages_ and lrunning (listen)
     std::mutex mtx_;
-    std::condition_variable cv_;
-    bool lrunning_ {false};
-    bool drunning_ {false};
-    sa_family_t domain_ {AF_UNSPEC};
-    int port_;
-    int sockFd_ {-1};
-    int stopWriteFd_ {-1};
+    std::shared_ptr<asio::io_context> ioContext_;
+    asio::ip::udp::socket sockFd_;
+    asio::ip::udp::endpoint sockAddrSend_;
+    asio::steady_timer publishTimer_;
 
-    SockAddr sockAddrSend_;
-
-    //Thread export to be joined
-    std::thread listening_;
-    std::thread sending_;
+    std::array<char, 64 * 1024> receiveBuf_;
+    asio::ip::udp::endpoint receiveFrom_;
 
     msgpack::sbuffer sbuf_;
     std::map<std::string, msgpack::sbuffer> messages_;
     std::map<std::string, ServiceDiscoveredCallback> callbackmap_;
+    bool lrunning_ {false};
+    bool drunning_ {false};
 
-    /**
-     * Multicast Socket Initialization, accept IPV4, IPV6
-    */
-    static int init(sa_family_t domain);
-
-    /**
-     * Receive messages
-    */
-    std::pair<SockAddr, msgpack::object_handle> recvFrom();
-
-    /**
-     * Listener pack thread loop
-    */
     void loopListener();
-
-    /**
-     * Listener Parameters Setup
-    */
-    void setupListener();
-
-    /**
-     * Sender Parameters Setup
-    */
-    void setupSender();
-    /**
-     * Sender Parameters Setup
-    */
     void reloadMessages();
 
     void stopDiscovery();
     void stopPublish();
+    void publish();
 };
 
-PeerDiscovery::DomainPeerDiscovery::DomainPeerDiscovery(sa_family_t domain, in_port_t port)
-    : domain_(domain), port_(port), sockFd_(init(domain))
+PeerDiscovery::DomainPeerDiscovery::DomainPeerDiscovery(asio::ip::udp domain, in_port_t port, Sp<asio::io_context> ioContext, Sp<Logger> logger)
+    : logger_(logger)
+    , ioContext_(ioContext)
+    , sockFd_(*ioContext_, domain)
+    , sockAddrSend_(asio::ip::address::from_string(domain.family() == AF_INET ? MULTICAST_ADDRESS_IPV4
+                                                                              : MULTICAST_ADDRESS_IPV6), port)
+    , publishTimer_(*ioContext)
 {
-    joinMulticast(sockFd_, domain);
-    setupListener();
-    setupSender();
+    sockFd_.set_option(asio::ip::multicast::join_group(sockAddrSend_.address()));
+    sockFd_.set_option(asio::ip::udp::socket::reuse_address(true));
+    sockFd_.bind({domain, port});
 }
 
 PeerDiscovery::DomainPeerDiscovery::~DomainPeerDiscovery()
 {
     stop();
-    join();
-    if (sockFd_ != -1)
-        close(sockFd_);
-
-#ifdef _WIN32
-    WSACleanup();
-#endif
-}
-
-int
-PeerDiscovery::DomainPeerDiscovery::init(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::setNonblocking(sockfd);
-    return sockfd;
-}
-
-void
-PeerDiscovery::DomainPeerDiscovery::setupListener()
-{
-    SockAddr sockAddrListen_;
-    sockAddrListen_.setFamily(domain_);
-    sockAddrListen_.setPort(port_);
-    sockAddrListen_.setAny();
-
-    unsigned int opt = 1;
-    if (setsockopt(sockFd_, SOL_SOCKET, SO_REUSEADDR, (char*)&opt, sizeof(opt)) < 0) {
-       std::cerr << "setsockopt SO_REUSEADDR failed: " << strerror(errno) << std::endl;
-    }
-#ifdef SO_REUSEPORT
-    if (setsockopt(sockFd_, SOL_SOCKET, SO_REUSEPORT, (char*)&opt, sizeof(opt)) < 0) {
-       std::cerr << "setsockopt SO_REUSEPORT failed: " << strerror(errno) << std::endl;
-    }
-#endif
-
-    // 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::DomainPeerDiscovery::joinMulticast(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, (char*)&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, (char*)&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, (char*)&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, (char*)&config_ipv6, sizeof(config_ipv6)) < 0){
-            throw std::runtime_error(std::string("Member Addition IPv6 Error: ") + strerror(errno));
-        }
-        break;
-    }
-    }
+    sockFd_.close();
 }
 
 void
 PeerDiscovery::DomainPeerDiscovery::startDiscovery(const std::string &type, ServiceDiscoveredCallback callback)
 {
-    std::unique_lock<std::mutex> lck(dmtx_);
+    std::lock_guard<std::mutex> lck(dmtx_);
     callbackmap_[type] = callback;
     if (not drunning_) {
-        if (listening_.joinable())
-            listening_.join();
         drunning_ = true;
-        listening_ = std::thread(&DomainPeerDiscovery::loopListener, this);
+        ioContext_->post(std::bind(&PeerDiscovery::DomainPeerDiscovery::loopListener, this));
     }
 }
 
-std::pair<SockAddr, msgpack::object_handle>
-PeerDiscovery::DomainPeerDiscovery::recvFrom()
-{
-    sockaddr_storage sa;
-    socklen_t sa_len = sizeof(sa);
-
-    std::array<uint8_t, 64 * 1024> buf;
-    ssize_t nbytes = recvfrom(
-        sockFd_,
-        (char*)buf.data(), buf.size(),
-        0,
-        (sockaddr*)&sa, &sa_len
-    );
-    if (nbytes < 0)
-        throw std::runtime_error(std::string("Error receiving packet: ") + strerror(errno));
-
-    return { SockAddr { sa, sa_len }, msgpack::unpack(reinterpret_cast<char*>(buf.data()), nbytes) };
-}
-
 void
 PeerDiscovery::DomainPeerDiscovery::loopListener()
 {
-    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];
-    stopWriteFd_ = stopfds_pipe[1];
-
-    while (true) {
-        {
-            std::lock_guard<std::mutex> lck(dmtx_);
-            if (not drunning_)
-                break;
+    std::lock_guard<std::mutex> lck(dmtx_);
+    if (not drunning_)
+        return;
+    sockFd_.async_receive_from(asio::buffer(receiveBuf_), receiveFrom_, [this](const asio::error_code& error, size_t bytes) {
+        if (error == asio::error::operation_aborted)
+            return;
+        if (error) {
+            if (logger_)
+                logger_->e("Error receiving message: %s", error.message().c_str());
         }
+        try {
+            auto rcv = msgpack::unpack(receiveBuf_.data(), bytes);
+            msgpack::object obj = rcv.get();
 
-        fd_set readfds;
-
-        FD_ZERO(&readfds);
-        FD_SET(stop_readfd, &readfds);
-        FD_SET(sockFd_, &readfds);
-
-        int data_coming = select(std::max(sockFd_, stop_readfd) + 1, &readfds, nullptr, nullptr, nullptr);
-
-        {
-            std::lock_guard<std::mutex> lck(dmtx_);
-            if (not drunning_)
-                break;
-        }
-
-        if (data_coming < 0) {
-            if(errno != EINTR) {
-                perror("Select Error");
-                std::this_thread::sleep_for( std::chrono::seconds(1) );
-            }
-        } else if (data_coming > 0) {
-            if (FD_ISSET(stop_readfd, &readfds)) {
-                std::array<uint8_t, 64 * 1024> buf;
-                recv(stop_readfd, (char*)buf.data(), buf.size(), 0);
-            }
-
-            try {
-                auto rcv = recvFrom();
-                msgpack::object obj = rcv.second.get();
-
-                if (obj.type != msgpack::type::MAP)
+            if (obj.type != msgpack::type::MAP)
+                throw msgpack::type_error{};
+            for (unsigned i = 0; i < obj.via.map.size; i++) {
+                auto& o = obj.via.map.ptr[i];
+                if (o.key.type != msgpack::type::STR)
                     continue;
-                for (unsigned i = 0; i < obj.via.map.size; i++) {
-                    auto& o = obj.via.map.ptr[i];
-                    if (o.key.type != msgpack::type::STR)
-                        continue;
-                    auto key = o.key.as<std::string>();
-                    ServiceDiscoveredCallback cb;
-                    {
-                        std::lock_guard<std::mutex> lck(dmtx_);
+                auto key = o.key.as<std::string>();
+                ServiceDiscoveredCallback cb;
+                {
+                    std::lock_guard<std::mutex> lck(dmtx_);
+                    if (drunning_) {
                         auto callback = callbackmap_.find(key);
                         if (callback != callbackmap_.end())
                             cb = callback->second;
-                    }
-                    if (cb)
-                        cb(std::move(o.val), std::move(rcv.first));
+                    } else
+                        return;
                 }
-            } catch (const std::exception& e) {
-                std::cerr << "Error receiving packet: " << e.what() << std::endl;
+                if (cb)
+                    cb(std::move(o.val), SockAddr{ receiveFrom_.data(), (socklen_t)receiveFrom_.size() });
             }
+        } catch (const std::exception& e) {
+            if (logger_)
+                logger_->e("Error receiving packet: %s", e.what());
         }
-    }
-    if (stop_readfd != -1)
-        close(stop_readfd);
-    if (stopWriteFd_ != -1) {
-        close(stopWriteFd_);
-        stopWriteFd_ = -1;
-    }
+        loopListener();
+    });
 }
 
 void
-PeerDiscovery::DomainPeerDiscovery::setupSender()
+PeerDiscovery::DomainPeerDiscovery::publish()
 {
-    // Setup sender address
-    sockAddrSend_.setFamily(domain_);
-    sockAddrSend_.setAddress(domain_ == AF_INET ? MULTICAST_ADDRESS_IPV4 : MULTICAST_ADDRESS_IPV6);
-    sockAddrSend_.setPort(port_);
+    std::lock_guard<std::mutex> lck(mtx_);
+    if (lrunning_) {
+        sockFd_.async_send_to(asio::buffer(sbuf_.data(), sbuf_.size()), sockAddrSend_, [logger=logger_](const asio::error_code& ec, size_t){
+            if (ec and ec != asio::error::operation_aborted) {
+                if (logger)
+                    logger->w("Error sending packet: %s", ec.message().c_str());
+            }
+        });
+        publishTimer_.expires_after(3s);
+        publishTimer_.async_wait(std::bind(&PeerDiscovery::DomainPeerDiscovery::publish, this));
+    }
 }
 
 void
 PeerDiscovery::DomainPeerDiscovery::startPublish(const std::string &type, const msgpack::sbuffer &pack_buf)
 {
-    //Set up New Sending pack
     msgpack::sbuffer pack_buf_c;
     pack_buf_c.write(pack_buf.data(),pack_buf.size());
 
-    std::unique_lock<std::mutex> lck(mtx_);
+    std::lock_guard<std::mutex> lck(mtx_);
     messages_[type] = std::move(pack_buf_c);
     reloadMessages();
     if (not lrunning_) {
-        if (sending_.joinable())
-            sending_.join();
         lrunning_ = true;
-        sending_ = std::thread([this](){
-            std::unique_lock<std::mutex> lck(mtx_);
-            while (lrunning_) {
-                ssize_t nbytes = sendto(
-                    sockFd_,
-                    sbuf_.data(),
-                    sbuf_.size(),
-                    0,
-                    sockAddrSend_.get(),
-                    sockAddrSend_.getLength()
-                );
-                if (nbytes < 0) {
-                    std::cerr << "Error sending packet: " << strerror(errno) << std::endl;
-                }
-                if (cv_.wait_for(lck,std::chrono::seconds(3),[&]{ return !lrunning_; }))
-                    break;
-            }
-        });
+        ioContext_->post(std::bind(&PeerDiscovery::DomainPeerDiscovery::publish, this));
     }
 }
 
 bool
-PeerDiscovery::DomainPeerDiscovery::stopDiscovery(const std::string &type)
+PeerDiscovery::DomainPeerDiscovery::stopDiscovery(const std::string& type)
 {
     std::lock_guard<std::mutex> lck(dmtx_);
     if (callbackmap_.erase(type) > 0) {
@@ -428,7 +193,7 @@ PeerDiscovery::DomainPeerDiscovery::stopDiscovery(const std::string &type)
 }
 
 bool
-PeerDiscovery::DomainPeerDiscovery::stopPublish(const std::string &type)
+PeerDiscovery::DomainPeerDiscovery::stopPublish(const std::string& type)
 {
     std::lock_guard<std::mutex> lck(mtx_);
     if (messages_.erase(type) > 0) {
@@ -444,20 +209,8 @@ PeerDiscovery::DomainPeerDiscovery::stopPublish(const std::string &type)
 void
 PeerDiscovery::DomainPeerDiscovery::stopDiscovery()
 {
-#ifdef _WIN32
-#define write(s, b, f) send((s), (b), (f), 0)
-#endif
-    if (drunning_) {
+    if (drunning_)
         drunning_ = false;
-        if (stopWriteFd_ != -1) {
-            if (write(stopWriteFd_, "\0", 1) == -1) {
-                std::cerr << "Can't send stop message: " << strerror(errno) << std::endl;
-            }
-        }
-    }
-#ifdef _WIN32
-#undef write
-#endif
 }
 
 void
@@ -465,7 +218,7 @@ PeerDiscovery::DomainPeerDiscovery::stopPublish()
 {
     if (lrunning_) {
         lrunning_ = false;
-        cv_.notify_all();
+        publishTimer_.cancel();
     }
 }
 
@@ -473,11 +226,11 @@ void
 PeerDiscovery::DomainPeerDiscovery::stop()
 {
     {
-        std::unique_lock<std::mutex> lck(dmtx_);
+        std::lock_guard<std::mutex> lck(dmtx_);
         stopDiscovery();
     }
     {
-        std::unique_lock<std::mutex> lck(mtx_);
+        std::lock_guard<std::mutex> lck(mtx_);
         stopPublish();
     }
 }
@@ -494,38 +247,65 @@ PeerDiscovery::DomainPeerDiscovery::reloadMessages()
     }
 }
 
-PeerDiscovery::PeerDiscovery(in_port_t port)
+PeerDiscovery::PeerDiscovery(in_port_t port, Sp<asio::io_context> ioContext, Sp<Logger> logger)
 {
+    if (not ioContext) {
+        ioContext = std::make_shared<asio::io_context>();
+        ioContext_ = ioContext;
+        ioRunnner_ = std::thread([logger, ioContext] {
+            try {
+                if (logger)
+                    logger->d("[peerdiscovery] starting io_context");
+                auto work = asio::make_work_guard(*ioContext);
+                ioContext->run();
+                if (logger)
+                    logger->d("[peerdiscovery] io_context stopped");
+            }
+            catch (const std::exception& ex){
+                if (logger)
+                    logger->e("[peerdiscovery] run error: %s", ex.what());
+            }
+        });
+    }
+
     try {
-        peerDiscovery4_.reset(new DomainPeerDiscovery(AF_INET, port));
+        peerDiscovery4_.reset(new DomainPeerDiscovery(asio::ip::udp::v4(), port, ioContext, logger));
     } catch(const std::exception& e){
-        std::cerr << "Can't start peer discovery (IPv4): " << e.what() << std::endl;
+        if (logger)
+            logger->e("[peerdiscovery] can't start IPv4: %s", e.what());
     }
     try {
-        peerDiscovery6_.reset(new DomainPeerDiscovery(AF_INET6, port));
+        peerDiscovery6_.reset(new DomainPeerDiscovery(asio::ip::udp::v6(), port, ioContext, logger));
     } catch(const std::exception& e) {
-        std::cerr << "Can't start peer discovery (IPv6): " << e.what() << std::endl;
+        if (logger)
+            logger->e("[peerdiscovery] can't start IPv6: %s", e.what());
     }
 }
 
-PeerDiscovery::~PeerDiscovery(){}
-
-void
-PeerDiscovery::startDiscovery(const std::string &type, ServiceDiscoveredCallback callback)
-{
-    if(peerDiscovery4_) peerDiscovery4_->startDiscovery(type, callback);
-    if(peerDiscovery6_) peerDiscovery6_->startDiscovery(type, callback);
+PeerDiscovery::~PeerDiscovery() {
+    stop();
+    if (ioContext_)
+        ioContext_->stop();
+    if (ioRunnner_.joinable())
+        ioRunnner_.join();
 }
 
 void
-PeerDiscovery::startPublish(const std::string &type, const msgpack::sbuffer &pack_buf)
+PeerDiscovery::startDiscovery(const std::string& type, ServiceDiscoveredCallback callback)
 {
-    if(peerDiscovery4_) peerDiscovery4_->startPublish(type, pack_buf);
-    if(peerDiscovery6_) peerDiscovery6_->startPublish(type, pack_buf);
+    if (peerDiscovery4_) peerDiscovery4_->startDiscovery(type, callback);
+    if (peerDiscovery6_) peerDiscovery6_->startDiscovery(type, callback);
 }
 
 void
-PeerDiscovery::startPublish(sa_family_t domain, const std::string &type, const msgpack::sbuffer &pack_buf)
+PeerDiscovery::startPublish(const std::string& type, const msgpack::sbuffer& pack_buf)
+{
+    if (peerDiscovery4_) peerDiscovery4_->startPublish(type, pack_buf);
+    if (peerDiscovery6_) peerDiscovery6_->startPublish(type, pack_buf);
+}
+
+void
+PeerDiscovery::startPublish(sa_family_t domain, const std::string& type, const msgpack::sbuffer& pack_buf)
 {
     if (domain == AF_INET) {
         if (peerDiscovery4_) peerDiscovery4_->startPublish(type, pack_buf);
@@ -537,8 +317,8 @@ PeerDiscovery::startPublish(sa_family_t domain, const std::string &type, const m
 void
 PeerDiscovery::stop()
 {
-    if(peerDiscovery4_) peerDiscovery4_->stop();
-    if(peerDiscovery6_) peerDiscovery6_->stop();
+    if (peerDiscovery4_) peerDiscovery4_->stop();
+    if (peerDiscovery6_) peerDiscovery6_->stop();
 }
 
 bool
@@ -558,7 +338,7 @@ PeerDiscovery::stopPublish(const std::string &type)
 }
 
 bool
-PeerDiscovery::stopPublish(sa_family_t domain, const std::string &type)
+PeerDiscovery::stopPublish(sa_family_t domain, const std::string& type)
 {
     if (domain == AF_INET) {
         return peerDiscovery4_ and peerDiscovery4_->stopPublish(type);
@@ -568,10 +348,4 @@ PeerDiscovery::stopPublish(sa_family_t domain, const std::string &type)
     return false;
 }
 
-void
-PeerDiscovery::join() {
-    if(peerDiscovery4_) peerDiscovery4_->join();
-    if(peerDiscovery6_) peerDiscovery6_->join();
-}
-
 }

src/securedht.cpp

@@ -17,6 +17,9 @@
  *  You should have received a copy of the GNU General Public License
  *  along with this program. If not, see <https://www.gnu.org/licenses/>.
  */
+#ifdef HAVE_CONFIG_H
+#include "config.h"
+#endif
 
 #include "securedht.h"
 #include "rng.h"

src/value.cpp

@@ -17,10 +17,17 @@
  *  along with this program. If not, see <https://www.gnu.org/licenses/>.
  */
 
-#include "value.h"
+#ifdef HAVE_CONFIG_H
+#include "config.h"
+#endif
 
+#include "value.h"
 #include "default_types.h"
+#include "callbacks.h"
+
+#ifndef OPENDHT_LIGHT
 #include "securedht.h" // print certificate ID
+#endif
 
 #ifdef OPENDHT_JSONCPP
 #include "base64.h"
@@ -38,6 +45,46 @@ Value::Filter bindFilterRaw(FilterRaw raw_filter, void* user_data) {
     };
 }
 
+#ifndef OPENDHT_LIGHT
+Value::Filter
+Value::OwnerFilter(const crypto::PublicKey& pk) {
+    return OwnerFilter(pk.getId());
+}
+
+
+Value::Filter
+Value::OwnerFilter(const InfoHash& pkh) {
+    return [pkh](const Value& v) {
+        return v.owner and v.owner->getId() == pkh;
+    };
+}
+
+void
+Value::sign(const crypto::PrivateKey& key) {
+    if (isEncrypted())
+        throw DhtException("Can't sign encrypted data.");
+    owner = std::make_shared<const crypto::PublicKey>(key.getPublicKey());
+    signature = key.sign(getToSign());
+}
+
+bool
+Value::checkSignature() const {
+    return isSigned() and owner->checkSignature(getToSign(), signature);
+}
+
+Value
+Value::encrypt(const crypto::PrivateKey& from, const crypto::PublicKey& to) {
+    if (isEncrypted())
+        throw DhtException("Data is already encrypted.");
+    setRecipient(to.getId());
+    sign(from);
+    Value nv {id};
+    nv.setCypher(to.encrypt(getToEncrypt()));
+    return nv;
+}
+
+#endif
+
 std::ostream& operator<< (std::ostream& s, const Value& v)
 {
     auto flags(s.flags());
@@ -52,7 +99,9 @@ std::ostream& operator<< (std::ostream& s, const Value& v)
     if (not v.isEncrypted()) {
         if (v.type == IpServiceAnnouncement::TYPE.id) {
             s << IpServiceAnnouncement(v.data);
-        } else if (v.type == CERTIFICATE_TYPE.id) {
+        }
+#ifndef OPENDHT_LIGHT
+        else if (v.type == CERTIFICATE_TYPE.id) {
             s << "Certificate";
 #ifdef OPENDHT_LOG_CRT_ID
             try {
@@ -62,7 +111,9 @@ std::ostream& operator<< (std::ostream& s, const Value& v)
                 s << " (invalid)";
             }
 #endif
-        } else {
+        }
+#endif
+        else {
             s << "Data (type: " << v.type << " ): ";
             s << std::hex;
             for (auto i : v.data)
@@ -159,9 +210,12 @@ Value::msgpack_unpack_body(const msgpack::object& o)
                 seq = rseq->as<decltype(seq)>();
             else
                 throw msgpack::type_error();
+
+#ifndef OPENDHT_LIGHT
             crypto::PublicKey new_owner;
             new_owner.msgpack_unpack(*rowner);
             owner = std::make_shared<const crypto::PublicKey>(std::move(new_owner));
+#endif
             if (auto rrecipient = findMapValue(*rbody, "to")) {
                 recipient = rrecipient->as<InfoHash>();
             }
@@ -190,11 +244,13 @@ Value::Value(Json::Value& json)
     }
     if (json.isMember("seq"))
         seq = json["seq"].asInt();
+#ifndef OPENDHT_LIGHT
     if (json.isMember("owner")) {
         auto ownerStr = json["owner"].asString();
         auto ownerBlob = std::vector<unsigned char>(ownerStr.begin(), ownerStr.end());
         owner = std::make_shared<const crypto::PublicKey>(ownerBlob);
     }
+#endif
     if (json.isMember("to")) {
         auto toStr = json["to"].asString();
         recipient = InfoHash(toStr);
@@ -314,9 +370,11 @@ FieldValueIndex::FieldValueIndex(const Value& v, const Select& s)
             case Value::Field::ValueType:
                 index[f] = {f, v.type};
                 break;
+#ifndef OPENDHT_LIGHT
             case Value::Field::OwnerPk:
                 index[f] = {f, v.owner ? v.owner->getId() : InfoHash() };
                 break;
+#endif
             case Value::Field::SeqNum:
                 index[f] = {f, v.seq};
                 break;

tests/compile_http.sh

@@ -1 +0,0 @@
-g++ -Wall -Wextra -Wfatal-errors -pedantic -g ../src/http.cpp tests_runner.cpp httptester.cpp -o run.out -lopendht -lpthread -lcppunit -ljsoncpp -DOPENDHT_JSONCPP -I../include/opendht -DASIO_STANDALONE

tests/httptester.cpp

@@ -19,41 +19,35 @@
 
 #include "httptester.h"
 
-// std
+#include <opendht/value.h>
+#include <opendht/dhtrunner.h>
+
 #include <iostream>
 #include <string>
-
 #include <chrono>
 #include <condition_variable>
 
-
 namespace test {
 CPPUNIT_TEST_SUITE_REGISTRATION(HttpTester);
 
 void
 HttpTester::setUp() {
-    logger = dht::log::getStdLogger();
+    nodePeer = std::make_shared<dht::DhtRunner>();
+    nodePeer->run(0);
 
-    nodePeer.run(0, /*identity*/{}, /*threaded*/true);
-
-    nodeProxy = std::make_shared<dht::DhtRunner>();
+    auto nodeProxy = std::make_shared<dht::DhtRunner>();
     nodeProxy->run(0, /*identity*/{}, /*threaded*/true);
-    nodeProxy->bootstrap(nodePeer.getBound());
+    nodeProxy->bootstrap(nodePeer->getBound());
 
     serverProxy = std::unique_ptr<dht::DhtProxyServer>(
         new dht::DhtProxyServer(
-            /*http*/dht::crypto::Identity{}, nodeProxy, 8080, /*pushServer*/"127.0.0.1:8090", logger));
-
+            /*http*/dht::crypto::Identity{}, nodeProxy, 8080, /*pushServer*/"127.0.0.1:8090"));
 }
 
 void
 HttpTester::tearDown() {
-    logger->d("[tester:http] stopping peer node");
-    nodePeer.join();
-    logger->d("[tester:http] stopping proxy server");
-    serverProxy.reset(nullptr);
-    logger->d("[tester:http] stopping proxy node");
-    nodeProxy->join();
+    serverProxy.reset();
+    nodePeer->join();
 }
 
 void
@@ -234,31 +228,72 @@ HttpTester::test_send_json() {
     std::mutex cv_m;
     std::unique_lock<std::mutex> lk(cv_m);
     bool done = false;
-
-    dht::Value val {"hey"};
-    auto json = val.toJson();
-    json["permanent"] = false;
-    std::cout << "[test_send_json] sending:\n" << json << std::endl;
-    Json::Value resp_val;
     unsigned int status = 0;
-    std::string url = "http://127.0.0.1:8080/key";
+
+    auto json = dht::Value("hey").toJson();
+    Json::Value resp_val;
+
     // Act
-    auto request = std::make_shared<dht::http::Request>(serverProxy->io_context(), url, json,
-                   [this, &cv, &done, &status, &resp_val](Json::Value value, unsigned int status_code){
-        if (status_code != 200 and logger)
-            logger->e("[tester] [status] failed with code=%i", status_code);
-        std::cout << "[tester] got response:\n" << value << std::endl;
-        resp_val = value;
-        status = status_code;
-        done = true;
-        cv.notify_all();
-    }, logger);
-    request->set_method(restinio::http_method_post());
+    auto request = std::make_shared<dht::http::Request>(serverProxy->io_context(), 
+        "http://127.0.0.1:8080/key",
+        json,
+        [&](Json::Value value, unsigned int status_code) {
+            std::lock_guard<std::mutex> lk(cv_m);
+            resp_val = std::move(value);
+            status = status_code;
+            done = true;
+            cv.notify_all();
+        });
     request->send();
+
     // Assert
     CPPUNIT_ASSERT(cv.wait_for(lk, std::chrono::seconds(10), [&]{ return done; }));
-    CPPUNIT_ASSERT(status == 200);
-    CPPUNIT_ASSERT(resp_val["data"] == val.toJson()["data"]);
+    CPPUNIT_ASSERT_EQUAL(200u, status);
+    CPPUNIT_ASSERT_EQUAL(json["data"].asString(), resp_val["data"].asString());
+    done = false;
+
+#if 0
+    request = std::make_shared<dht::http::Request>(serverProxy->io_context(),
+        "http://google.ca",
+        [&](const dht::http::Response& response) {
+            std::lock_guard<std::mutex> lk(cv_m);
+            status = response.status_code;
+            done = true;
+            cv.notify_all();
+        });
+    request->send();
+
+    CPPUNIT_ASSERT(cv.wait_for(lk, std::chrono::seconds(10), [&]{ return done; }));
+    CPPUNIT_ASSERT(status != 0);
+    done = false;
+
+    request = std::make_shared<dht::http::Request>(serverProxy->io_context(),
+        "https://google.ca",
+        [&](const dht::http::Response& response) {
+            std::lock_guard<std::mutex> lk(cv_m);
+            status = response.status_code;
+            done = true;
+            cv.notify_all();
+        });
+    request->send();
+
+    CPPUNIT_ASSERT(cv.wait_for(lk, std::chrono::seconds(10), [&]{ return done; }));
+    CPPUNIT_ASSERT(status != 0);
+    done = false;
+
+    request = std::make_shared<dht::http::Request>(serverProxy->io_context(),
+        "https://google.ca/sdbjklwGBIP",
+        [&](const dht::http::Response& response) {
+            std::lock_guard<std::mutex> lk(cv_m);
+            status = response.status_code;
+            done = true;
+            cv.notify_all();
+        });
+    request->send();
+
+    CPPUNIT_ASSERT(cv.wait_for(lk, std::chrono::seconds(10), [&]{ return done; }));
+    CPPUNIT_ASSERT_EQUAL(404u, status);
+#endif
 }
 
 }  // namespace test

tests/httptester.h

@@ -23,12 +23,12 @@
 #include <cppunit/TestFixture.h>
 #include <cppunit/extensions/HelperMacros.h>
 
-#include <opendht/value.h>
-#include <opendht/log.h>
 #include <opendht/http.h>
-
-#include <opendht/dhtrunner.h>
 #include <opendht/dht_proxy_server.h>
+#include <asio.hpp>
+
+#include <thread>
+#include <memory>
 
 namespace test {
 
@@ -88,10 +88,7 @@ class HttpTester : public CppUnit::TestFixture {
    void test_send_json();
 
  private:
-    std::shared_ptr<dht::Logger> logger {};
-
-    dht::DhtRunner nodePeer;
-    std::shared_ptr<dht::DhtRunner> nodeProxy;
+    std::shared_ptr<dht::DhtRunner> nodePeer;
     std::unique_ptr<dht::DhtProxyServer> serverProxy;
 };
 

tests/peerdiscoverytester.cpp

@@ -20,9 +20,12 @@
 #include "peerdiscoverytester.h"
 #include "opendht/value.h"
 
+#include <mutex>
+#include <condition_variable>
+
 namespace test {
 
-constexpr int MULTICAST_PORT = 2222;
+constexpr unsigned MULTICAST_PORT = 2222;
 const std::string DHT_NODE_NAME {"dht"};
 const std::string JAMI_NODE_NAME {"jami"};
 

tools/dhtchat.cpp

@@ -16,6 +16,9 @@
  *  You should have received a copy of the GNU General Public License
  *  along with this program. If not, see <https://www.gnu.org/licenses/>.
  */
+#ifdef HAVE_CONFIG_H
+#include "config.h"
+#endif
 
 #include "tools_common.h"
 #include <opendht/rng.h>

tools/dhtnode.cpp

@@ -18,6 +18,9 @@
  *  You should have received a copy of the GNU General Public License
  *  along with this program. If not, see <https://www.gnu.org/licenses/>.
  */
+#ifdef HAVE_CONFIG_H
+#include "config.h"
+#endif
 
 #include "tools_common.h"
 extern "C" {

tools/dhtscanner.cpp

@@ -16,6 +16,9 @@
  *  You should have received a copy of the GNU General Public License
  *  along with this program. If not, see <https://www.gnu.org/licenses/>.
  */
+#ifdef HAVE_CONFIG_H
+#include "config.h"
+#endif
 
 #include "tools_common.h"
 #include <opendht/node.h>

tools/tools_common.h

@@ -25,6 +25,13 @@
 #include <opendht/log.h>
 #include <opendht/crypto.h>
 #include <opendht/network_utils.h>
+#ifdef OPENDHT_INDEXATION
+#include <opendht/indexation/pht.h>
+#endif
+#ifdef OPENDHT_PROXY_SERVER
+#include <opendht/dht_proxy_server.h>
+#endif
+
 #ifndef WIN32_NATIVE
 #include <getopt.h>
 #include <readline/readline.h>