fix

WIP fix
http tester
2026-11-08 20:37:23 +00:00 · 2019-12-05 17:41:46 -05:00 · 2019-12-05 17:31:46 -05:00 · 2019-12-05 13:18:38 -05:00 · 2019-12-05 13:18:36 -05:00 · 2019-12-05 13:18:33 -05:00
40 changed files with 1097 additions and 846 deletions
@@ -13,19 +13,12 @@ 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 libfmt-dev libhttp-parser-dev libmsgpack-dev
+        autotools-dev autoconf libasio-dev 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
@@ -7,105 +7,83 @@ 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_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)
+option (OPENDHT_C "Build C bindings" 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 (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("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
-        )
+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 ()

-    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)
+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})
+        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()
 endif()

@@ -178,6 +156,7 @@ 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
@@ -195,7 +174,12 @@ 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
@@ -203,6 +187,7 @@ 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
@@ -218,33 +203,12 @@ 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)
@@ -358,12 +322,7 @@ if (OPENDHT_C)
        c/opendht_c.h
    )
    target_compile_definitions(opendht-c PRIVATE OPENDHT_C_BUILD)
-    if (OPENDHT_SHARED)
-        target_link_libraries(opendht-c opendht)
-    else ()
-        target_link_libraries(opendht-c opendht-static)
-    endif ()
-
+    target_link_libraries(opendht-c opendht)
    # PkgConfig module
    configure_file (
        opendht-c.pc.in
@@ -411,17 +370,15 @@ 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
@@ -430,12 +387,6 @@ 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}
@@ -450,11 +401,8 @@ if (OPENDHT_TESTS)
       ${CMAKE_THREAD_LIBS_INIT}
       ${CPPUNIT_LIBRARIES}
       ${GNUTLS_LIBRARIES}
-       ${Jsoncpp_LIBRARIES}
+       ssl crypto jsoncpp
    )
-    if (OPENDHT_PROXY_OPENSSL)
-        target_link_libraries(opendht_unit_tests ${OPENSSL_LIBRARIES})
-    endif()
    enable_testing()
    add_test(TEST opendht_unit_tests)
 endif()
@@ -122,7 +122,7 @@ void dht_publickey_delete(dht_publickey* pk) {
    delete reinterpret_cast<PubkeySp*>(pk);
 }

-int dht_publickey_export(const dht_publickey* pk, char* out, size_t* outlen) {
+int dht_publickey_pack(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,13 +164,6 @@ 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())));
@@ -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_export(const dht_publickey* pk, char* out, size_t* out_size);
+OPENDHT_C_PUBLIC int dht_publickey_pack(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,7 +56,6 @@ 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*);

@@ -35,14 +35,10 @@ 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")
-AM_COND_IF(ENABLE_INDEXATION, [
-	AC_DEFINE([OPENDHT_INDEXATION], [], [Define if DHT indexation is enabled])
+AS_IF([test "x$enable_indexation" != "xno"], [
+	AC_DEFINE([OPENDHT_INDEXATION], [1], [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"], [
@@ -139,7 +135,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"
+    CPPFLAGS+=" -DOPENDHT_JSONCPP=1"
 ], [
    AC_MSG_NOTICE([Not using JsonCpp])
    AM_COND_IF(PROXY_CLIENT_OR_SERVER, AC_MSG_ERROR(["JsonCpp is required for proxy/push notification support"]))
@@ -158,7 +154,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])
-    AC_DEFINE([OPENDHT_PROXY_HTTP_PARSER_FORK], [], [Define if using http parser fork])
+    CPPFLAGS+=" -DOPENDHT_PROXY_HTTP_PARSER_FORK"
 ], [
    AC_MSG_NOTICE([Not using http_parser fork])
 ])
@@ -209,8 +205,8 @@ AM_COND_IF([ENABLE_TOOLS], [
  ])
 ])

-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_PROXY_SERVER, [CPPFLAGS+=" -DOPENDHT_PROXY_SERVER"], [])
+AM_COND_IF(ENABLE_PROXY_CLIENT, [CPPFLAGS+=" -DOPENDHT_PROXY_CLIENT"], [])
 AM_COND_IF(ENABLE_PUSH_NOTIFICATIONS, [CPPFLAGS+=" -DOPENDHT_PUSH_NOTIFICATIONS"], [])
 AM_COND_IF(ENABLE_PROXY_SERVER_IDENTITY, [CPPFLAGS+=" -DOPENDHT_PROXY_SERVER_IDENTITY"], [])

@@ -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 libssl-dev \
+        autotools-dev autoconf libasio-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 -DFMT_TEST=Off -DFMT_DOC=Off . \
+    && cmake -DCMAKE_INSTALL_PREFIX=/usr -DBUILD_SHARED_LIBS=On . \
    && 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 install \
+    && PREFIX=/usr make -j8 && make 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/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  \
+    && 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 \
    && make install \
    && cd ../../ && rm -rf asio*

 RUN echo "*** Downloading RESTinio ***" \
    && mkdir restinio && cd restinio \
-    && wget https://github.com/aberaud/restinio/archive/a7a10e419d9089c5b8ee63f5e3098c892f22fae4.tar.gz \
-    && ls -l && tar -xzf a7a10e419d9089c5b8ee63f5e3098c892f22fae4.tar.gz \
-    && cd restinio-a7a10e419d9089c5b8ee63f5e3098c892f22fae4/dev \
+    && wget https://github.com/aberaud/restinio/archive/8d5d3e8237e0947adb9ba1ffc8281f4ad7cb2a59.tar.gz \
+    && ls -l && tar -xzf 8d5d3e8237e0947adb9ba1ffc8281f4ad7cb2a59.tar.gz \
+    && cd restinio-8d5d3e8237e0947adb9ba1ffc8281f4ad7cb2a59/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 . \
@@ -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 libfmt-dev libhttp-parser-dev libmsgpack-dev \
+        autotools-dev autoconf libasio-dev 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/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  \
+    && 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 \
    && make install \
    && cd ../../ && rm -rf asio*

@@ -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 cython3 python3-dev \
+       nettle-dev libgnutls28-dev libuv1-dev libmsgpack-dev libjsoncpp-dev libasio-dev cython3 python3-dev \
       python3-setuptools libcppunit-dev python3-pip \
