update version to 3.2.0

http: simplify Connection::async_connect()
dht_proxy_client: add listenKeepIdle setting

.github/workflows/ccpp.yml

@@ -12,13 +12,36 @@ jobs:
       run: |
         sudo apt install libncurses5-dev libreadline-dev nettle-dev libasio-dev \
         libgnutls28-dev libuv1-dev python3-pip python3-dev python3-setuptools libcppunit-dev libjsoncpp-dev \
-        autotools-dev autoconf libfmt-dev libhttp-parser-dev libmsgpack-dev libargon2-0-dev
+        autotools-dev autoconf libfmt-dev libmsgpack-dev libargon2-0-dev
     - name: Cython
       run: sudo pip3 install Cython
+    - name: llhttp
+      run: |
+        mkdir llhttp
+        wget https://github.com/nodejs/llhttp/archive/refs/tags/release/v9.2.0.tar.gz -O llhttp.tar.gz
+        tar -xzf llhttp.tar.gz -C llhttp --strip-components=1
+        cd llhttp && cmake -DCMAKE_INSTALL_PREFIX=/usr . && make -j2 && sudo make install
+        cd ../ && rm -rf llhttp*
+    - name: expected-lite
+      run: |
+        sudo mkdir /usr/local/include/nonstd && \
+        sudo wget https://raw.githubusercontent.com/martinmoene/expected-lite/master/include/nonstd/expected.hpp \
+          -O /usr/local/include/nonstd/expected.hpp
+    - name: restinio
+      run: |
+        mkdir restinio && cd restinio
+        wget https://github.com/Stiffstream/restinio/releases/download/v.0.7.2/restinio-0.7.2.tar.bz2
+        ls -l && tar -xjf restinio-0.7.2.tar.bz2
+        cd restinio-0.7.2/dev
+        cmake -DCMAKE_INSTALL_PREFIX=/usr/local -DRESTINIO_TEST=Off -DRESTINIO_SAMPLE=Off -DRESTINIO_BENCHMARK=Off \
+                -DRESTINIO_WITH_SOBJECTIZER=Off -DRESTINIO_DEP_STANDALONE_ASIO=system -DRESTINIO_DEP_LLHTTP=system \
+                -DRESTINIO_DEP_FMT=system -DRESTINIO_DEP_EXPECTED_LITE=system .
+        sudo make install
+        cd ../../ && rm -rf restinio*
     - name: Configure
       run: |
         ./autogen.sh
-        ./configure
+        ./configure --enable-proxy-client
     - name: Build
       run: make
 
@@ -31,7 +54,7 @@ jobs:
       run: |
         sudo apt install ninja-build libncurses5-dev libreadline-dev nettle-dev libasio-dev \
         libgnutls28-dev libuv1-dev python3-dev python3-setuptools python3-pip \
-        libcppunit-dev libjsoncpp-dev libfmt-dev libhttp-parser-dev libmsgpack-dev libargon2-0-dev
+        libcppunit-dev libjsoncpp-dev libfmt-dev libmsgpack-dev libargon2-0-dev
     - name: Python dependencies
       run: sudo pip3 install meson Cython
     - name: Configure
@@ -66,19 +89,25 @@ jobs:
     - uses: actions/checkout@v3
     - name: Dependencies
       run: |
-        brew install msgpack-cxx asio gnutls nettle readline fmt jsoncpp argon2 openssl http-parser cppunit
+        brew install msgpack-cxx asio gnutls nettle readline fmt jsoncpp argon2 openssl cppunit
+
+    - name: expected-lite
+      run: |
+        sudo mkdir /usr/local/include/nonstd && \
+        sudo wget https://raw.githubusercontent.com/martinmoene/expected-lite/master/include/nonstd/expected.hpp \
+          -O /usr/local/include/nonstd/expected.hpp
 
     - name: restinio
       run: |
         mkdir restinio && cd restinio
-        wget https://github.com/aberaud/restinio/archive/6fd08b65f6f15899dd0de3c801f6a5462b811c64.tar.gz
-        ls -l && tar -xzf 6fd08b65f6f15899dd0de3c801f6a5462b811c64.tar.gz
-        cd restinio-6fd08b65f6f15899dd0de3c801f6a5462b811c64/dev
-        cmake -DCMAKE_INSTALL_PREFIX=/usr/local -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 .
-        make -j8 && sudo make install
-        cd ../../.. && rm -rf restinio
+        wget https://github.com/Stiffstream/restinio/releases/download/v.0.7.2/restinio-0.7.2.tar.bz2
+        ls -l && tar -xjf restinio-0.7.2.tar.bz2
+        cd restinio-0.7.2/dev
+        cmake -DCMAKE_INSTALL_PREFIX=/usr/local -DRESTINIO_TEST=Off -DRESTINIO_SAMPLE=Off -DRESTINIO_BENCHMARK=Off \
+                -DRESTINIO_WITH_SOBJECTIZER=Off -DRESTINIO_DEP_STANDALONE_ASIO=system -DRESTINIO_DEP_LLHTTP=system \
+                -DRESTINIO_DEP_FMT=system -DRESTINIO_DEP_EXPECTED_LITE=system .
+        make -j2 && sudo make install
+        cd ../../ && rm -rf restinio*
 
     - name: Configure
       run: |

.github/workflows/clang-analyzer.yml

@@ -11,21 +11,27 @@ jobs:
         run: |
           sudo apt install libncurses5-dev libreadline-dev nettle-dev \
           libgnutls28-dev libuv1-dev python3-dev python3-setuptools libcppunit-dev libjsoncpp-dev \
-          autotools-dev autoconf libfmt-dev libhttp-parser-dev libmsgpack-dev libargon2-0-dev libasio-dev \
+          autotools-dev autoconf libfmt-dev libmsgpack-dev libargon2-0-dev libasio-dev \
           llvm llvm-dev clang clang-tools && \
           sudo apt remove gcc g++
 
+      - name: expected-lite
+        run: |
+          sudo mkdir /usr/include/nonstd && \
+          sudo wget https://raw.githubusercontent.com/martinmoene/expected-lite/master/include/nonstd/expected.hpp \
+            -O /usr/include/nonstd/expected.hpp
+
       - name: restinio
         run: |
-          mkdir restinio && cd restinio \
-          && wget https://github.com/aberaud/restinio/archive/6fd08b65f6f15899dd0de3c801f6a5462b811c64.tar.gz \
-          && ls -l && tar -xzf 6fd08b65f6f15899dd0de3c801f6a5462b811c64.tar.gz \
-          && cd restinio-6fd08b65f6f15899dd0de3c801f6a5462b811c64/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 . \
-          && make -j8 && sudo make install \
-          && cd ../../.. && rm -rf restinio
+          mkdir restinio && cd restinio
+          wget https://github.com/Stiffstream/restinio/releases/download/v.0.7.2/restinio-0.7.2.tar.bz2
+          ls -l && tar -xjf restinio-0.7.2.tar.bz2
+          cd restinio-0.7.2/dev
+          cmake -DCMAKE_INSTALL_PREFIX=/usr -DRESTINIO_TEST=Off -DRESTINIO_SAMPLE=Off -DRESTINIO_BENCHMARK=Off \
+                  -DRESTINIO_WITH_SOBJECTIZER=Off -DRESTINIO_DEP_STANDALONE_ASIO=system -DRESTINIO_DEP_LLHTTP=system \
+                  -DRESTINIO_DEP_FMT=system -DRESTINIO_DEP_EXPECTED_LITE=system .
+          make -j2 && sudo make install
+          cd ../../ && rm -rf restinio*
 
       - name: cmake
         run: |

CMakeLists.txt

@@ -1,4 +1,7 @@
 cmake_minimum_required (VERSION 3.16..3.28)
+if(POLICY CMP0135)
+  cmake_policy(SET CMP0135 NEW)
+endif()
 project (opendht)
 
 include(CMakePackageConfigHelpers)
@@ -7,9 +10,10 @@ include(CheckIncludeFileCXX)
 include(FindPkgConfig)
 include(cmake/CheckAtomic.cmake)
 include(CTest)
+include(FetchContent)
 
 set (opendht_VERSION_MAJOR 3)
-set (opendht_VERSION_MINOR 1.9)
+set (opendht_VERSION_MINOR 2.0)
 set (opendht_VERSION ${opendht_VERSION_MAJOR}.${opendht_VERSION_MINOR})
 set (PACKAGE_VERSION ${opendht_VERSION})
 set (VERSION "${opendht_VERSION}")
@@ -87,9 +91,35 @@ if (NOT MSVC)
 
     if (OPENDHT_HTTP)
         find_package(Restinio REQUIRED)
-        find_library(HTTP_PARSER_LIBRARY http_parser)
-        add_library(http_parser SHARED IMPORTED)
-        set(http_parser_lib "-lhttp_parser")
+
+        # llhttp
+        find_path(LLHTTP_INCLUDE_DIR llhttp.h)
+        find_library(LLHTTP_LIBRARY libllhttp.a)
+        if (LLHTTP_INCLUDE_DIR AND LLHTTP_LIBRARY)
+            message(STATUS "Found llhttp ${LLHTTP_INCLUDE_DIR} ${LLHTTP_LIBRARY}")
+            add_library(llhttp_static STATIC IMPORTED)
+            set_target_properties(llhttp_static PROPERTIES
+                IMPORTED_LOCATION ${LLHTTP_LIBRARY}
+                INTERFACE_INCLUDE_DIRECTORIES ${LLHTTP_INCLUDE_DIR}
+            )
+            set(llhttp_target llhttp_static)
+        else()
+            FetchContent_Declare(llhttp-local URL "https://github.com/nodejs/llhttp/archive/refs/tags/release/v9.2.0.tar.gz")
+            if (BUILD_SHARED_LIBS)
+                set(BUILD_SHARED_LIBS ON CACHE INTERNAL "")
+            else()
+                set(BUILD_SHARED_LIBS OFF CACHE INTERNAL "")
+                set(BUILD_STATIC_LIBS ON CACHE INTERNAL "")
+            endif()
+            FetchContent_MakeAvailable(llhttp-local)
+            if (BUILD_SHARED_LIBS)
+                set(llhttp_target llhttp_shared)
+            else()
+                set(llhttp_target llhttp_static)
+            endif()
+        endif()
+        set(http_lib "-lllhttp")
+        
         if (NOT Jsoncpp_FOUND)
             message(SEND_ERROR "Jsoncpp is required for DHT proxy support")
         endif()