-       autotools-dev autoconf libssl-dev \
+       autotools-dev autoconf libasio-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 -DFMT_TEST=Off -DFMT_DOC=Off . \
+    && cmake -DCMAKE_INSTALL_PREFIX=/usr -DBUILD_SHARED_LIBS=On . \
    && 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 install \
+    && PREFIX=/usr make -j8 && make 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/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  \
+    && 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 \
    && 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/aberaud/restinio/archive/a7a10e419d9089c5b8ee63f5e3098c892f22fae4.tar.gz \
-    && ls -l && tar -xzf a7a10e419d9089c5b8ee63f5e3098c892f22fae4.tar.gz \
-    && cd restinio-a7a10e419d9089c5b8ee63f5e3098c892f22fae4/dev \
+    && 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 \
    && 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 . \
@@ -19,4 +19,10 @@
 #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
@@ -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 {0};
+    NodeStats ipv4;
+    NodeStats ipv6;
+    size_t ongoing_ops;

 #ifdef OPENDHT_JSONCPP
    /**
@@ -122,7 +122,6 @@ struct OPENDHT_PUBLIC Config {
    ssize_t max_peer_req_per_sec {0};
 };

-#ifndef OPENDHT_LIGHT
 /**
 * SecureDht configuration.
 */
@@ -131,7 +130,6 @@ struct OPENDHT_PUBLIC SecureDhtConfig
    Config node_config {};
    crypto::Identity id {};
 };
-#endif

 static constexpr size_t DEFAULT_STORAGE_LIMIT {1024 * 1024 * 64};

@@ -160,7 +160,6 @@ 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;

    /**
@@ -58,7 +58,6 @@ public:
    MSGPACK_DEFINE(service, data)
 };

-#ifndef OPENDHT_LIGHT
 template <typename T>
 class OPENDHT_PUBLIC SignedValue : public Value::Serializable<T>
 {
@@ -103,6 +102,7 @@ public:



+
 class OPENDHT_PUBLIC ImMessage : public SignedValue<ImMessage>
 {
 private:
@@ -177,8 +177,16 @@ 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)
@@ -192,7 +200,6 @@ public:
    Value::Id id {0};
    Blob ice_data;
 };
-#endif

 /* "Peer" announcement
 */
@@ -254,11 +261,8 @@ 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;

@@ -53,9 +53,9 @@ public:
    DhtProxyClient();

    explicit DhtProxyClient(
-        std::shared_ptr<crypto::Certificate> serverCA, crypto::Identity clientIdentity,
+        std::shared_ptr<dht::crypto::Certificate> serverCA, dht::crypto::Identity clientIdentity,
        std::function<void()> loopSignal, const std::string& serverHost,
-        const std::string& pushClientId = "", std::shared_ptr<Logger> logger = {});
+        const std::string& pushClientId = "", std::shared_ptr<dht::Logger> logger = {});

    void setHeaderFields(http::Request& request);

@@ -404,7 +404,7 @@ private:
 #endif // OPENDHT_PUSH_NOTIFICATIONS

    Json::StreamWriterBuilder jsonBuilder_;
-    std::unique_ptr<Json::CharReader> jsonReader_;
+    Json::CharReaderBuilder jsonReaderBuilder_;
    std::shared_ptr<dht::Logger> logger_;

    std::shared_ptr<http::Request> buildRequest(const std::string& target = {});
@@ -31,7 +31,10 @@

 #include <restinio/all.hpp>
 #include <restinio/tls.hpp>
+
+#ifdef OPENDHT_JSONCPP
 #include <json/json.h>
+#endif

 #include <memory>
 #include <mutex>
@@ -51,6 +54,8 @@ 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
@@ -82,15 +87,15 @@ public:

    struct ServerStats {
        /** Current number of listen operations */
-        size_t listenCount {0};
+        size_t listenCount;
        /** Current number of permanent put operations */
-        size_t putCount {0};
+        size_t putCount;
        /** Current number of push tokens with at least one listen operation */
-        size_t pushListenersCount {0};
+        size_t pushListenersCount;
        /** Average requests per second */
-        double requestRate {0};
+        double requestRate;
        /** Node Info **/
-        NodeInfo nodeInfo {};
+        NodeInfo nodeInfo;

        std::string toString() const {
            std::ostringstream ss;
@@ -108,6 +113,7 @@ public:
            return ss.str();
        }

+#ifdef OPENDHT_JSONCPP
        /**
         * Build a json object from a NodeStats
         */
@@ -120,6 +126,7 @@ public:
            result["nodeInfo"] = nodeInfo.toJson();
            return result;
        }
+#endif
    };

    ServerStats stats() const { return stats_; }
@@ -99,7 +99,7 @@ public:
    asio::streambuf& input();
    std::istream& data() { return istream_; }

-    std::string read_bytes(size_t bytes = 0);
+    std::string read_bytes(size_t bytes);
    std::string read_until(const char delim);

    void async_connect(std::vector<asio::ip::tcp::endpoint>&& endpoints, ConnectHandlerCb);
@@ -108,20 +108,11 @@ 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_;

@@ -241,7 +232,7 @@ public:

    ~Request();

-    inline unsigned int id() const { return  id_; };
+    unsigned int id() const;
    void set_connection(std::shared_ptr<Connection> connection);
    std::shared_ptr<Connection> get_connection() const;
    inline const Url& get_url() const {
@@ -269,7 +260,6 @@ 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();

@@ -282,14 +272,19 @@ 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(State state);
+    void notify_state_change(const State state);

    void build();

@@ -306,11 +301,19 @@ 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, size_t bytes);
+
+    void handle_response_body(const asio::error_code& ec, size_t bytes);

    void onHeadersComplete();
-    void onBody(const char* at, size_t length);
-    void onComplete();
+
+    /**
+     * 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);

    std::shared_ptr<dht::Logger> logger_;

@@ -319,6 +322,7 @@ 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,6 +340,7 @@ 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_;
@@ -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);
 }

 /**
@@ -71,13 +71,13 @@ OPENDHT_PUBLIC void
 printLog(std::ostream &s, char const *m, va_list args);

 OPENDHT_PUBLIC
-std::shared_ptr<Logger> getStdLogger();
+std::unique_ptr<Logger> getStdLogger();

 OPENDHT_PUBLIC
-std::shared_ptr<Logger> getFileLogger(const std::string &path);
+std::unique_ptr<Logger> getFileLogger(const std::string &path);

 OPENDHT_PUBLIC
-std::shared_ptr<Logger> getSyslogLogger(const char* name);
+std::unique_ptr<Logger> getSyslogLogger(const char* name);

 OPENDHT_PUBLIC void
 enableLogging(dht::DhtRunner &dht);
@@ -22,13 +22,10 @@
 #include "def.h"
 #include "sockaddr.h"
 #include "infohash.h"
-#include "log_enable.h"

 #include <thread>
-
-namespace asio {
-class io_context;
-}
+#include <mutex>
+#include <condition_variable>

 namespace dht {

@@ -38,7 +35,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, std::shared_ptr<asio::io_context> ioContext = {}, std::shared_ptr<Logger> logger = {});
+    PeerDiscovery(in_port_t port = DEFAULT_PORT);
    ~PeerDiscovery();

    /**
@@ -54,7 +51,7 @@ public:
    }

    /**
-     * startPublish - Keeping sending data until node is joinned or stop is called
+     * startPublish - Keeping sending data until node is joinned or stop is called - msgpack
    */
    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);
@@ -82,12 +79,20 @@ 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_;
 };

 }
@@ -20,7 +20,7 @@
 #pragma once

 #include "infohash.h"
-//#include "crypto.h"
+#include "crypto.h"
 #include "utils.h"
 #include "sockaddr.h"

@@ -46,12 +46,6 @@ 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.
@@ -247,9 +241,15 @@ struct OPENDHT_PUBLIC Value
        };
    }

-    static Filter OwnerFilter(const crypto::PublicKey& pk);
+    static Filter OwnerFilter(const crypto::PublicKey& pk) {
+        return OwnerFilter(pk.getId());
+    }