@@ -204,6 +234,7 @@ list (APPEND opendht_SOURCES
 )
 
 list (APPEND opendht_HEADERS
+    include/opendht.h
     include/opendht/def.h
     include/opendht/utils.h
     include/opendht/sockaddr.h
@@ -356,12 +387,14 @@ else()
         PRIVATE
             PkgConfig::argon2
             PkgConfig::Nettle
-            ${HTTP_PARSER_LIBRARY}
         PUBLIC
             ${CMAKE_THREAD_LIBS_INIT}
             PkgConfig::GnuTLS
             fmt::fmt
     )
+    if (OPENDHT_HTTP)
+        target_link_libraries(opendht PUBLIC ${llhttp_target})
+    endif()
     if (Jsoncpp_FOUND)
         target_link_libraries(opendht PUBLIC PkgConfig::Jsoncpp)
     endif()

configure.ac

@@ -1,7 +1,7 @@
 dnl define macros
 m4_define([opendht_major_version], 3)
-m4_define([opendht_minor_version], 1)
-m4_define([opendht_patch_version], 9)
+m4_define([opendht_minor_version], 2)
+m4_define([opendht_patch_version], 0)
 m4_define([opendht_version],
 		  [opendht_major_version.opendht_minor_version.opendht_patch_version])
 
@@ -21,10 +21,10 @@ AS_IF([test "x$enable_debug" = "xyes"],
       [CXXFLAGS="${CXXFLAGS} -g -Og -Wno-return-type -Wall -Wextra -Wnon-virtual-dtor -Wno-deprecated -pedantic-errors"],
       [CXXFLAGS="${CXXFLAGS} -O3 -Wno-deprecated -pedantic-errors -fvisibility=hidden"])
 
-CPPFLAGS+=" -DOPENDHT_BUILD"
+CPPFLAGS="${CPPFLAGS} -DOPENDHT_BUILD"
 AM_CONDITIONAL([OPENDHT_SHARED], [test "x$enable_shared" != xno])
 AM_COND_IF(OPENDHT_SHARED, [
-  CPPFLAGS+=" -Dopendht_EXPORTS"
+  CPPFLAGS="${CPPFLAGS} -Dopendht_EXPORTS"
 ])
 
 AM_PROG_AR