-    static Filter OwnerFilter(const InfoHash& pkh);
+    static Filter OwnerFilter(const InfoHash& pkh) {
+        return [pkh](const Value& v) {
+            return v.owner and v.owner->getId() == pkh;
+        };
+    }

    static Filter SeqNumFilter(uint16_t seq_no) {
        return [seq_no](const Value& v) {
@@ -341,13 +341,20 @@ 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);
+    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());
+    }

    /**
     * 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;
+    bool checkSignature() const {
+        return isSigned() and owner->checkSignature(getToSign(), signature);
+    }

    std::shared_ptr<const crypto::PublicKey> getOwner() const {
        return std::static_pointer_cast<const crypto::PublicKey>(owner);
@@ -356,7 +363,15 @@ 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);
+    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() {}

@@ -409,11 +424,7 @@ struct OPENDHT_PUBLIC Value
    inline bool operator== (const Value& o) {
        return id == o.id &&
        (isEncrypted() ? cypher == o.cypher :
-        (
-#ifndef OPENDHT_LIGHT
-            (owner == o.owner || *owner == *o.owner) &&
-#endif
-            type == o.type && data == o.data && user_type == o.user_type && signature == o.signature));
+        ((owner == o.owner || *owner == *o.owner) && type == o.type && data == o.data && user_type == o.user_type && signature == o.signature));
    }

    void setRecipient(const InfoHash& r) {
@@ -471,8 +482,7 @@ struct OPENDHT_PUBLIC Value
    template <typename Packer>
    void msgpack_pack_to_sign(Packer& pk) const
    {
-#ifndef OPENDHT_LIGHT
-        bool has_owner = owner && *owner
+        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);
@@ -481,9 +491,6 @@ 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());
@@ -528,11 +535,9 @@ 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:
@@ -27,6 +27,7 @@ libopendht_la_SOURCES  = \
        dhtrunner.cpp \
        default_types.cpp \
        log.cpp \
+        peer_discovery.cpp \
        network_utils.cpp \
        thread_pool.cpp

@@ -55,6 +56,7 @@ nobase_include_HEADERS = \
        ../include/opendht/default_types.h \
        ../include/opendht/log.h \
        ../include/opendht/log_enable.h \
+        ../include/opendht/peer_discovery.h \
        ../include/opendht/network_utils.h \
        ../include/opendht/rng.h \
        ../include/opendht/thread_pool.h
@@ -75,11 +77,6 @@ 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
@@ -1,25 +1,3 @@
-/*
- *  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 {
@@ -348,7 +348,9 @@ PrivateKey::serialize(const std::string& password) const
    size_t buf_sz = 8192;
    Blob buffer;
    buffer.resize(buf_sz);
-    auto err = serialize(buffer.data(), &buf_sz, password);
+    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);
    if (err != GNUTLS_E_SUCCESS) {
        std::cerr << "Could not export private key - " << gnutls_strerror(err) << std::endl;
        return {};
@@ -357,16 +359,6 @@ 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
 {
@@ -15,9 +15,6 @@
 *  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"

@@ -86,8 +83,6 @@ 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)};
@@ -101,13 +96,6 @@ 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
@@ -76,9 +76,8 @@ 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();
    }
 }
@@ -24,7 +24,6 @@
 #include "utils.h"

 #include <http_parser.h>
-#include <deque>

 namespace dht {

@@ -70,44 +69,22 @@ 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)
-    : proxyUrl_(serverHost)
-    , clientIdentity_(clientIdentity), serverCertificate_(serverCA)
-    , pushClientId_(pushClientId), loopSignal_(signal)
-    , jsonReader_(Json::CharReaderBuilder{}.newCharReader())
-    , logger_(logger)
+    :
+        clientIdentity_(clientIdentity), serverCertificate_(serverCA),
+        pushClientId_(pushClientId), loopSignal_(signal), 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());
@@ -309,17 +286,18 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Va
        auto opstate = std::make_shared<OperationState>();
        Value::Filter filter = w.empty() ? f : f.chain(w.getFilter());

-        auto rxBuf = std::make_shared<LineSplit>();
-        request->add_on_body_callback([this, key, opstate, filter, rxBuf, cb](const char* at, size_t length){
+        request->add_on_body_callback([this, key, opstate, filter, cb](const char* at, size_t length){
            try {
-                auto& b = *rxBuf;
-                b.append(at, length);
+                auto body = std::string(at, length);
                // one value per body line
-                while (b.getLine('\n') and !opstate->stop){
+                std::string data_line;
+                std::stringstream body_stream(body);
+                while (std::getline(body_stream, data_line, '\n') and !(opstate->stop.load())){
                    std::string err;
                    Json::Value json;
-                    const auto& line = b.line();
-                    if (!jsonReader_->parse(line.data(), line.data() + line.size(), &json, &err)){
+                    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)){
                        opstate->ok.store(false);
                        return;
                    }
@@ -342,27 +320,30 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Va
                opstate->ok.store(false);
            }
        });
-        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);
-                    });
+        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);
                }
-                loopSignal_();
-            }
-            if (not isDestroying_) {
-                std::lock_guard<std::mutex> l(requestLock_);
-                requests_.erase(reqid);
            }
        });
        {
@@ -488,19 +469,22 @@ DhtProxyClient::doPut(const InfoHash& key, Sp<Value> val, DoneCallbackSimple cb,
            }
        }
        request->set_body(Json::writeString(jsonBuilder_, json));
-        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);
+        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);
+                }
            }
        });
        {
@@ -600,29 +584,33 @@ 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_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);
+        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())
                    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);
            }
        });

@@ -886,17 +874,20 @@ 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_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);
+                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);
+                        }
                    }
                });
                {
@@ -950,18 +941,19 @@ DhtProxyClient::sendListen(const restinio::http_request_header_t header,
            body = fillBody(method == ListenMethod::RESUBSCRIBE);
        request->set_body(body);
 #endif
-        auto rxBuf = std::make_shared<LineSplit>();
-        request->add_on_body_callback([this, reqid, opstate, rxBuf, cb](const char* at, size_t length){
-            try {
-                auto& b = *rxBuf;
-                b.append(at, length);

+        request->add_on_body_callback([this, reqid, opstate, cb](const char* at, size_t length){
+            try {
+                auto body = std::string(at, length);
                // one value per body line
-                while (b.getLine('\n') and !opstate->stop){
+                std::string data_line;
+                std::stringstream body_stream(body);
+                while (std::getline(body_stream, data_line, '\n') and !opstate->stop){
                    std::string err;
                    Json::Value json;
-                    const auto& line = b.line();
-                    if (!jsonReader_->parse(line.data(), line.data() + line.size(), &json, &err)){
+                    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)){
                        opstate->ok.store(false);
                        return;
                    }
@@ -988,18 +980,21 @@ DhtProxyClient::sendListen(const restinio::http_request_header_t header,
                opstate->ok.store(false);
            }
        });
-        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);
+        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);
+                }
            }
        });
        {
@@ -18,10 +18,6 @@
 *  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"
@@ -72,9 +68,6 @@ 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:
@@ -963,7 +956,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()){
@@ -1009,6 +1002,7 @@ 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){
@@ -18,16 +18,11 @@
 *  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 "network_utils.h"
-#ifdef OPENDHT_PEER_DISCOVERY
 #include "peer_discovery.h"
-#endif
+#include "network_utils.h"
+
 #ifdef OPENDHT_PROXY_CLIENT
 #include "dht_proxy_client.h"
 #endif
@@ -129,8 +124,7 @@ DhtRunner::run(const Config& config, Context&& context)
        {
            std::lock_guard<std::mutex> lck(sock_mtx);
            if (rcv.size() >= RX_QUEUE_MAX_SIZE) {
-                if (logger_)
-                    logger_->e("Dropping packet: queue is full!");
+                std::cerr << "Dropping packet: queue is full!" << std::endl;
                rcv.pop();
            }
            rcv.emplace(std::move(pkt));
@@ -190,16 +184,11 @@ 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){
@@ -228,7 +217,6 @@ DhtRunner::run(const Config& config, Context&& context)
            peerDiscovery_->startPublish(AF_INET6, PEER_DISCOVERY_DHT_SERVICE, sbuf_node);
        }
    }
-#endif
 }

 void
@@ -304,10 +292,8 @@ 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();
@@ -321,6 +307,9 @@ 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)();
@@ -476,14 +465,12 @@ DhtRunner::getNodesStats(sa_family_t af) const
 NodeInfo
 DhtRunner::getNodeInfo() const {
    std::lock_guard<std::mutex> lck(dht_mtx);
-    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);
-    }
+    NodeInfo info;
+    info.id = getId();
+    info.node_id = getNodeId();
    info.ongoing_ops = ongoing_ops;
+    info.ipv4 = dht_->getNodesStats(AF_INET);
+    info.ipv6 = dht_->getNodesStats(AF_INET6);
    return info;
 }

@@ -29,8 +29,15 @@
 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)
 {
@@ -123,7 +130,8 @@ 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();
 }

@@ -192,8 +200,6 @@ 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);
@@ -213,9 +219,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), wrapCallabck(std::move(cb)));
+        asio::async_connect(ssl_socket_->lowest_layer(), std::move(endpoints), cb);
    else if (socket_)
-        asio::async_connect(*socket_, std::move(endpoints), wrapCallabck(std::move(cb)));
+        asio::async_connect(*socket_, std::move(endpoints), cb);
    else if (cb)
        cb(asio::error::operation_aborted, {});
 }
@@ -255,8 +261,8 @@ void
 Connection::async_write(BytesHandlerCb cb)
 {
    if (!is_open())   return;
-    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)));
+    if (ssl_socket_)  asio::async_write(*ssl_socket_, write_buf_, cb);
+    else if (socket_) asio::async_write(*socket_, write_buf_, cb);
    else if (cb)      cb(asio::error::operation_aborted, 0);
 }

@@ -264,8 +270,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, wrapCallabck(std::move(cb)));
-    else if (socket_) asio::async_read_until(*socket_, read_buf_, delim, wrapCallabck(std::move(cb)));
+    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);
    else if (cb)      cb(asio::error::operation_aborted, 0);
 }

@@ -273,8 +279,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, wrapCallabck(std::move(cb)));
-    else if (socket_) asio::async_read_until(*socket_, read_buf_, delim, wrapCallabck(std::move(cb)));
+    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);
    else if (cb)      cb(asio::error::operation_aborted, 0);
 }

@@ -282,27 +288,11 @@ 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), wrapCallabck(std::move(cb)));
-    else if (socket_) asio::async_read(*socket_, read_buf_, asio::transfer_exactly(bytes), wrapCallabck(std::move(cb)));
+    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);
    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)
 {
@@ -448,9 +438,10 @@ 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_done_callback([this, jsoncb](const Response& response){
+    add_on_state_change_callback([this, jsoncb](State state, const Response& response){
+        if (state != Request::State::DONE)
+            return;
        Json::Value json;
        if (response.status_code != 0) {
            std::string err;
@@ -460,7 +451,7 @@ Request::Request(asio::io_context& ctx, const std::string& url, const Json::Valu
                logger_->e("[http:request:%i] can't parse response to json", id_, err.c_str());
        }
        if (jsoncb)
-            jsoncb(std::move(json), response.status_code);
+            jsoncb(json, response.status_code);
    });
 }

@@ -475,7 +466,10 @@ Request::Request(asio::io_context& ctx, const std::string& url, OnDoneCb onDone,
    : 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));
+    add_on_state_change_callback([this, onDone](State state, const Response& response){
+        if (state == Request::State::DONE)
+            onDone(response);
+    });
 }

 Request::Request(asio::io_context& ctx, const std::string& host, const std::string& service,
@@ -529,28 +523,39 @@ Request::cancel()
        c->close();
 }

+unsigned int
+Request::id() const
+{
+    return id_;
+}
+
 void
-Request::set_connection(std::shared_ptr<Connection> connection) {
-    conn_ = std::move(connection);
+Request::set_connection(std::shared_ptr<Connection> connection)
+{
+    conn_ = 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;
 }

@@ -561,27 +566,32 @@ 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);
 }

@@ -640,30 +650,29 @@ Request::build()
 }

 void
-Request::add_on_status_callback(OnStatusCb cb) {
+Request::add_on_status_callback(OnStatusCb cb)
+{
+    std::lock_guard<std::mutex> lock(cbs_mutex_);
    cbs_.on_status = std::move(cb);
 }

 void
-Request::add_on_body_callback(OnDataCb cb) {
+Request::add_on_body_callback(OnDataCb cb)
+{
+    std::lock_guard<std::mutex> lock(cbs_mutex_);
    cbs_.on_body = std::move(cb);
 }

 void
-Request::add_on_state_change_callback(OnStateChangeCb cb) {
+Request::add_on_state_change_callback(OnStateChangeCb cb)
+{
+    std::lock_guard<std::mutex> lock(cbs_mutex_);
    cbs_.on_state_change = std::move(cb);
 }

 void
-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) {
+Request::notify_state_change(const State state)
+{
    state_ = state;
    if (cbs_.on_state_change)
        cbs_.on_state_change(state, response_);
@@ -680,20 +689,44 @@ Request::init_parser()
    if (!parser_s_)
        parser_s_ = std::make_unique<http_parser_settings>();
    http_parser_settings_init(parser_s_.get());
-
-    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);
-    };
-
+    {
+        // 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){
+            std::cout << "on_status " << status_code << std::endl;
+            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) {
+            std::cout << "on_header_field " << std::string(at, length) << std::endl;
+            *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) {
+            std::cout << "on_header_value " << std::string(at, length) << std::endl;
+            response_.headers[*header_field] = std::string(at, length);
+            if (headerValueCb)
+                headerValueCb(at, length);
+        };
+        cbs_.on_headers_complete = [this](){
+            std::cout << "on_headers_complete " << std::endl;
+            onHeadersComplete();
+        };
+        cbs_.on_body = [bodyCb = std::move(cbs_.on_body)](const char* at, size_t length) {
+            std::cout << "on_body " << length << std::endl;
+            if (bodyCb)
+                bodyCb(at, length);
+        };
+        cbs_.on_message_complete = [this](){
+            std::cout << "on_message_complete " << std::endl;
+            if (logger_)
+                logger_->d("[http:request:%i] response: message complete", id_);
+            message_complete_.store(true);
+        };
+    }
    // 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);
@@ -708,15 +741,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)->onBody(at, length);
+        static_cast<Request*>(parser->data)->cbs_.on_body(at, length);
        return 0;
    };
    parser_s_->on_headers_complete = [](http_parser* parser) -> int {
-        static_cast<Request*>(parser->data)->onHeadersComplete();
+        static_cast<Request*>(parser->data)->cbs_.on_headers_complete();
        return 0;
    };
    parser_s_->on_message_complete = [](http_parser* parser) -> int {
-        static_cast<Request*>(parser->data)->onComplete();
+        static_cast<Request*>(parser->data)->cbs_.on_message_complete();
        return 0;
    };
 }
@@ -758,12 +791,11 @@ Request::connect(std::vector<asio::ip::tcp::endpoint>&& endpoints, HandlerCb cb)
            this_.terminate(ec);
            return;
        }
-        else if (ec) {
+        else if (ec and this_.logger_)
+            this_.logger_->e("[http:request:%i] connect: failed with all endpoints", this_.id_);
+        else {
            if (this_.logger_)
-                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_);
+                this_.logger_->d("[http:request:%i] connect success", this_.id_);

            const auto& url = this_.get_url();
            auto port = endpoint.port();
@@ -784,8 +816,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);
                    });
@@ -842,7 +874,7 @@ Request::post()
    init_parser();

    if (logger_)
-        logger_->d("[http:request:%i] sending %zu bytes", id_, request_.size());
+        logger_->d("[http:request:%i] send:\n%s", id_, request_.c_str());

    // write the request to buffer
    std::ostream request_stream(&conn_->input());
@@ -864,16 +896,20 @@ 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_) {
-        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 (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 (auto c = conn_)
            c->close();
    notify_state_change(State::DONE);
@@ -890,77 +926,259 @@ 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 n_bytes){
        if (auto sthis = wthis.lock())
-            sthis->handle_response(ec, n_bytes);
+            sthis->handle_response_header(ec, n_bytes);
    });
 }

 void
-Request::handle_response(const asio::error_code& ec, size_t n_bytes)
+Request::handle_response_header(const asio::error_code& ec, size_t n_bytes)
 {
    if (ec && ec != asio::error::eof){
        terminate(ec);
        return;
    }
-    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 (!conn_->is_open()){
+        terminate(asio::error::not_connected);
+        return;
+    }
+    const auto& dat = conn_->read_bytes(n_bytes);
+    if (logger_)
+        logger_->d("[http:request:%i] read %zu bytes:\n\"%.*s\"", id_, n_bytes, dat.size(), dat.data());
+
+    std::cout << "\n " << std::hex
+     << std::setfill('0') << std::setw(2) << (unsigned)dat[dat.size() - 4]
+     << std::setfill('0') << std::setw(2) << (unsigned)dat[dat.size() - 3]
+     << std::setfill('0') << std::setw(2) << (unsigned)dat[dat.size() - 2]
+     << std::setfill('0') << std::setw(2) << (unsigned)dat[dat.size() - 1]
+        << std::endl;
+
+    // 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);*/
+    /*auto& dat = conn_->data();
+    std::vector<uint8_t> indat;*/
+    auto ret = http_parser_execute(parser_.get(), parser_s_.get(), dat.data(), dat.size());
+    if (ret != dat.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;
+        terminate(asio::error::invalid_argument);
    }
-
-    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);
+    logger_->d("[http:request:%i] read END", id_);
 }

 void
 Request::onHeadersComplete() {
    notify_state_change(State::HEADER_RECEIVED);

+    std::weak_ptr<Request> wthis = shared_from_this();
+
    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){
        notify_state_change(State::SENDING);
        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()) {
+        size_t content_length = atoi(content_length_it->second.c_str());
+        auto dat = conn_->read_bytes(content_length);
+        /*if (logger_)
+            logger_->d("[http:request:%i] response body: %zu/%zu bytes received:\n\"%.*s\"", id_, dat.size(), content_length, dat.size(), dat.data());
+        if (not dat.empty()) {
+            if (dat.size() == content_length)
+                dat += HTTP_HEADER_DELIM;
+            http_parser_execute(parser_.get(), parser_s_.get(), dat.data(), dat.size());
+        }*/
+        if (dat.size() == content_length)
+            ctx_.post([wthis, dat=std::move(dat)]{
+                if (auto sthis = wthis.lock()) {
+                    http_parser_execute(sthis->parser_.get(), sthis->parser_s_.get(), dat.data(), dat.size());
+                }
+            });
+        else
+            conn_->async_read(content_length, [wthis](const asio::error_code& ec, size_t bytes){
+                if (auto sthis = wthis.lock())
+                    sthis->handle_response_header(ec, bytes);
+            });
+        /*auto r = conn_->read_bytes(content_length);
+        std::cout << "Has content length " << content_length << std::endl;
+        std::cout << "Read content " << r.size() << std::endl;
+        if (r.size() == content_length) {
+            parse_request(r);
+        }*/
+        /*
+        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_header(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_header(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_header(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
@@ -56,9 +56,9 @@ printLog(std::ostream& s, char const *m, va_list args) {
    s << std::endl;
 }

-std::shared_ptr<Logger>
+std::unique_ptr<Logger>
 getStdLogger() {
-    return std::make_shared<Logger>(
+    return std::unique_ptr<Logger>(new 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::shared_ptr<Logger>
+std::unique_ptr<Logger>
 getFileLogger(const std::string &path) {
    auto logfile = std::make_shared<std::ofstream>();
    logfile->open(path, std::ios::out);

-    return std::make_shared<Logger>(
+    return std::unique_ptr<Logger>(new 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::shared_ptr<Logger>
+std::unique_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::make_shared<Logger>(
+    return std::unique_ptr<Logger>(new 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::make_shared<Logger>();
+    return std::unique_ptr<Logger>(new Logger());
 #endif
 }

@@ -2,7 +2,6 @@
 *  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
@@ -22,166 +21,402 @@
 #include "network_utils.h"
 #include "utils.h"

-#include <asio.hpp>
+#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>

-using namespace std::chrono_literals;
+#ifndef IPV6_JOIN_GROUP
+#define IPV6_JOIN_GROUP IPV6_ADD_MEMBERSHIP
+#endif

 namespace dht {

 // Organization-local Scope multicast
-constexpr char MULTICAST_ADDRESS_IPV4[] = "239.192.0.1";
-constexpr char MULTICAST_ADDRESS_IPV6[] = "ff08::101";
+constexpr char MULTICAST_ADDRESS_IPV4[12] = "239.192.0.1";
+constexpr char MULTICAST_ADDRESS_IPV6[10] = "ff08::101";

 class PeerDiscovery::DomainPeerDiscovery
 {
 public:
-    DomainPeerDiscovery(asio::ip::udp domain, in_port_t port, Sp<asio::io_context> ioContext = {}, Sp<Logger> logger = {});
+    DomainPeerDiscovery(sa_family_t domain, in_port_t port);
    ~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::shared_ptr<asio::io_context> ioContext_;
-    asio::ip::udp::socket sockFd_;
-    asio::ip::udp::endpoint sockAddrSend_;
-    asio::steady_timer publishTimer_;
+    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::array<char, 64 * 1024> receiveBuf_;
-    asio::ip::udp::endpoint receiveFrom_;
+    SockAddr sockAddrSend_;
+
+    //Thread export to be joined
+    std::thread listening_;
+    std::thread sending_;

    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(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)
+PeerDiscovery::DomainPeerDiscovery::DomainPeerDiscovery(sa_family_t domain, in_port_t port)
+    : domain_(domain), port_(port), sockFd_(init(domain))
 {
-    sockFd_.set_option(asio::ip::multicast::join_group(sockAddrSend_.address()));
-    sockFd_.set_option(asio::ip::udp::socket::reuse_address(true));
-    sockFd_.bind({domain, port});
+    joinMulticast(sockFd_, domain);
+    setupListener();
+    setupSender();
 }

 PeerDiscovery::DomainPeerDiscovery::~DomainPeerDiscovery()
 {
    stop();
-    sockFd_.close();
+    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;
+    }
+    }
 }

 void
 PeerDiscovery::DomainPeerDiscovery::startDiscovery(const std::string &type, ServiceDiscoveredCallback callback)
 {
-    std::lock_guard<std::mutex> lck(dmtx_);
+    std::unique_lock<std::mutex> lck(dmtx_);
    callbackmap_[type] = callback;
    if (not drunning_) {
+        if (listening_.joinable())
+            listening_.join();
        drunning_ = true;
-        ioContext_->post(std::bind(&PeerDiscovery::DomainPeerDiscovery::loopListener, this));
+        listening_ = std::thread(&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()
 {
-    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();
+    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];

-            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)
+    while (true) {
+        {
+            std::lock_guard<std::mutex> lck(dmtx_);
+            if (not drunning_)
+                break;
+        }
+
+        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)
                    continue;
-                auto key = o.key.as<std::string>();
-                ServiceDiscoveredCallback cb;
-                {
-                    std::lock_guard<std::mutex> lck(dmtx_);
-                    if (drunning_) {
+                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 callback = callbackmap_.find(key);
                        if (callback != callbackmap_.end())
                            cb = callback->second;
-                    } else
-                        return;
+                    }
+                    if (cb)
+                        cb(std::move(o.val), std::move(rcv.first));
                }
-                if (cb)
-                    cb(std::move(o.val), SockAddr{ receiveFrom_.data(), (socklen_t)receiveFrom_.size() });
+            } catch (const std::exception& e) {
+                std::cerr << "Error receiving packet: " << e.what() << std::endl;
            }
-        } catch (const std::exception& e) {
-            if (logger_)
-                logger_->e("Error receiving packet: %s", e.what());
        }
-        loopListener();
-    });
+    }
+    if (stop_readfd != -1)
+        close(stop_readfd);
+    if (stopWriteFd_ != -1) {
+        close(stopWriteFd_);
+        stopWriteFd_ = -1;
+    }
 }

 void
-PeerDiscovery::DomainPeerDiscovery::publish()
+PeerDiscovery::DomainPeerDiscovery::setupSender()
 {
-    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));
-    }
+    // Setup sender address
+    sockAddrSend_.setFamily(domain_);
+    sockAddrSend_.setAddress(domain_ == AF_INET ? MULTICAST_ADDRESS_IPV4 : MULTICAST_ADDRESS_IPV6);
+    sockAddrSend_.setPort(port_);
 }

 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::lock_guard<std::mutex> lck(mtx_);
+    std::unique_lock<std::mutex> lck(mtx_);
    messages_[type] = std::move(pack_buf_c);
    reloadMessages();
    if (not lrunning_) {
+        if (sending_.joinable())
+            sending_.join();
        lrunning_ = true;
-        ioContext_->post(std::bind(&PeerDiscovery::DomainPeerDiscovery::publish, this));
+        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;
+            }
+        });
    }
 }

 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) {
@@ -193,7 +428,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) {
@@ -209,8 +444,20 @@ PeerDiscovery::DomainPeerDiscovery::stopPublish(const std::string& type)
 void
 PeerDiscovery::DomainPeerDiscovery::stopDiscovery()
 {
-    if (drunning_)
+#ifdef _WIN32
+#define write(s, b, f) send((s), (b), (f), 0)
+#endif
+    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
@@ -218,7 +465,7 @@ PeerDiscovery::DomainPeerDiscovery::stopPublish()
 {
    if (lrunning_) {
        lrunning_ = false;
-        publishTimer_.cancel();
+        cv_.notify_all();
    }
 }

@@ -226,11 +473,11 @@ void
 PeerDiscovery::DomainPeerDiscovery::stop()
 {
    {
-        std::lock_guard<std::mutex> lck(dmtx_);
+        std::unique_lock<std::mutex> lck(dmtx_);
        stopDiscovery();
    }
    {
-        std::lock_guard<std::mutex> lck(mtx_);
+        std::unique_lock<std::mutex> lck(mtx_);
        stopPublish();
    }
 }
@@ -247,65 +494,38 @@ PeerDiscovery::DomainPeerDiscovery::reloadMessages()
    }
 }

-PeerDiscovery::PeerDiscovery(in_port_t port, Sp<asio::io_context> ioContext, Sp<Logger> logger)
+PeerDiscovery::PeerDiscovery(in_port_t port)
 {
-    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(asio::ip::udp::v4(), port, ioContext, logger));
+        peerDiscovery4_.reset(new DomainPeerDiscovery(AF_INET, port));
    } catch(const std::exception& e){
-        if (logger)
-            logger->e("[peerdiscovery] can't start IPv4: %s", e.what());
+        std::cerr << "Can't start peer discovery (IPv4): " << e.what() << std::endl;
    }
    try {
-        peerDiscovery6_.reset(new DomainPeerDiscovery(asio::ip::udp::v6(), port, ioContext, logger));
+        peerDiscovery6_.reset(new DomainPeerDiscovery(AF_INET6, port));
    } catch(const std::exception& e) {
-        if (logger)
-            logger->e("[peerdiscovery] can't start IPv6: %s", e.what());
+        std::cerr << "Can't start peer discovery (IPv6): " << e.what() << std::endl;
    }
 }

-PeerDiscovery::~PeerDiscovery() {
-    stop();
-    if (ioContext_)
-        ioContext_->stop();
-    if (ioRunnner_.joinable())
-        ioRunnner_.join();
-}
+PeerDiscovery::~PeerDiscovery(){}

 void
-PeerDiscovery::startDiscovery(const std::string& type, ServiceDiscoveredCallback callback)
+PeerDiscovery::startDiscovery(const std::string &type, ServiceDiscoveredCallback callback)
 {
-    if (peerDiscovery4_) peerDiscovery4_->startDiscovery(type, callback);
-    if (peerDiscovery6_) peerDiscovery6_->startDiscovery(type, callback);
+    if(peerDiscovery4_) peerDiscovery4_->startDiscovery(type, callback);
+    if(peerDiscovery6_) peerDiscovery6_->startDiscovery(type, callback);
 }

 void
-PeerDiscovery::startPublish(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);
+    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)
+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);
@@ -317,8 +537,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
@@ -338,7 +558,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);
@@ -348,4 +568,10 @@ PeerDiscovery::stopPublish(sa_family_t domain, const std::string& type)
    return false;
 }

+void
+PeerDiscovery::join() {
+    if(peerDiscovery4_) peerDiscovery4_->join();
+    if(peerDiscovery6_) peerDiscovery6_->join();
+}
+
 }
@@ -17,9 +17,6 @@
 *  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"
@@ -17,17 +17,10 @@
 *  along with this program. If not, see <https://www.gnu.org/licenses/>.
 */

-#ifdef HAVE_CONFIG_H
-#include "config.h"
-#endif
-
 #include "value.h"
-#include "default_types.h"
-#include "callbacks.h"

-#ifndef OPENDHT_LIGHT
+#include "default_types.h"
 #include "securedht.h" // print certificate ID
-#endif

 #ifdef OPENDHT_JSONCPP
 #include "base64.h"
@@ -45,46 +38,6 @@ 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());
@@ -99,9 +52,7 @@ std::ostream& operator<< (std::ostream& s, const Value& v)
    if (not v.isEncrypted()) {
        if (v.type == IpServiceAnnouncement::TYPE.id) {
            s << IpServiceAnnouncement(v.data);
-        }
-#ifndef OPENDHT_LIGHT
-        else if (v.type == CERTIFICATE_TYPE.id) {
+        } else if (v.type == CERTIFICATE_TYPE.id) {
            s << "Certificate";
 #ifdef OPENDHT_LOG_CRT_ID
            try {
@@ -111,9 +62,7 @@ std::ostream& operator<< (std::ostream& s, const Value& v)
                s << " (invalid)";
            }
 #endif
-        }
-#endif
-        else {
+        } else {
            s << "Data (type: " << v.type << " ): ";
            s << std::hex;
            for (auto i : v.data)
@@ -210,12 +159,9 @@ 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>();
            }
@@ -244,13 +190,11 @@ 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);
@@ -370,11 +314,9 @@ 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;
@@ -0,0 +1 @@
+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
@@ -19,6 +19,7 @@

 #include "httptester.h"