@@ -147,7 +147,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="${CPPFLAGS} -DOPENDHT_JSONCPP"
     AC_SUBST(jsoncpp_lib, [", jsoncpp"])
 ], [
     AC_MSG_NOTICE([Not using JsonCpp])
@@ -175,11 +175,11 @@ AS_IF([test "x$have_fmt" = "xyes"], [
 AM_COND_IF([PROXY_CLIENT_OR_SERVER], [
     AC_CHECK_HEADERS([asio.hpp],, AC_MSG_ERROR([Missing Asio headers files]))
     CXXFLAGS="${CXXFLAGS} -DASIO_STANDALONE"
-    # http_parser has no pkgconfig, instead we check with:
-    AC_CHECK_LIB(http_parser, exit,, AC_MSG_ERROR([Missing HttpParser library files]))
-    AC_CHECK_HEADERS([http_parser.h], [http_parser_headers=yes; break;])
-    AC_SUBST(http_parser_lib, ["-lhttp_parser"])
-    AS_IF([test "x$http_parser_headers" != "xyes"], AC_MSG_ERROR([Missing HttpParser headers files]))
+    # llhttp has no pkgconfig, instead we check with:
+    AC_CHECK_LIB(llhttp, exit,, AC_MSG_ERROR([Missing llhttp library files]))
+    AC_CHECK_HEADERS([llhttp.h], [llhttp_headers=yes; break;])
+    AC_SUBST(http_lib, ["-lllhttp"])
+    AS_IF([test "x$llhttp_headers" != "xyes"], AC_MSG_ERROR([Missing llhttp headers files]))
 ])
 
 CXXFLAGS="${CXXFLAGS} -DMSGPACK_NO_BOOST -DMSGPACK_DISABLE_LEGACY_NIL -DMSGPACK_DISABLE_LEGACY_CONVERT"
@@ -194,8 +194,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_PUSH_NOTIFICATIONS, [CPPFLAGS+=" -DOPENDHT_PUSH_NOTIFICATIONS"], [])
-AM_COND_IF(ENABLE_PROXY_SERVER_IDENTITY, [CPPFLAGS+=" -DOPENDHT_PROXY_SERVER_IDENTITY"], [])
+AM_COND_IF(ENABLE_PUSH_NOTIFICATIONS, [CPPFLAGS="${CPPFLAGS} -DOPENDHT_PUSH_NOTIFICATIONS"], [])
+AM_COND_IF(ENABLE_PROXY_SERVER_IDENTITY, [CPPFLAGS="${CPPFLAGS} -DOPENDHT_PROXY_SERVER_IDENTITY"], [])
 
 dnl Configure setup.py if we build the python module
 AC_SUBST(CURRENT_SOURCE_DIR, ".")

docker/DockerfileAlpine

@@ -19,7 +19,7 @@ RUN cd opendht && mkdir build && cd build \
 				-DOPENDHT_SYSTEMD=On \
 	&& make -j8 && make install
 
-FROM alpine:3.18 AS install
+FROM alpine:3.19 AS install
 COPY --from=build /install /
 RUN apk add --no-cache \
         libstdc++ \
@@ -28,8 +28,8 @@ RUN apk add --no-cache \
         openssl \
         argon2-dev \
         jsoncpp \
+        llhttp \
         fmt \
-        http-parser \
         readline \
         ncurses
 CMD ["dhtnode", "-b", "bootstrap.jami.net", "-p", "4222", "--proxyserver", "8080"]

docker/DockerfileDeps

@@ -13,18 +13,22 @@ RUN apt-get update && apt-get install -y \
         python3-pip python3-dev python3-setuptools python3-build python3-virtualenv \
         libncurses5-dev libreadline-dev nettle-dev libcppunit-dev \
         libgnutls28-dev libuv1-dev libjsoncpp-dev libargon2-dev \
-        libssl-dev libfmt-dev libhttp-parser-dev libasio-dev libmsgpack-dev \
+        libssl-dev libfmt-dev libasio-dev libmsgpack-dev \
     && apt-get clean && rm -rf /var/lib/apt/lists/* /var/cache/apt/*
 
 RUN pip3 install meson Cython
 
+RUN mkdir /usr/include/nonstd \
+    && wget https://raw.githubusercontent.com/martinmoene/expected-lite/master/include/nonstd/expected.hpp \
+            -O /usr/include/nonstd/expected.hpp
+
 RUN echo "*** Downloading RESTinio ***" \
     && mkdir restinio && cd restinio \
-    && wget https://github.com/aberaud/restinio/archive/6fd08b65f6f15899dd0de3c801f6a5462b811c64.tar.gz \
-    && ls -l && tar -xzf 6fd08b65f6f15899dd0de3c801f6a5462b811c64.tar.gz \
-    && cd restinio-6fd08b65f6f15899dd0de3c801f6a5462b811c64/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 . \
-    && make -j8 && make install \
-    && cd ../../.. && rm -rf restinio
+    && wget https://github.com/Stiffstream/restinio/releases/download/v.0.7.2/restinio-0.7.2.tar.bz2 \
+    && ls -l && tar -xjf restinio-0.7.2.tar.bz2 \
+    && cd restinio-0.7.2/dev \
+    && cmake -DCMAKE_INSTALL_PREFIX=/usr -DRESTINIO_TEST=Off -DRESTINIO_SAMPLE=Off -DRESTINIO_BENCHMARK=Off \
+             -DRESTINIO_WITH_SOBJECTIZER=Off -DRESTINIO_DEP_STANDALONE_ASIO=system -DRESTINIO_DEP_LLHTTP=system \
+             -DRESTINIO_DEP_FMT=system -DRESTINIO_DEP_EXPECTED_LITE=system . \
+    && make -j2 && make install \
+    && cd ../../ && rm -rf restinio*

docker/DockerfileDepsAlpine

@@ -1,4 +1,4 @@
-FROM alpine:3.18
+FROM alpine:3.19
 LABEL maintainer="Adrien Béraud <adrien.beraud@savoirfairelinux.com>"
 LABEL org.opencontainers.image.source https://github.com/savoirfairelinux/opendht
 
@@ -8,16 +8,20 @@ RUN apk add --no-cache \
         ncurses-dev readline-dev nettle-dev \
         cppunit-dev gnutls-dev jsoncpp-dev \
         argon2-dev openssl-dev fmt-dev \
-        http-parser-dev asio-dev msgpack-cxx-dev \
+        llhttp-dev asio-dev msgpack-cxx-dev \
     && rm -rf /var/cache/apk/*
 
+RUN mkdir /usr/include/nonstd \
+    && wget https://raw.githubusercontent.com/martinmoene/expected-lite/master/include/nonstd/expected.hpp \
+            -O /usr/include/nonstd/expected.hpp
+
 RUN echo "*** Downloading RESTinio ***" \
     && mkdir restinio && cd restinio \
-    && wget https://github.com/aberaud/restinio/archive/6fd08b65f6f15899dd0de3c801f6a5462b811c64.tar.gz \
-    && tar -xzf 6fd08b65f6f15899dd0de3c801f6a5462b811c64.tar.gz \
-    && cd restinio-6fd08b65f6f15899dd0de3c801f6a5462b811c64/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 . \
-    && make -j8 && make install \
-    && cd ../../.. && rm -rf restinio
+    && wget https://github.com/Stiffstream/restinio/releases/download/v.0.7.2/restinio-0.7.2.tar.bz2 \
+    && ls -l && tar -xjf restinio-0.7.2.tar.bz2 \
+    && cd restinio-0.7.2/dev \
+    && cmake -DCMAKE_INSTALL_PREFIX=/usr -DRESTINIO_TEST=Off -DRESTINIO_SAMPLE=Off -DRESTINIO_BENCHMARK=Off \
+             -DRESTINIO_WITH_SOBJECTIZER=Off -DRESTINIO_DEP_STANDALONE_ASIO=system -DRESTINIO_DEP_LLHTTP=system \
+             -DRESTINIO_DEP_FMT=system -DRESTINIO_DEP_EXPECTED_LITE=system . \
+    && make -j2 && make install \
+    && cd ../../ && rm -rf restinio*

docker/DockerfileDepsBionic

@@ -7,7 +7,7 @@ RUN apt-get update && apt-get install -y \
         apt-transport-https build-essential pkg-config git wget libncurses5-dev libreadline-dev nettle-dev \
         libgnutls28-dev libuv1-dev python3-dev python3-setuptools libcppunit-dev libjsoncpp-dev \
         libargon2-0-dev \
-        autotools-dev autoconf libfmt-dev libhttp-parser-dev libmsgpack-dev libssl-dev python3-pip \
+        autotools-dev autoconf libfmt-dev libmsgpack-dev libssl-dev python3-pip \
     && apt-get clean && rm -rf /var/lib/apt/lists/* /var/cache/apt/*
 
 RUN pip3 install --upgrade cmake meson Cython
@@ -20,13 +20,17 @@ RUN echo "** Building a recent version of asio ***" \
     && make install \
     && cd ../../ && rm -rf asio*
 
+RUN mkdir /usr/include/nonstd \
+    && wget https://raw.githubusercontent.com/martinmoene/expected-lite/master/include/nonstd/expected.hpp \
+            -O /usr/include/nonstd/expected.hpp
+
 RUN echo "*** Downloading RESTinio ***" \
     && mkdir restinio && cd restinio \
-    && wget https://github.com/aberaud/restinio/archive/8d5d3e8237e0947adb9ba1ffc8281f4ad7cb2a59.tar.gz \
-    && ls -l && tar -xzf 8d5d3e8237e0947adb9ba1ffc8281f4ad7cb2a59.tar.gz \
-    && cd restinio-8d5d3e8237e0947adb9ba1ffc8281f4ad7cb2a59/dev \
-    && 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 . \
-    && make -j8 && make install \
+    && wget https://github.com/Stiffstream/restinio/releases/download/v.0.7.2/restinio-0.7.2.tar.bz2 \
+    && ls -l && tar -xjf restinio-0.7.2.tar.bz2 \
+    && cd restinio-0.7.2/dev \
+    && cmake -DCMAKE_INSTALL_PREFIX=/usr -DRESTINIO_TEST=Off -DRESTINIO_SAMPLE=Off -DRESTINIO_BENCHMARK=Off \
+             -DRESTINIO_WITH_SOBJECTIZER=Off -DRESTINIO_DEP_STANDALONE_ASIO=system -DRESTINIO_DEP_LLHTTP=system \
+             -DRESTINIO_DEP_FMT=system -DRESTINIO_DEP_EXPECTED_LITE=system . \
+    && make -j2 && make install \
     && cd ../../ && rm -rf restinio*

docker/DockerfileDepsFocal

@@ -13,18 +13,22 @@ RUN apt-get update && apt-get install -y \
         python3-dev python3-setuptools python3-pip \
         libncurses5-dev libreadline-dev nettle-dev libcppunit-dev \
         libgnutls28-dev libuv1-dev libjsoncpp-dev libargon2-dev \
-        libssl-dev libfmt-dev libhttp-parser-dev libasio-dev libmsgpack-dev \
+        libssl-dev libfmt-dev libasio-dev libmsgpack-dev \
     && apt-get clean && rm -rf /var/lib/apt/lists/* /var/cache/apt/*
 
 RUN pip3 install meson Cython
 
+RUN mkdir /usr/include/nonstd \
+    && wget https://raw.githubusercontent.com/martinmoene/expected-lite/master/include/nonstd/expected.hpp \
+            -O /usr/include/nonstd/expected.hpp
+
 RUN echo "*** Downloading RESTinio ***" \
     && mkdir restinio && cd restinio \
-    && wget https://github.com/aberaud/restinio/archive/e0a261dd8488246a3cb8bbb3ea781ea5139c3c94.tar.gz \
-    && ls -l && tar -xzf e0a261dd8488246a3cb8bbb3ea781ea5139c3c94.tar.gz \
-    && cd restinio-e0a261dd8488246a3cb8bbb3ea781ea5139c3c94/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 . \
-    && make -j8 && make install \
-    && cd ../../.. && rm -rf restinio
+    && wget https://github.com/Stiffstream/restinio/releases/download/v.0.7.2/restinio-0.7.2.tar.bz2 \
+    && ls -l && tar -xjf restinio-0.7.2.tar.bz2 \
+    && cd restinio-0.7.2/dev \
+    && cmake -DCMAKE_INSTALL_PREFIX=/usr -DRESTINIO_TEST=Off -DRESTINIO_SAMPLE=Off -DRESTINIO_BENCHMARK=Off \
+             -DRESTINIO_WITH_SOBJECTIZER=Off -DRESTINIO_DEP_STANDALONE_ASIO=system -DRESTINIO_DEP_LLHTTP=system \
+             -DRESTINIO_DEP_FMT=system -DRESTINIO_DEP_EXPECTED_LITE=system . \
+    && make -j2 && make install \
+    && cd ../../ && rm -rf restinio*

docker/DockerfileDepsLlvm

@@ -13,7 +13,7 @@ RUN apt-get update \
        nettle-dev libgnutls28-dev libuv1-dev libmsgpack-dev libjsoncpp-dev python3-dev \
        python3-setuptools libcppunit-dev python3-pip python3-build python3-virtualenv \
        autotools-dev autoconf libssl-dev libargon2-dev \
-       libfmt-dev libhttp-parser-dev libasio-dev \
+       libfmt-dev libasio-dev \
     && apt-get remove -y gcc g++ && apt-get autoremove -y \
     && apt-get clean && rm -rf /var/lib/apt/lists/* /var/cache/apt/*
 
@@ -22,13 +22,17 @@ RUN pip3 install meson Cython
 ENV CC cc
 ENV CXX c++
 
+RUN mkdir /usr/include/nonstd \
+    && wget https://raw.githubusercontent.com/martinmoene/expected-lite/master/include/nonstd/expected.hpp \
+            -O /usr/include/nonstd/expected.hpp
+
 RUN echo "*** Downloading RESTinio ***" \
     && mkdir restinio && cd restinio \
-    && wget https://github.com/aberaud/restinio/archive/6fd08b65f6f15899dd0de3c801f6a5462b811c64.tar.gz \
-    && ls -l && tar -xzf 6fd08b65f6f15899dd0de3c801f6a5462b811c64.tar.gz \
-    && cd restinio-6fd08b65f6f15899dd0de3c801f6a5462b811c64/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 . \
-    && make -j8 && make install \
+    && wget https://github.com/Stiffstream/restinio/releases/download/v.0.7.2/restinio-0.7.2.tar.bz2 \
+    && ls -l && tar -xjf restinio-0.7.2.tar.bz2 \
+    && cd restinio-0.7.2/dev \
+    && cmake -DCMAKE_INSTALL_PREFIX=/usr -DRESTINIO_TEST=Off -DRESTINIO_SAMPLE=Off -DRESTINIO_BENCHMARK=Off \
+             -DRESTINIO_WITH_SOBJECTIZER=Off -DRESTINIO_DEP_STANDALONE_ASIO=system -DRESTINIO_DEP_LLHTTP=system \
+             -DRESTINIO_DEP_FMT=system -DRESTINIO_DEP_EXPECTED_LITE=system . \
+    && make -j2 && make install \
     && cd ../../ && rm -rf restinio*

include/opendht/dht_proxy_client.h

@@ -291,6 +291,11 @@ public:
         loopSignal_();
     }
 
+    void listenKeepIdle(uint32_t seconds) {
+        listenKeepIdle_ = seconds;
+    }
+    inline uint32_t listenKeepIdle() { return listenKeepIdle_; }
+
 private:
     /**
      * Start the connection with a server.
@@ -391,6 +396,11 @@ private:
     size_t listenerToken_ {0};
     std::map<InfoHash, ProxySearch> searches_;
 
+    /**
+     * Keepalive idle time for listen
+     */
+    uint32_t listenKeepIdle_ {120};
+
     /**
      * Callbacks should be executed in the main thread.
      */

include/opendht/dht_proxy_server.h

@@ -46,17 +46,17 @@ enum class PushType {
 }
 MSGPACK_ADD_ENUM(dht::PushType)
 
-namespace http {
-class Request;
-struct ListenerSession;
-}
-
 namespace Json {
 class Value;
 }
 
 namespace dht {
 
+namespace http {
+class Request;
+struct ListenerSession;
+}
+
 class DhtRunner;
 
 using RestRouter = restinio::router::express_router_t<>;
@@ -275,6 +275,12 @@ private:
     RequestStatus options(restinio::request_handle_t request,
                            restinio::router::route_params_t params);
 
+    struct PushSessionContext {
+        std::mutex lock;
+        std::string sessionId;
+        PushSessionContext(const std::string& id) : sessionId(id) {}
+    };
+
 #ifdef OPENDHT_PUSH_NOTIFICATIONS
     PushType getTypeFromString(const std::string& type);
     std::string getDefaultTopic(PushType type);
@@ -331,6 +337,24 @@ private:
     void handleCancelPushListen(const asio::error_code &ec, const std::string pushToken,
                                 const InfoHash key, const std::string clientId);
 
+    /**
+     * Handles a push listen request.
+     *
+     * @param infoHash The information hash associated with the push listen request.
+     * @param pushToken The push token associated with the push listen request.
+     * @param type The type of the push listen request.
+     * @param clientId The client ID associated with the push listen request.
+     * @param sessionCtx The shared pointer to the push session context associated with the push listen request.
+     * @param topic The topic associated with the push listen request.
+     * @param values The vector of shared pointers to values associated with the push listen request.
+     * @param expired A boolean indicating whether the push listen request has expired.
+     * @return true.
+     */
+    bool handlePushListen(const InfoHash& infoHash, const std::string& pushToken,
+                          PushType type, const std::string& clientId,
+                          const std::shared_ptr<DhtProxyServer::PushSessionContext>& sessionCtx, const std::string& topic,
+                          const std::vector<std::shared_ptr<Value>>& values, bool expired);
+
 #endif //OPENDHT_PUSH_NOTIFICATIONS
 
     void handlePrintStats(const asio::error_code &ec);
@@ -376,12 +400,6 @@ private:
     std::map<restinio::connection_id_t, http::ListenerSession> listeners_;
     // Connection Listener observing conn state changes.
     std::shared_ptr<ConnectionListener> connListener_;
-
-    struct PushSessionContext {
-        std::mutex lock;
-        std::string sessionId;
-        PushSessionContext(const std::string& id) : sessionId(id) {}
-    };
     struct PermanentPut {
         time_point expiration;
         std::string pushToken;

include/opendht/http.h

@@ -29,23 +29,20 @@
 #   define snprintf snprintf
 #endif
 
+#include <asio/ip/tcp.hpp>
+#include <asio/streambuf.hpp>
 #include <asio/ssl/context.hpp>
-#include <restinio/http_headers.hpp>
 #include <restinio/message_builders.hpp>
 
 #include <memory>
 #include <queue>
 #include <mutex>
+#include <future>
 
 namespace Json {
 class Value;
 }
 
-extern "C" {
-struct http_parser;
-struct http_parser_settings;
-}
-
 namespace restinio {
 namespace impl {
 class tls_socket_t;
@@ -116,6 +113,8 @@ public:
     void async_read(size_t bytes, BytesHandlerCb cb);
     void async_read_some(size_t bytes, BytesHandlerCb cb);
 
+    void set_keepalive(uint32_t seconds);
+
     const asio::ip::address& local_address() const;
 
     void timeout(const std::chrono::seconds& timeout, HandlerCb cb = {});
@@ -154,17 +153,6 @@ private:
     bool checkOcsp_ {false};
 };
 
-/**
- * Session value associated with a connection_id_t key.
- */
-struct ListenerSession
-{
-    ListenerSession() = default;
-    dht::InfoHash hash;
-    std::future<size_t> token;
-    std::shared_ptr<restinio::response_builder_t<restinio::chunked_output_t>> response;
-};
-
 /* @class Resolver
  * @brief The purpose is to only resolve once to avoid mutliple dns requests per operation.
  */
@@ -336,7 +324,7 @@ private:
 
     void init_default_headers();
     /**
-     * Initialized and wraps the http_parser callbacks with our user defined callbacks.
+     * Initialized and wraps the llhttp callbacks with our user defined callbacks.
      */
     void init_parser();
 
@@ -378,8 +366,8 @@ private:
     Response response_ {};
     std::string request_;
     std::atomic<bool> finishing_ {false};
-    std::unique_ptr<http_parser> parser_;
-    std::unique_ptr<http_parser_settings> parser_s_;
+    std::unique_ptr<llhttp_t> parser_;
+    std::unique_ptr<llhttp_settings_t> parser_s_;
 
     // Next request in case of redirect following
     std::shared_ptr<Request> next_;

include/opendht/thread_pool.h

@@ -63,6 +63,7 @@ public:
 
     void stop(bool wait = true);
     void join();
+    void detach();
 
 private:
     std::mutex lock_ {};

meson.build

@@ -1,5 +1,5 @@
 project('opendht', 'c', 'cpp', 'cython',
-    version: '3.1.9',
+    version: '3.2.0',
     default_options: [
         'cpp_std=c++17',
         'warning_level=3'
@@ -12,15 +12,9 @@ argon2 = dependency('libargon2')
 openssl = dependency('openssl', required: get_option('proxy_client'))
 jsoncpp = dependency('jsoncpp', required: get_option('proxy_client'))
 fmt = dependency('fmt')
+llhttp = dependency('llhttp', 'libllhttp', required: get_option('proxy_client'))
 
-dirs=[]
-if host_machine.system() == 'freebsd'
-    dirs+='/usr/local/lib'
-elif host_machine.system() == 'darwin'
-    dirs+='/opt/homebrew/lib'
-endif
-http_parser = meson.get_compiler('c').find_library('http_parser', dirs: dirs, required: get_option('proxy_client'))
-deps = [fmt, gnutls, nettle, msgpack, argon2, openssl, jsoncpp, http_parser]
+deps = [fmt, gnutls, nettle, msgpack, argon2, openssl, jsoncpp, llhttp]
 
 add_project_arguments('-DMSGPACK_NO_BOOST', language : 'cpp')
 add_project_arguments(['-Wno-return-type','-Wno-deprecated','-Wnon-virtual-dtor','-pedantic-errors','-fvisibility=hidden'], language : 'cpp')
@@ -42,9 +36,12 @@ opendht_src = [
     'src/dhtrunner.cpp',
     'src/log.cpp',
     'src/op_cache.cpp',
-    'src/network_utils.cpp'
+    'src/network_utils.cpp',
+    'src/thread_pool.cpp',
 ]
 
+conf_data = configuration_data()
+
 if get_option('indexation').enabled()
     opendht_src += 'src/indexation/pht.cpp'
     add_project_arguments('-DOPENDHT_INDEXATION', language : 'cpp')
@@ -52,12 +49,17 @@ endif
 if jsoncpp.found()
     opendht_src += ['src/base64.cpp']
     add_project_arguments('-DOPENDHT_JSONCPP', language : 'cpp')
+    conf_data.set('jsoncpp_lib', ', jsoncpp')
 endif
-if http_parser.found()
+if llhttp.found()
     opendht_src += ['src/http.cpp', 'src/compat/os_cert.cpp']
     if host_machine.system() == 'darwin'
         deps+=dependency('appleframeworks', modules : ['CoreFoundation', 'Security'])
     endif
+    conf_data.set('http_lib', '-lllhttp')
+endif
+if openssl.found()
+    conf_data.set('openssl_lib', ', openssl')
 endif
 if get_option('proxy_client').enabled()
     opendht_src += ['src/dht_proxy_client.cpp']
@@ -74,23 +76,42 @@ if get_option('peer_discovery').enabled()
     opendht_src += 'src/peer_discovery.cpp'
     add_project_arguments('-DOPENDHT_PEER_DISCOVERY', language : 'cpp')
 endif
-opendht = shared_library('opendht',
+opendht = library('opendht',
     opendht_src,
     include_directories : opendht_inc,
     dependencies : deps,
     cpp_args : ['-DOPENHT_BUILD', '-Dopendht_EXPORTS'],
     install : true)
 
+conf_data.set('VERSION', meson.project_version())
+conf_data.set('prefix', get_option('prefix'))
+conf_data.set('exec_prefix', get_option('prefix'))
+conf_data.set('libdir', join_paths(get_option('prefix'), get_option('libdir')))
+conf_data.set('includedir', join_paths(get_option('prefix'), get_option('includedir')))
+conf_data.set('argon2_lib', ', libargon2')
+
+configure_file(input : 'opendht.pc.in',
+               output : 'opendht.pc',
+               configuration : conf_data,
+               install: true,
+               install_dir: join_paths(get_option('prefix'), get_option('libdir'), 'pkgconfig'))
+
 readline = meson.get_compiler('c').find_library('readline', required: get_option('tools'))
 
 if get_option('c').enabled()
-    opendht_c = shared_library('opendht-c',
+    opendht_c = library('opendht-c',
         ['c/opendht.cpp',  'c/opendht_c.h'],
         include_directories : opendht_interface_inc,
         link_with : opendht,
-        dependencies : [msgpack, fmt],
+        dependencies : [msgpack, jsoncpp, fmt],
         cpp_args : ['-DOPENDHT_C_BUILD', '-Dopendht_c_EXPORTS'],
         install : true)
+    
+    configure_file(input : 'opendht-c.pc.in',
+                   output : 'opendht-c.pc',
+                   configuration : conf_data,
+                   install: true,
+                   install_dir: join_paths(get_option('prefix'), get_option('libdir'), 'pkgconfig'))
 
     if get_option('tools').enabled()
         dhtcnode = executable('dhtcnode', 'tools/dhtcnode.c',
@@ -110,7 +131,7 @@ if get_option('python').require(get_option('indexation').enabled(), error_messag
         include_directories : opendht_interface_inc,
         override_options : ['cython_language=cpp'],
         link_with : [opendht],
-        dependencies : [msgpack, fmt, dep_py],
+        dependencies : [msgpack, jsoncpp, fmt, dep_py],
         install : true
     )
 endif
@@ -119,22 +140,98 @@ if get_option('tools').enabled()
     dhtnode = executable('dhtnode', 'tools/dhtnode.cpp',
         include_directories : opendht_interface_inc,
         link_with : opendht,
-        dependencies : [readline, msgpack, fmt, openssl],
+        dependencies : [readline, jsoncpp, msgpack, fmt, openssl],
         install : true)
     dhtchat = executable('dhtchat', 'tools/dhtchat.cpp',
         include_directories : opendht_interface_inc,
         link_with : opendht,
-        dependencies : [readline, msgpack, fmt, openssl],
+        dependencies : [readline, jsoncpp, msgpack, fmt, openssl],
         install : true)
     dhtscanner = executable('dhtscanner', 'tools/dhtscanner.cpp',
         include_directories : opendht_interface_inc,
         link_with : opendht,
-        dependencies : [readline, msgpack, fmt, openssl],
+        dependencies : [readline, jsoncpp, msgpack, fmt, openssl],
         install : true)
-    if http_parser.found()
+    if llhttp.found()
         durl = executable('durl', 'tools/durl.cpp',
             include_directories : opendht_interface_inc,
             link_with : opendht,
             dependencies : [msgpack, openssl])
     endif
 endif
+
+# Unit tests
+if get_option('tests').enabled()
+    cppunit = dependency('cppunit')
+
+    test_infohash = executable('test_infohash',
+        'tests/infohashtester.cpp', 'tests/tests_runner.cpp',
+        include_directories : opendht_interface_inc,
+        link_with : opendht,
+        dependencies : [cppunit, jsoncpp, fmt, openssl, msgpack])
+    test('Infohash', test_infohash)
+
+    test_value = executable('test_value',
+        'tests/valuetester.cpp', 'tests/tests_runner.cpp',
+        include_directories : opendht_interface_inc,
+        link_with : opendht,
+        dependencies : [cppunit, jsoncpp, fmt, openssl, msgpack])
+    test('Value', test_value)
+
+    test_crypto = executable('test_crypto',
+        'tests/cryptotester.cpp', 'tests/tests_runner.cpp',
+        include_directories : opendht_interface_inc,
+        link_with : opendht,
+        dependencies : [cppunit, jsoncpp, fmt, openssl, msgpack])
+    test('Crypto', test_crypto)
+
+    test_dhtrunner = executable('test_dhtrunner',
+        'tests/dhtrunnertester.cpp', 'tests/tests_runner.cpp',
+        include_directories : opendht_interface_inc,
+        link_with : opendht,
+        dependencies : [cppunit, jsoncpp, fmt, openssl, msgpack])
+    test('DhtRunner', test_dhtrunner)
+
+    test_threadpool = executable('test_threadpool',
+        'tests/threadpooltester.cpp', 'tests/tests_runner.cpp',
+        include_directories : opendht_interface_inc,
+        link_with : opendht,
+        dependencies : [cppunit, jsoncpp, fmt, openssl, msgpack])
+    test('ThreadPool', test_threadpool)
+
+    if get_option('proxy_client').enabled() or get_option('proxy_server').enabled()
+        test_http = executable('test_http',
+            'tests/httptester.cpp', 'tests/tests_runner.cpp',
+            include_directories : opendht_interface_inc,
+            link_with : opendht,
+            dependencies : [cppunit, jsoncpp, fmt, openssl, msgpack])
+        test('Http', test_http)
+    endif
+
+    if get_option('peer_discovery').enabled()
+        test_peerdiscovery = executable('test_peerdiscovery',
+            'tests/peerdiscoverytester.cpp', 'tests/tests_runner.cpp',
+            include_directories : opendht_interface_inc,
+            link_with : opendht,
+            dependencies : [cppunit, jsoncpp, fmt, openssl, msgpack])
+        test('PeerDiscovery', test_peerdiscovery)
+    endif
+
+    if get_option('proxy_server').enabled()
+        test_proxy = executable('test_dhtproxy',
+            'tests/dhtproxytester.cpp', 'tests/tests_runner.cpp',
+            include_directories : opendht_interface_inc,
+            link_with : opendht,
+            dependencies : [cppunit, jsoncpp, fmt, openssl, msgpack])
+        test('DhtProxy', test_proxy)
+
+        if get_option('long_tests').enabled()
+            test_proxystress = executable('test_dhtproxystress',
+                'tests/dhtproxy_stress_tester.cpp', 'tests/tests_runner.cpp',
+                include_directories : opendht_interface_inc,
+                link_with : opendht,
+                dependencies : [cppunit, jsoncpp, fmt, openssl, msgpack])
+            test('DhtProxy Stress', test_proxystress)
+        endif
+    endif
+endif

meson.options

@@ -6,3 +6,5 @@ option('tools', type : 'feature', value : 'enabled')
 option('c', type : 'feature', value : 'enabled')
 option('indexation', type : 'feature', value : 'enabled')
 option('python', type : 'feature', value : 'enabled')
+option('tests', type : 'feature', value : 'enabled')
+option('long_tests', type : 'feature', value : 'disabled')

opendht.pc.in

@@ -6,7 +6,7 @@ Name: OpenDHT
 Description: C++17 Distributed Hash Table library
 Version: @VERSION@
 Libs: -L${libdir} -lopendht
-Libs.private: @http_parser_lib@ -pthread
+Libs.private: @http_lib@ -pthread
 Requires: gnutls >= 3.3@jsoncpp_lib@@openssl_lib@
 Requires.private: nettle >= 2.4@argon2_lib@
 Cflags: -I${includedir}

src/dht_proxy_client.cpp

@@ -23,7 +23,7 @@
 #include "op_cache.h"
 #include "utils.h"
 
-#include <http_parser.h>
+#include <llhttp.h>
 #include <deque>
 
 namespace dht {
@@ -1010,7 +1010,10 @@ DhtProxyClient::handleExpireListener(const asio::error_code &ec, const InfoHash&
             }
         } else {
             // stop the request
-            listener.request.reset();
+            if (listener.request) {
+                listener.request->cancel();
+                listener.request.reset();
+            }
         }
         search->second.listeners.erase(it);
         if (logger_)
@@ -1105,6 +1108,12 @@ DhtProxyClient::sendListen(const restinio::http_request_header_t& header,
             std::lock_guard<std::mutex> l(requestLock_);
             requests_[reqid] = request;
         }
+        request->add_on_status_callback([request, seconds = this->listenKeepIdle()] (unsigned status_code) {
+            if(status_code == 200) {
+                // increase TCP_KEEPIDLE to save power
+                request->get_connection()->set_keepalive(seconds);
+            }
+        });
         request->send();
     }
     catch (const std::exception &e){

src/dht_proxy_server.cpp

@@ -58,6 +58,19 @@ constexpr const std::chrono::minutes PRINT_STATS_PERIOD {2};
 using ResponseByParts = restinio::chunked_output_t;
 using ResponseByPartsBuilder = restinio::response_builder_t<ResponseByParts>;
 
+namespace http {
+/**
+ * Session value associated with a connection_id_t key.
+ */
+struct ListenerSession
+{
+    ListenerSession() = default;
+    dht::InfoHash hash;
+    std::future<size_t> token;
+    std::shared_ptr<restinio::response_builder_t<restinio::chunked_output_t>> response;
+};
+}
+
 class opendht_logger_t
 {
 public:
@@ -116,7 +129,7 @@ private:
 void
 DhtProxyServer::ConnectionListener::state_changed(const restinio::connection_state::notice_t& notice) noexcept
 {
-    if (restinio::holds_alternative<restinio::connection_state::closed_t>(notice.cause())) {
+    if (std::holds_alternative<restinio::connection_state::closed_t>(notice.cause())) {
         onClosed_(notice.connection_id());
     }
 }
@@ -402,31 +415,12 @@ DhtProxyServer::loadState(Is& is, size_t size) {
                 for (auto& pushListener : pushListeners_) {
                     for (auto& listeners : pushListener.second.listeners) {
                         for (auto& listener : listeners.second) {
+                            // start listening
                             listener.internalToken = dht_->listen(listeners.first,
-                                [this, infoHash=listeners.first, pushToken=pushListener.first, type=listener.type, clientId=listener.clientId, sessionCtx = listener.sessionCtx, topic=listener.topic]
-                                (const std::vector<std::shared_ptr<Value>>& values, bool expired) {
-                                    // Build message content
-                                    Json::Value json;
-                                    json["key"] = infoHash.toString();
-                                    json["to"] = clientId;
-                                    json["t"] = Json::Value::Int64(std::chrono::duration_cast<std::chrono::milliseconds>(system_clock::now().time_since_epoch()).count());
-                                    {
-                                        std::lock_guard<std::mutex> l(sessionCtx->lock);
-                                        json["s"] = sessionCtx->sessionId;
-                                    }
-                                    if (expired and values.size() < 2){
-                                        std::ostringstream ss;
-                                        for(size_t i = 0; i < values.size(); ++i){
-                                            if(i != 0) ss << ",";
-                                            ss << values[i]->id;
-                                        }
-                                        json["exp"] = ss.str();
-                                    }
-                                    auto maxPrio = 1000u;
-                                    for (const auto& v : values)
-                                        maxPrio = std::min(maxPrio, v->priority);
-                                    sendPushNotification(pushToken, std::move(json), type, !expired and maxPrio == 0, topic);
-                                    return true;
+                                [this, infoHash=listeners.first, pushToken=pushListener.first, type=listener.type,
+                                 clientId=listener.clientId, sessionCtx = listener.sessionCtx, topic=listener.topic]
+                                (const std::vector<Sp<Value>>& values, bool expired) {
+                                    return this->handlePushListen(infoHash, pushToken, type, clientId, sessionCtx, topic, values, expired);
                                 }
                             );
                             // expire notify
@@ -897,13 +891,13 @@ DhtProxyServer::subscribe(restinio::request_handle_t request,
         // Send response
         if (not newListener) {
             if (logger_)
-                logger_->d("[proxy:server] [subscribe] found [client %s]", listener.clientId.c_str());
+                logger_->d("[proxy:server] [subscribe %s] found [client %s]", infoHash.toString().c_str(), listener.clientId.c_str());
             // Send response header
             auto response = std::make_shared<ResponseByPartsBuilder>(initHttpResponse(request->create_response<ResponseByParts>()));
             response->flush();
             if (!root["refresh"].asBool()) {
                 // No Refresh
-                dht_->get(infoHash, [this, response](const Sp<Value>& value){
+                dht_->get(infoHash, [this, response](const Sp<Value>& value) {
                     auto output = Json::writeString(jsonBuilder_, value->toJson()) + "\n";
                     response->append_chunk(output);
                     response->flush();
@@ -920,33 +914,15 @@ DhtProxyServer::subscribe(restinio::request_handle_t request,
         } else {
             // =========== No existing listener for an infoHash ============
             // Add listen on dht
+            if (logger_)
+                logger_->d("[proxy:server] [subscribe %s] new", infoHash.toString().c_str());
             listener.internalToken = dht_->listen(infoHash,
                 [this, infoHash, pushToken, type, clientId, sessionCtx = listener.sessionCtx, topic]
-                (const std::vector<std::shared_ptr<Value>>& values, bool expired){
-                    // Build message content
-                    Json::Value json;
-                    json["key"] = infoHash.toString();
-                    json["to"] = clientId;
-                    json["t"] = Json::Value::Int64(std::chrono::duration_cast<std::chrono::milliseconds>(system_clock::now().time_since_epoch()).count());
-                    {
-                        std::lock_guard<std::mutex> l(sessionCtx->lock);
-                        json["s"] = sessionCtx->sessionId;
-                    }
-                    if (expired and values.size() < 2){
-                        std::ostringstream ss;
-                        for(size_t i = 0; i < values.size(); ++i) {
-                            if(i != 0) ss << ",";
-                            ss << values[i]->id;
-                        }
-                        json["exp"] = ss.str();
-                    }
-                    auto maxPrio = 1000u;
-                    for (const auto& v : values)
-                        maxPrio = std::min(maxPrio, v->priority);
-                    sendPushNotification(pushToken, std::move(json), type, !expired and maxPrio == 0, topic);
-                    return true;
+                (const std::vector<Sp<Value>>& values, bool expired) {
+                    return this->handlePushListen(infoHash, pushToken, type, clientId, sessionCtx, topic, values, expired);
                 }
             );
+            // Send response header
             auto response = initHttpResponse(request->create_response());
             response.set_body("{}\n");
             return response.done();
@@ -1050,6 +1026,50 @@ DhtProxyServer::handleCancelPushListen(const asio::error_code &ec, const std::st
         pushListeners_.erase(pushListener);
 }
 
+bool
+DhtProxyServer::handlePushListen(const InfoHash& infoHash, const std::string& pushToken, PushType type, const std::string& clientId,
+                                 const std::shared_ptr<DhtProxyServer::PushSessionContext>& sessionCtx, const std::string& topic,
+                                 const std::vector<std::shared_ptr<Value>>& values, bool expired)
+{
+    Json::Value json;
+    json["key"] = infoHash.toString();
+    json["to"] = clientId;
+    using namespace std::chrono;
+    json["t"] = Json::Value::Int64(duration_cast<milliseconds>(system_clock::now().time_since_epoch()).count());
+
+    {
+        std::lock_guard<std::mutex> l(sessionCtx->lock);
+        json["s"] = sessionCtx->sessionId;
+    }
+
+    // Build a comma-separated list of ids from the values. This is intended to be used when
+    // streaming from the iOS notification extension in order to filter out unwanted values.
+    std::string ids;
+    for(const auto& value : values) {
+        if (!ids.empty()) ids += ",";
+        ids += std::to_string(value->id);
+    }
+    json["ids"] = ids;
+
+    // If message is expired copy the value ID to the `exp` field.
+    // This is presumably used by the Android notification system.
+    if (expired && values.size() < 2) {
+        json["exp"] = json["ids"];
+    }
+
+    auto minPriority = 1000u;
+    for (const auto& v : values)
+        minPriority = std::min(minPriority, v->priority);
+
+    if (logger_)
+        logger_->d("[proxy:server] [listen %s] [client %s] [session %s] [expired %i] [priority %i] [values %zu]",
+                   infoHash.toString().c_str(), clientId.c_str(), sessionCtx->sessionId.c_str(), expired, minPriority, values.size());
+
+    sendPushNotification(pushToken, std::move(json), type, !expired and minPriority == 0, topic);
+
+    return true;
+}
+
 void
 DhtProxyServer::sendPushNotification(const std::string& token, Json::Value&& json, PushType type, bool highPriority, const std::string& topic)
 {

src/http.cpp

@@ -24,7 +24,7 @@
 
 #include <asio.hpp>
 #include <restinio/impl/tls_socket.hpp>
-#include <http_parser.h>
+#include <llhttp.h>
 #include <json/json.h>
 
 #include <openssl/ocsp.h>
@@ -536,48 +536,11 @@ Connection::async_connect(std::vector<asio::ip::tcp::endpoint>&& endpoints, Conn
 #pragma GCC diagnostic ignored "-Wunused-variable"
     ConnectHandlerCb wcb = [this, &base, cb=std::move(cb)](const asio::error_code& ec, const asio::ip::tcp::endpoint& endpoint) {
         if (!ec) {
-            auto socket = base.native_handle();
             local_address_ = base.local_endpoint().address();
             // Once connected, set a keep alive on the TCP socket with 30 seconds delay
             // This will generate broken pipes as soon as possible.
             // Note this needs to be done once connected to have a valid native_handle()
-            uint32_t start = 30;
-            uint32_t interval = 30;
-            uint32_t cnt = 1;
-#ifdef _WIN32
-            std::string val = "1";
-            setsockopt(socket, SOL_SOCKET, SO_KEEPALIVE, val.c_str(), sizeof(val));
-
-            // TCP_KEEPIDLE and TCP_KEEPINTVL are available since Win 10 version 1709
-            // TCP_KEEPCNT since Win 10 version 1703
-#ifdef TCP_KEEPIDLE
-            std::string start_str = std::to_string(start);
-            setsockopt(socket, IPPROTO_TCP, TCP_KEEPIDLE,
-                    start_str.c_str(), sizeof(start_str));
-#endif
-#ifdef TCP_KEEPINTVL
-            std::string interval_str = std::to_string(interval);
-            setsockopt(socket, IPPROTO_TCP, TCP_KEEPINTVL,
-                    interval_str.c_str(), sizeof(interval_str));
-#endif
-#ifdef TCP_KEEPCNT
-            std::string cnt_str = std::to_string(cnt);
-            setsockopt(socket, IPPROTO_TCP, TCP_KEEPCNT,
-                    cnt_str.c_str(), sizeof(cnt_str));
-#endif
-#else
-            uint32_t val = 1;
-            setsockopt(socket, SOL_SOCKET, SO_KEEPALIVE, &val, sizeof(uint32_t));
-#ifdef __APPLE__
-            // Apple devices only have one parameter
-            setsockopt(socket, IPPROTO_TCP, TCP_KEEPALIVE, &start, sizeof(uint32_t));
-#else
-            // Linux based systems
-            setsockopt(socket, SOL_TCP, TCP_KEEPIDLE, &start, sizeof(uint32_t));
-            setsockopt(socket, SOL_TCP, TCP_KEEPINTVL, &interval, sizeof(uint32_t));
-            setsockopt(socket, SOL_TCP, TCP_KEEPCNT, &cnt, sizeof(uint32_t));
-#endif
-#endif
+            this->set_keepalive(30);
         }
         if (cb)
             cb(ec, endpoint);
@@ -693,6 +656,64 @@ Connection::async_read_some(size_t bytes, BytesHandlerCb cb)
     else              socket_->async_read_some(buf, onEnd);
 }
 
+void
+Connection::set_keepalive(uint32_t seconds)
+{
+    std::lock_guard<std::mutex> lock(mutex_);
+    if (!ssl_socket_ && !socket_) return;
+
+    auto& base = ssl_socket_? ssl_socket_->lowest_layer() : *socket_;
+    auto socket = base.native_handle();
+
+    uint32_t interval = 1;
+    uint32_t cnt = 10;
+#ifdef _WIN32
+    // TCP_KEEPIDLE and TCP_KEEPINTVL are available since Win 10 version 1709
+    // TCP_KEEPCNT since Win 10 version 1703
+#if defined(TCP_KEEPIDLE) && defined(TCP_KEEPINTVL) && defined(TCP_KEEPCNT)
+    std::string val = "1";
+    setsockopt(socket, SOL_SOCKET, SO_KEEPALIVE, val.c_str(), sizeof(val));
+    std::string seconds_str = std::to_string(seconds);
+    setsockopt(socket, IPPROTO_TCP, TCP_KEEPIDLE,
+        seconds_str.c_str(), sizeof(seconds_str));
+    std::string interval_str = std::to_string(interval);
+    setsockopt(socket, IPPROTO_TCP, TCP_KEEPINTVL,
+        interval_str.c_str(), sizeof(interval_str));
+    std::string cnt_str = std::to_string(cnt);
+    setsockopt(socket, IPPROTO_TCP, TCP_KEEPCNT,
+        cnt_str.c_str(), sizeof(cnt_str));
+#else
+    struct {
+        uint32_t onoff;
+        uint32_t keepalivetime;
+        uint32_t keepaliveinterval;
+    } keepalive;
+    keepalive.onoff = 1;
+    keepalive.keepalivetime = seconds * 1000;
+    keepalive.keepaliveinterval = interval * 1000;
+    int32_t out = 0;
+    WSAIoctl(socket, SIO_KEEPALIVE_VALS, keepalive, sizeof(tcp_keepalive),
+        nullptr, 0, &out, nullptr, nullptr);
+#endif
+#else
+    uint32_t val = 1;
+    setsockopt(socket, SOL_SOCKET, SO_KEEPALIVE, &val, sizeof(uint32_t));
+#ifdef __APPLE__
+    // Old Apple devices only have one parameter
+    setsockopt(socket, IPPROTO_TCP, TCP_KEEPALIVE, &seconds, sizeof(uint32_t));
+#if defined(TCP_KEEPINTVL) && defined(TCP_KEEPCNT)
+    setsockopt(socket, IPPROTO_TCP, TCP_KEEPINTVL, &interval, sizeof(uint32_t));
+    setsockopt(socket, IPPROTO_TCP, TCP_KEEPCNT, &cnt, sizeof(uint32_t));
+#endif
+#else
+    // Linux based systems
+    setsockopt(socket, SOL_TCP, TCP_KEEPIDLE, &seconds, sizeof(uint32_t));
+    setsockopt(socket, SOL_TCP, TCP_KEEPINTVL, &interval, sizeof(uint32_t));
+    setsockopt(socket, SOL_TCP, TCP_KEEPCNT, &cnt, sizeof(uint32_t));
+#endif
+#endif
+}
+
 const asio::ip::address&
 Connection::local_address() const
 {
@@ -1116,13 +1137,11 @@ Request::init_parser()
     response_.request = shared_from_this();
 
     if (!parser_)
-        parser_ = std::make_unique<http_parser>();
-    http_parser_init(parser_.get(), HTTP_RESPONSE);
-    parser_->data = static_cast<void*>(this);
+        parser_ = std::make_unique<llhttp_t>();
 
     if (!parser_s_)
-        parser_s_ = std::make_unique<http_parser_settings>();
-    http_parser_settings_init(parser_s_.get());
+        parser_s_ = std::make_unique<llhttp_settings_t>();
+    llhttp_settings_init(parser_s_.get());
 
     cbs_.on_status = [this, statusCb = std::move(cbs_.on_status)](unsigned int status_code){
         response_.status_code = status_code;
@@ -1137,31 +1156,33 @@ Request::init_parser()
         response_.headers[*header_field] = std::string(at, length);
     };
 
-    // http_parser raw c callback (note: no context can be passed into them)
-    parser_s_->on_status = [](http_parser* parser, const char* /*at*/, size_t /*length*/) -> int {
+    // llhttp raw c callback (note: no context can be passed into them)
+    parser_s_->on_status = [](llhttp_t* parser, const char* /*at*/, size_t /*length*/) -> int {
         static_cast<Request*>(parser->data)->cbs_.on_status(parser->status_code);
         return 0;
     };
-    parser_s_->on_header_field = [](http_parser* parser, const char* at, size_t length) -> int {
+    parser_s_->on_header_field = [](llhttp_t* parser, const char* at, size_t length) -> int {
         static_cast<Request*>(parser->data)->cbs_.on_header_field(at, length);
         return 0;
     };
-    parser_s_->on_header_value = [](http_parser* parser, const char* at, size_t length) -> int {
+    parser_s_->on_header_value = [](llhttp_t* parser, const char* at, size_t length) -> int {
         static_cast<Request*>(parser->data)->cbs_.on_header_value(at, length);
         return 0;
     };
-    parser_s_->on_body = [](http_parser* parser, const char* at, size_t length) -> int {
+    parser_s_->on_body = [](llhttp_t* parser, const char* at, size_t length) -> int {
         static_cast<Request*>(parser->data)->onBody(at, length);
         return 0;
     };
-    parser_s_->on_headers_complete = [](http_parser* parser) -> int {
+    parser_s_->on_headers_complete = [](llhttp_t* parser) -> int {
         static_cast<Request*>(parser->data)->onHeadersComplete();
         return 0;
     };
-    parser_s_->on_message_complete = [](http_parser* parser) -> int {
+    parser_s_->on_message_complete = [](llhttp_t* parser) -> int {
         static_cast<Request*>(parser->data)->onComplete();
         return 0;
     };
+    llhttp_init(parser_.get(), HTTP_RESPONSE, parser_s_.get());
+    parser_->data = static_cast<void*>(this);
 }
 
 void
@@ -1318,7 +1339,7 @@ Request::terminate(const asio::error_code& ec)
             logger_->debug("[http:request:{:d}] done with status code {:d}", id_, response_.status_code);
     }
 
-    if (!parser_ or !http_should_keep_alive(parser_.get()))
+    if (!parser_ or !llhttp_should_keep_alive(parser_.get()))
         if (auto c = conn_)
             c->close();
     notify_state_change(State::DONE);
@@ -1357,17 +1378,15 @@ Request::handle_response(const asio::error_code& ec, size_t /* n_bytes */)
         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()) {
+    enum llhttp_errno ret = llhttp_execute(parser_.get(), request.c_str(), request.size());
+    if (ret != HPE_OK && ret != HPE_PAUSED) {
         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_)));
+            logger_->e("Error parsing HTTP: %zu %s %d", (int)ret, llhttp_errno_name(ret), llhttp_get_error_reason(parser_.get()));
         terminate(asio::error::basic_errors::broken_pipe);
         return;
     }
 
-    if (state_ != State::DONE and parser_ and not http_body_is_final(parser_.get())) {
+    if (state_ != State::DONE and parser_ and not llhttp_message_needs_eof(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){

src/network_utils.cpp

@@ -25,11 +25,16 @@
 #include <cstring>
 #define close(x) closesocket(x)
 #define write(s, b, f) send(s, b, (int)strlen(b), 0)
+#define _poll(fds, nfds, timeout) WSAPoll(fds, nfds, timeout)
 #else
-#include <sys/select.h>
 #include <fcntl.h>
+#include <poll.h>
+#define _poll(fds, nfds, timeout) poll(fds, nfds, timeout)
 #endif
 
+// number of file descriptors to poll in openSockets()
+#define NUM_FDS 3
+
 #include <iostream>
 
 namespace dht {
@@ -235,23 +240,23 @@ UdpSocket::openSockets(const SockAddr& bind4, const SockAddr& bind6)
 
     running = true;
     rcv_thread = std::thread([this, stop_readfd, ls4=s4, ls6=s6]() mutable {
-        int selectFd = std::max({ls4, ls6, stop_readfd}) + 1;
+        struct pollfd fds[NUM_FDS];
+        for (int i = 0; i < NUM_FDS; i++)
+            fds[i].events = POLLIN;
+        constexpr size_t stop_readfd_index = 0, ls4_index = 1, ls6_index = 2;
         try {
             while (running) {
-                fd_set readfds;
+                // ls4 and ls6 can be negative, but this doesn't require any special handling
+                // because poll() will simply ignore them (and set revents to 0) in that case
+                fds[stop_readfd_index].fd = stop_readfd;
+                fds[ls4_index].fd = ls4;
+                fds[ls6_index].fd = ls6;
 
-                FD_ZERO(&readfds);
-                FD_SET(stop_readfd, &readfds);
-                if(ls4 >= 0)
-                    FD_SET(ls4, &readfds);
-                if(ls6 >= 0)
-                    FD_SET(ls6, &readfds);
-
-                int rc = select(selectFd, &readfds, nullptr, nullptr, nullptr);
+                int rc = _poll(fds, NUM_FDS, -1);
                 if (rc < 0) {
                     if (errno != EINTR) {
                         if (logger)
-                            logger->e("Select error: %s", strerror(errno));
+                            logger->e("Poll error: %s", strerror(errno));
                         std::this_thread::sleep_for(std::chrono::seconds(1));
                     }
                 }
@@ -264,16 +269,16 @@ UdpSocket::openSockets(const SockAddr& bind4, const SockAddr& bind6)
                     sockaddr_storage from;
                     socklen_t from_len = sizeof(from);
 
-                    if (FD_ISSET(stop_readfd, &readfds)) {
+                    if (fds[stop_readfd_index].revents & POLLIN) {
                         if (recv(stop_readfd, (char*)buf.data(), buf.size(), 0) < 0) {
                             if (logger)
                                 logger->e("Got stop packet error: %s", strerror(errno));
                             break;
                         }
                     }
-                    else if (ls4 >= 0 && FD_ISSET(ls4, &readfds))
+                    else if (fds[ls4_index].revents & POLLIN)
                         rc = recvfrom(ls4, (char*)buf.data(), buf.size(), 0, (sockaddr*)&from, &from_len);
-                    else if (ls6 >= 0 && FD_ISSET(ls6, &readfds))
+                    else if (fds[ls6_index].revents & POLLIN)
                         rc = recvfrom(ls6, (char*)buf.data(), buf.size(), 0, (sockaddr*)&from, &from_len);
                     else
                         continue;
@@ -316,7 +321,6 @@ UdpSocket::openSockets(const SockAddr& bind4, const SockAddr& bind6)
                                     break;
                                 s4 = ls4;
                                 s6 = ls6;
-                                selectFd = std::max({ls4, ls6, stop_readfd}) + 1;
                             } else {
                                 break;
                             }

src/op_cache.h

@@ -97,8 +97,7 @@ public:
     }) {}
 
     bool onValue(const std::vector<Sp<Value>>& vals, bool expired) {
-        cache.onValue(vals, expired);
-        return not listeners.empty();
+        return cache.onValue(vals, expired);
     }
     void onNodeChanged(ListenSyncStatus status) {
         cache.onNodeChanged(status);

src/thread_pool.cpp

@@ -85,7 +85,7 @@ ThreadPool::run(std::function<void()>&& cb)
                         auto waitCond = [&](){ return not running_ or not tasks_.empty(); };
                         if (permanent_thread)
                             cv_.wait(l, waitCond);
-                        else 
+                        else
                             cv_.wait_for(l, e, waitCond);
                         readyThreads_--;
                         if (not running_ or tasks_.empty())
@@ -166,6 +166,16 @@ ThreadPool::join()
     tasks_ = {};
 }
 
+void
+ThreadPool::detach()
+{
+    stop(false);
+    for (auto& t : threads_)
+        t->detach();
+    threads_.clear();
+    tasks_ = {};
+}
+
 void
 Executor::run(std::function<void()>&& task)
 {

tests/dhtproxy_stress_tester.cpp

@@ -0,0 +1,179 @@
+/*
+ *  Copyright (C) 2014-2024 Savoir-faire Linux Inc.
+ *
+ *  This program is free software; you can redistribute it and/or modify
+ *  it under the terms of the GNU General Public License as published by
+ *  the Free Software Foundation; either version 3 of the License, or
+ *  (at your option) any later version.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+
+#include "dhtproxy_stress_tester.h"
+
+#include <iostream>
+#include <string>
+#include <chrono>
+#include <condition_variable>
+
+using namespace std::chrono_literals;
+
+namespace test {
+CPPUNIT_TEST_SUITE_REGISTRATION(DhtProxyStressTester);
+
+void
+DhtProxyStressTester::setUp() {
+    logger = dht::log::getStdLogger();
+
+    dht::DhtRunner::Context ctx;
+    ctx.logger = logger;
+
+    clientConfig.dht_config.node_config.max_peer_req_per_sec = -1;
+    clientConfig.dht_config.node_config.max_req_per_sec = -1;
+
+    nodePeer.run(0, clientConfig, std::move(ctx));
+
+    ctx.logger = logger;
+
+    nodeProxy = std::make_shared<dht::DhtRunner>();
+    nodeProxy->run(0, clientConfig, std::move(ctx));
+    nodeProxy->bootstrap(nodePeer.getBound());
+
+    dht::ProxyServerConfig serverConfig;
+    serverConfig.port = 8084;
+    serverProxy = std::make_unique<dht::DhtProxyServer>(nodeProxy, serverConfig, logger);
+    clientConfig.proxy_server = "http://127.0.0.1:8084";
+}
+
+void
+DhtProxyStressTester::tearDown() {
+    nodePeer.join();
+    nodeClient.join();
+
+    bool done = false;
+    std::condition_variable cv;
+    std::mutex cv_m;
+    nodeProxy->shutdown([&]{
+        std::lock_guard<std::mutex> lk(cv_m);
+        done = true;
+        cv.notify_all();
+    });
+    std::unique_lock<std::mutex> lk(cv_m);
+    CPPUNIT_ASSERT(cv.wait_for(lk, 15s, [&]{ return done; }));
+    serverProxy.reset();
+    nodeProxy.reset();
+}
+
+void
+DhtProxyStressTester::testRepeatValues() {
+    dht::DhtRunner::Context ctx;
+    ctx.logger = logger;
+    nodeClient.run(0, clientConfig, std::move(ctx));
+
+    std::condition_variable cv;
+    std::mutex cv_m;
+    std::unique_lock<std::mutex> lk(cv_m);
+    auto key = dht::InfoHash::get("GLaDOs");
+    bool done = false;
+
+    // If a peer send a value, the listen operation from the client
+    // should retrieve this value
+    dht::Value firstVal {"Hey! It's been a long time. How have you been?"};
+    auto firstVal_data = firstVal.data;
+    nodePeer.put(key, std::move(firstVal), [&](bool ok) {
+        CPPUNIT_ASSERT(ok);
+        std::lock_guard<std::mutex> lk(cv_m);
+        done = true;
+        cv.notify_all();
+    });
+    CPPUNIT_ASSERT(cv.wait_for(lk, 10s, [&]{ return done; }));
+    done = false;
+
+    std::vector<dht::Blob> values;
+    auto token = nodeClient.listen(key, [&](const std::vector<std::shared_ptr<dht::Value>>& v, bool expired) {
+        if (not expired) {
+            std::lock_guard<std::mutex> lk(cv_m);
+            for (const auto& value : v)
+                values.emplace_back(value->data);
+            done = true;
+            cv.notify_all();
+        }
+        return true;
+    });
+    CPPUNIT_ASSERT(cv.wait_for(lk, 10s, [&]{ return done; }));
+    done = false;
+    // Here values should contains 1 values
+    CPPUNIT_ASSERT_EQUAL(static_cast<int>(values.size()), 1);
+    CPPUNIT_ASSERT(values.front() == firstVal_data);
+
+    // And the listen should retrieve futures values
+    dht::Value secondVal {"You're a monster"};
+    auto secondVal_data = secondVal.data;
+    nodePeer.put(key, std::move(secondVal));
+    CPPUNIT_ASSERT(cv.wait_for(lk, 10s, [&]{ return done; }));
+    nodeClient.cancelListen(key, std::move(token));
+    CPPUNIT_ASSERT_EQUAL(static_cast<int>(values.size()), 2);
+    CPPUNIT_ASSERT(values.back() == secondVal_data);
+    done = false;
+
+    clientConfig.proxy_server = {};
+    constexpr auto DELAY = 60s;
+    dht::Value::Id id = 42;
+    for (auto now = std::chrono::steady_clock::now(), end = now + 12min; now < end; now += DELAY) {
+        auto tmpNode = std::make_shared<dht::DhtRunner>();
+        dht::DhtRunner::Context ctx;
+        ctx.logger = logger;
+        tmpNode->run(0, clientConfig, std::move(ctx));
+        tmpNode->bootstrap(nodePeer.getBound());
+        // 1 minute
+        std::this_thread::sleep_until(now + DELAY);
+        auto val = std::make_shared<dht::Value>("I'm doing science and I'm still alive");
+        val->id = id++;
+        nodePeer.put(key, val, [&](bool ok) {
+            CPPUNIT_ASSERT(ok);
+            std::lock_guard<std::mutex> lk(cv_m);
+            done = true;
+            cv.notify_all();
+        });
+        CPPUNIT_ASSERT(cv.wait_for(lk, 20s, [&]{ return done; }));
+        done = false;
+        // temporary node to make sure the value is retrieved
+        auto tmpValues = tmpNode->get(key);
+
+        auto token = nodeClient.listen(key, [&, val](const std::vector<std::shared_ptr<dht::Value>>& v, bool expired) {
+            if (not expired) {
+                for (const auto& value : v) {
+                    if (value->id == val->id) {
+                        std::lock_guard<std::mutex> lk(cv_m);
+                        done = true;
+                        cv.notify_all();
+                        break;
+                    }
+                }
+            }
+            return true;
+        });
+        CPPUNIT_ASSERT(cv.wait_for(lk, 25s, [&]{ return done; }));
+        done = false;
+
+        bool found_value = false;
+        for (const auto& value : tmpValues.get()) {
+            if (value->data == val->data) {
+                found_value = true;
+                break;
+            }
+        }
+        CPPUNIT_ASSERT(found_value);
+
+        nodeClient.cancelListen(key, std::move(token));
+        tmpNode->shutdown();
+    }
+}
+
+}  // namespace test

tests/dhtproxy_stress_tester.h

@@ -0,0 +1,59 @@
+/*
+ *  Copyright (C) 2014-2023 Savoir-faire Linux Inc.
+ *
+ *  This program is free software; you can redistribute it and/or modify
+ *  it under the terms of the GNU General Public License as published by
+ *  the Free Software Foundation; either version 3 of the License, or
+ *  (at your option) any later version.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+
+#pragma once
+
+// cppunit
+#include <cppunit/TestFixture.h>
+#include <cppunit/extensions/HelperMacros.h>
+
+#include <opendht/dhtrunner.h>
+#include <opendht/dht_proxy_server.h>
+#include <opendht/log.h>
+
+namespace test {
+
+class DhtProxyStressTester : public CppUnit::TestFixture {
+    CPPUNIT_TEST_SUITE(DhtProxyStressTester);
+    CPPUNIT_TEST(testRepeatValues);
+    CPPUNIT_TEST_SUITE_END();
+
+ public:
+    /**
+     * Method automatically called before each test by CppUnit
+     * Init nodes
+     */
+   void setUp();
+    /**
+     * Method automatically called after each test CppUnit
+     */
+   void tearDown();
+    /**
+     * Test get and put methods
+     */
+   void testRepeatValues();
+
+ private:
+    std::shared_ptr<dht::log::Logger> logger;
+    dht::DhtRunner::Config clientConfig {};
+    dht::DhtRunner nodePeer;
+    dht::DhtRunner nodeClient;
+    std::shared_ptr<dht::DhtRunner> nodeProxy;
+    std::unique_ptr<dht::DhtProxyServer> serverProxy;
+};
+
+}  // namespace test

tests/dhtproxytester.cpp

@@ -43,18 +43,20 @@ DhtProxyTester::setUp() {
     nodeProxy->run(0, clientConfig);
     nodeProxy->bootstrap(nodePeer.getBound());
 
-    auto serverCAIdentity = dht::crypto::generateEcIdentity("DHT Node CA");
+    //auto serverCAIdentity = dht::crypto::generateEcIdentity("DHT Node CA");
+
+    uint16_t port = 1024 + (std::rand() % (65535 - 1024));
 
     dht::ProxyServerConfig serverConfig;
     //serverConfig.identity = dht::crypto::generateIdentity("DHT Node", serverCAIdentity);
-    serverConfig.port = 8080;
+    serverConfig.port = port;
     // serverConfig.pushServer = "127.0.0.1:8090";
     serverProxy = std::make_unique<dht::DhtProxyServer>(nodeProxy, serverConfig);
 
     /*clientConfig.server_ca = serverCAIdentity.second;
     clientConfig.client_identity = dht::crypto::generateIdentity("DhtProxyTester");
     clientConfig.push_node_id = "dhtnode";*/
-    clientConfig.proxy_server = "http://127.0.0.1:8080";//"https://127.0.0.1:8080";
+    clientConfig.proxy_server = "http://127.0.0.1:" + std::to_string(port);//"https://127.0.0.1:8080";
 }
 
 void