+#include <opendht/log.h>
 #include <opendht/value.h>
 #include <opendht/dhtrunner.h>

@@ -32,6 +33,8 @@ CPPUNIT_TEST_SUITE_REGISTRATION(HttpTester);

 void
 HttpTester::setUp() {
+    logger = dht::log::getStdLogger();
+
    nodePeer = std::make_shared<dht::DhtRunner>();
    nodePeer->run(0);

@@ -41,7 +44,7 @@ HttpTester::setUp() {

    serverProxy = std::unique_ptr<dht::DhtProxyServer>(
        new dht::DhtProxyServer(
-            /*http*/dht::crypto::Identity{}, nodeProxy, 8080, /*pushServer*/"127.0.0.1:8090"));
+            /*http*/dht::crypto::Identity{}, nodeProxy, 8080, /*pushServer*/"127.0.0.1:8090", logger));
 }

 void
@@ -228,72 +231,66 @@ HttpTester::test_send_json() {
    std::mutex cv_m;
    std::unique_lock<std::mutex> lk(cv_m);
    bool done = false;
-    unsigned int status = 0;

-    auto json = dht::Value("hey").toJson();
+    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";
    // Act
-    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();
-        });
+    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 = std::move(value);
+        status = status_code;
+        done = true;
+        cv.notify_all();
+    }, logger);
+    request->set_method(restinio::http_method_post());
    request->send();
-
    // Assert
    CPPUNIT_ASSERT(cv.wait_for(lk, std::chrono::seconds(10), [&]{ return done; }));
-    CPPUNIT_ASSERT_EQUAL(200u, status);
-    CPPUNIT_ASSERT_EQUAL(json["data"].asString(), resp_val["data"].asString());
+    CPPUNIT_ASSERT(status == 200);
+    CPPUNIT_ASSERT(resp_val["data"] == json["data"]);
+
    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 = std::make_shared<dht::http::Request>(serverProxy->io_context(), "http://google.ca",
+                   [&](const dht::http::Response& response){
+        logger->w("got answer: %.*s", response.body.size(), response.body.data());
+        status = response.status_code;
+        done = true;
+        cv.notify_all();
+    }, logger);
    request->send();
-
    CPPUNIT_ASSERT(cv.wait_for(lk, std::chrono::seconds(10), [&]{ return done; }));
-    CPPUNIT_ASSERT(status != 0);
+    //CPPUNIT_ASSERT(status == 200);
+
    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 = std::make_shared<dht::http::Request>(serverProxy->io_context(), "https://google.ca",
+                   [&](const dht::http::Response& response){
+        logger->w("got answer: %.*s", response.body.size(), response.body.data());
+        status = response.status_code;
+        done = true;
+        cv.notify_all();
+    }, logger);
    request->send();
-
    CPPUNIT_ASSERT(cv.wait_for(lk, std::chrono::seconds(10), [&]{ return done; }));
-    CPPUNIT_ASSERT(status != 0);
+    //CPPUNIT_ASSERT(status == 200);
+
    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 = std::make_shared<dht::http::Request>(serverProxy->io_context(), "https://google.ca/sdbjklwGBIP",
+                   [&](const dht::http::Response& response){
+        logger->w("got answer: %.*s", response.body.size(), response.body.data());
+        status = response.status_code;
+        done = true;
+        cv.notify_all();
+    }, logger);
    request->send();
-
    CPPUNIT_ASSERT(cv.wait_for(lk, std::chrono::seconds(10), [&]{ return done; }));
-    CPPUNIT_ASSERT_EQUAL(404u, status);
-#endif
 }

 }  // namespace test
@@ -88,6 +88,7 @@ class HttpTester : public CppUnit::TestFixture {
   void test_send_json();

 private:
+    std::shared_ptr<dht::Logger> logger {};
    std::shared_ptr<dht::DhtRunner> nodePeer;
    std::unique_ptr<dht::DhtProxyServer> serverProxy;
 };
@@ -20,12 +20,9 @@
 #include "peerdiscoverytester.h"
 #include "opendht/value.h"

-#include <mutex>
-#include <condition_variable>
-
 namespace test {

-constexpr unsigned MULTICAST_PORT = 2222;
+constexpr int MULTICAST_PORT = 2222;
 const std::string DHT_NODE_NAME {"dht"};
 const std::string JAMI_NODE_NAME {"jami"};

@@ -16,9 +16,6 @@
 *  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>
@@ -18,9 +18,6 @@
 *  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" {
@@ -16,9 +16,6 @@
 *  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>
@@ -25,13 +25,6 @@
 #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>
Author	SHA1	Message	Date
Adrien Béraud	c63030b591	fix	2019-12-05 17:41:46 -05:00
Adrien Béraud	b3ae25602f	WIP fix	2019-12-05 17:31:46 -05:00
Adrien Béraud	16f64684ed	http tester	2019-12-05 13:18:38 -05:00
Adrien Béraud	82938289e2	proxy client: avoid destroying twice	2019-12-05 13:18:36 -05:00
Adrien Béraud	a671b57273	http/request: add constructor	2019-12-05 13:18:33 -05:00
				`@@ -0,0 +1 @@`
				`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`