readme: update

2026-11-09 04:49:14 +00:00 · 2018-02-28 21:36:17 -05:00
52 changed files with 1082 additions and 2965 deletions
@@ -48,7 +48,6 @@ Makefile.in

 # Python
 *.pyc
-*.stamp
 python/opendht.cpp
 python/setup.py

@@ -61,6 +60,3 @@ doc/Doxyfile
 # vim swap files
 *.swp
 *.swo
-
-# build dir
-build
@@ -11,13 +11,18 @@ env:
    - OPENDHT_TEST_JOB="opendht.llvm"
    - OPENDHT_TEST_JOB="opendht.proxyserver"
    - OPENDHT_TEST_JOB="opendht.proxyclient"
-    - OPENDHT_TEST_JOB="opendht.push"
+    - OPENDHT_TEST_JOB="opendht.proxyserverpush"
+    - OPENDHT_TEST_JOB="opendht.proxyclientpush"
+    - OPENDHT_TEST_JOB="opendht.all"

 before_install:
  - |
    # non llvm builds
    if [[ "$OPENDHT_TEST_JOB" != *"opendht.llvm"* ]]; then
      docker pull aberaud/opendht-deps;
+      if [[ "$OPENDHT_TEST_JOB" != *"opendht.classic"* ]]; then
+        docker build -t opendht-deps-proxy -f docker/DockerfileDepsProxy .;
+      fi
    fi

  - |
@@ -38,22 +43,22 @@ script:
    if [[ "$OPENDHT_TEST_JOB" != *"opendht.llvm"* ]] && [[ "$OPENDHT_TEST_JOB" != *"opendht.classic"* ]]; then
      docker build -t opendht-proxy -f docker/DockerfileTravisProxy .;
      options='';
-      if [[ "$OPENDHT_TEST_JOB" == *"opendht.proxyserver"* ]] || [[ "$OPENDHT_TEST_JOB" == *"opendht.push"* ]]; then
+      if [[ "$OPENDHT_TEST_JOB" == *"opendht.proxyserver"* ]] || [[ "$OPENDHT_TEST_JOB" == *"opendht.proxyserverpush"* ]] || [[ "$OPENDHT_TEST_JOB" == *"opendht.all"* ]]; then
        options+='-DOPENDHT_PROXY_SERVER=ON ';
      else
        options+='-DOPENDHT_PROXY_SERVER=OFF ';
      fi
-      if [[ "$OPENDHT_TEST_JOB" == *"opendht.proxyclient"* ]] || [[ "$OPENDHT_TEST_JOB" == *"opendht.push"* ]]; then
+      if [[ "$OPENDHT_TEST_JOB" == *"opendht.proxyclient"* ]] || [[ "$OPENDHT_TEST_JOB" == *"opendht.proxyclientpush"* ]] || [[ "$OPENDHT_TEST_JOB" == *"opendht.all"* ]]; then
        options+='-DOPENDHT_PROXY_CLIENT=ON ';
      else
        options+='-DOPENDHT_PROXY_CLIENT=OFF ';
      fi
-      if [[ "$OPENDHT_TEST_JOB" == *"opendht.push"* ]]; then
+      if [[ "$OPENDHT_TEST_JOB" == *"opendht.proxyserverpush"* ]] || [[ "$OPENDHT_TEST_JOB" == *"opendht.proxyclientpush"* ]] || [[ "$OPENDHT_TEST_JOB" == *"opendht.all"* ]]; then
        options+='-DOPENDHT_PUSH_NOTIFICATIONS=ON ';
      else
        options+='-DOPENDHT_PUSH_NOTIFICATIONS=OFF ';
      fi
-      docker run opendht-proxy /bin/sh -c "cd /root/opendht && mkdir build && cd build && cmake -DCMAKE_INSTALL_PREFIX=/usr -DOPENDHT_PYTHON=ON -DOPENDHT_LTO=ON -DOPENDHT_TESTS=ON $options .. && make -j8 && ./opendht_unit_tests && make install";
+      docker run opendht-proxy /bin/sh -c "cd /root/opendht && mkdir build && cd build && cmake -DCMAKE_INSTALL_PREFIX=/usr -DOPENDHT_PYTHON=ON -DOPENDHT_LTO=ON $options .. && make -j8 && make install";
    fi

  - |
@@ -1,7 +1,7 @@
 cmake_minimum_required (VERSION 3.1)
 project (opendht)
 set (opendht_VERSION_MAJOR 1)
-set (opendht_VERSION_MINOR 7.1)
+set (opendht_VERSION_MINOR 6.0)
 set (opendht_VERSION ${opendht_VERSION_MAJOR}.${opendht_VERSION_MINOR})
 set (PACKAGE_VERSION ${opendht_VERSION})
 set (VERSION "${opendht_VERSION}")
@@ -21,7 +21,6 @@ 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 Restbed and jsoncpp" OFF)
 option (OPENDHT_INDEX "Build DHT indexation feature" OFF)
-option (OPENDHT_TESTS "Add unit tests executable" OFF)

 find_package(Doxygen)
 option (OPENDHT_DOCUMENTATION "Create and install the HTML based API documentation (requires Doxygen)" ${DOXYGEN_FOUND})
@@ -121,12 +120,9 @@ list (APPEND opendht_SOURCES
    src/node.cpp
    src/value.cpp
    src/dht.cpp
-    src/op_cache.cpp
    src/storage.h
    src/listener.h
    src/search.h
-    src/value_cache.h
-    src/op_cache.h
    src/net.h
    src/parsed_message.h
    src/request.h
@@ -204,9 +200,6 @@ else ()
 endif ()

 if (OPENDHT_PROXY_SERVER OR OPENDHT_PROXY_CLIENT)
-  list (APPEND opendht_HEADERS
-    include/opendht/proxy.h
-  )
  list (APPEND opendht_SOURCES
    src/base64.h
    src/base64.cpp
@@ -299,28 +292,3 @@ install (DIRECTORY include DESTINATION ${CMAKE_INSTALL_PREFIX})
 install (FILES ${CMAKE_CURRENT_BINARY_DIR}/opendht.pc DESTINATION ${CMAKE_INSTALL_LIBDIR}/pkgconfig)
 install (EXPORT opendht DESTINATION ${CMAKE_INSTALL_LIBDIR}/cmake/opendht FILE opendhtConfig.cmake)
 install (FILES ${CMAKE_CURRENT_BINARY_DIR}/opendhtConfigVersion.cmake DESTINATION ${CMAKE_INSTALL_LIBDIR}/cmake/opendht)
-
-# Unit tests
-IF(OPENDHT_TESTS)
-  FIND_PACKAGE(Cppunit REQUIRED)
-  # unit testing
-    add_executable(opendht_unit_tests
-      tests/tests_runner.cpp
-      tests/infohashtester.h
-      tests/infohashtester.cpp
-      tests/dhtrunnertester.h
-      tests/dhtrunnertester.cpp
-    )
-    if (OPENDHT_SHARED)
-      TARGET_LINK_LIBRARIES(opendht_unit_tests opendht)
-    else ()
-      TARGET_LINK_LIBRARIES(opendht_unit_tests opendht-static)
-    endif ()
-    TARGET_LINK_LIBRARIES(opendht_unit_tests
-       ${CMAKE_THREAD_LIBS_INIT}
-       ${CPPUNIT_LIBRARIES}
-       ${GNUTLS_LIBRARIES}
-    )
-    enable_testing()
-    add_test(TEST opendht_unit_tests)
-ENDIF()
@@ -12,10 +12,6 @@ if USE_CYTHON
 SUBDIRS += python
 endif

-if ENABLE_TESTS
-SUBDIRS += tests
-endif
-
 SUBDIRS += doc

 ACLOCAL_AMFLAGS = -I m4
@@ -6,17 +6,12 @@

 A lightweight C++11 Distributed Hash Table implementation.

-OpenDHT provides an easy to use distributed in-memory data store.
-Every node in the network can read and write values to the store.
-Values are distributed over the network, with redundancy.
-
- * Lightweight and scalable, designed for large networks and small devices
- * High resilience to network disruption
+ * Light and fast C++11 Kademlia DHT library
+ * Distributed shared key->value data-store
+ * Clean and powerful distributed map API with storage of arbitrary binary values of up to 56 KB
 * Public key cryptography layer providing optional data signature and encryption (using GnuTLS)
 * IPv4 and IPv6 support
- * Clean and powerful C++11 map API
- * Python 3 bindings
- * REST API
+ * Python binding

 ## Documentation
 See the wiki: <https://github.com/savoirfairelinux/opendht/wiki>
@@ -105,14 +100,14 @@ for r in results:

 IRC: join us on Freenode at [`#opendht`](https://webchat.freenode.net/?channels=%23opendht).

-## License
+## Licence
 Copyright (C) 2014-2018 Savoir-faire Linux Inc.

 OpenDHT 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.

-See COPYING or https://www.gnu.org/licenses/gpl-3.0.en.html for the full GPLv3 license.
+See COPYING or https://www.gnu.org/licenses/gpl-3.0.en.html for the full GPLv3 licence.

 ## Acknowledgements
 This project was originally based on https://github.com/jech/dht by Juliusz Chroboczek.
@@ -1,28 +0,0 @@
-INCLUDE(FindPkgConfig)
-PKG_CHECK_MODULES(PC_CPPUNIT "cppunit")
-
-FIND_PATH(CPPUNIT_INCLUDE_DIRS
-    NAMES cppunit/TestCase.h
-    HINTS ${PC_CPPUNIT_INCLUDE_DIR}
-    ${CMAKE_INSTALL_PREFIX}/include
-    PATHS
-    /usr/local/include
-    /usr/include
-)
-
-FIND_LIBRARY(CPPUNIT_LIBRARIES
-    NAMES cppunit
-    HINTS ${PC_CPPUNIT_LIBDIR}
-    ${CMAKE_INSTALL_PREFIX}/lib
-    ${CMAKE_INSTALL_PREFIX}/lib64
-    PATHS
-    ${CPPUNIT_INCLUDE_DIRS}/../lib
-    /usr/local/lib
-    /usr/lib
-)
-
-LIST(APPEND CPPUNIT_LIBRARIES ${CMAKE_DL_LIBS})
-
-INCLUDE(FindPackageHandleStandardArgs)
-FIND_PACKAGE_HANDLE_STANDARD_ARGS(CPPUNIT DEFAULT_MSG CPPUNIT_LIBRARIES CPPUNIT_INCLUDE_DIRS)
-MARK_AS_ADVANCED(CPPUNIT_LIBRARIES CPPUNIT_INCLUDE_DIRS)
@@ -1,7 +1,7 @@
 dnl define macros
 m4_define([opendht_major_version], 1)
-m4_define([opendht_minor_version], 7)
-m4_define([opendht_patch_version], 1)
+m4_define([opendht_minor_version], 6)
+m4_define([opendht_patch_version], 0)
 m4_define([opendht_version],
 		  [opendht_major_version.opendht_minor_version.opendht_patch_version])

@@ -25,7 +25,7 @@ AC_PROG_CXX
 AM_PROG_AR

 dnl Check for logs
-AC_ARG_ENABLE([logs], [AS_HELP_STRING([--disable-logs], [Disable DHT logs])])
+AC_ARG_ENABLE([logs], [AS_HELP_STRING([--disable-logs], [Disble DHT logs])])
 AS_IF([test "x$enable_logs" != "xno"], [
 	AC_DEFINE([OPENDHT_LOG], [true], [Define if DHT logs are enabled])
 ], [
@@ -33,9 +33,9 @@ AS_IF([test "x$enable_logs" != "xno"], [
 ])

 dnl Check for indexation
-AC_ARG_ENABLE([indexation], [AS_HELP_STRING([--disable-indexation], [Disable DHT indexation])])
-AM_CONDITIONAL(ENABLE_INDEXATION, test x$enable_indexation != "xno")
-AS_IF([test "x$enable_indexation" != "xno"], [
+AC_ARG_ENABLE([indexation], [AS_HELP_STRING([--enable-indexation], [Enable DHT indexation])])
+AM_CONDITIONAL(ENABLE_INDEXATION, test x$enable_indexation == xyes)
+AS_IF([test "x$enable_indexation" = "xyes"], [
 	AC_DEFINE([OPENDHT_INDEXATION], [1], [Define if DHT indexation is enabled])
 ])

@@ -182,11 +182,6 @@ AM_COND_IF([PROXY_CLIENT_OR_SERVER], [
 	LDFLAGS="${LDFLAGS} ${Jsoncpp_LIBS} -lrestbed"
 ], [])

-AC_ARG_ENABLE([tests], AS_HELP_STRING([--enable-tests], [Enable tests]), build_tests=yes, build_tests=no)
-AM_CONDITIONAL(ENABLE_TESTS, test x$build_tests == xyes)
-AM_COND_IF([ENABLE_TESTS], [
-	PKG_CHECK_MODULES([CppUnit], [cppunit >= 1.12])
-])

 AC_CONFIG_FILES([doc/Doxyfile doc/Makefile])

@@ -202,6 +197,5 @@ AM_COND_IF([USE_CYTHON], [
 AC_CONFIG_FILES([Makefile
                 src/Makefile
                 tools/Makefile
-								 tests/Makefile
                 opendht.pc])
 AC_OUTPUT
@@ -1,18 +1,3 @@
-FROM ubuntu:16.04
+FROM ubuntu:17.04
 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 libcppunit-dev libjsoncpp-dev libasio-dev libssl-dev python3-setuptools && apt-get clean
-
-# build restbed from sources
-RUN git clone --recursive https://github.com/corvusoft/restbed.git \
-	&& cd restbed && mkdir build && cd build \
-	&& cmake -DBUILD_TESTS=NO -DBUILD_EXAMPLES=NO -DBUILD_SSL=NO -DBUILD_SHARED=YES -DCMAKE_INSTALL_PREFIX=/usr -DCMAKE_INSTALL_LIBDIR=lib .. \
-	&& make -j8 install \
-	&& cd .. && rm -rf restbed
-
-#build msgpack from source
-RUN wget https://github.com/msgpack/msgpack-c/releases/download/cpp-2.1.5/msgpack-2.1.5.tar.gz \
-	&& tar -xzf msgpack-2.1.5.tar.gz \
-	&& cd msgpack-2.1.5 && mkdir build && cd build \
-	&& cmake -DMSGPACK_CXX11=ON -DMSGPACK_BUILD_EXAMPLES=OFF -DCMAKE_INSTALL_PREFIX=/usr .. \
-	&& make -j8 && make install \
-	&& cd ../.. && rm -rf msgpack-2.1.5 msgpack-2.1.5.tar.gz
+RUN apt-get update && apt-get install -y build-essential cmake git wget libncurses5-dev libreadline-dev nettle-dev libgnutls28-dev libuv1-dev libmsgpack-dev libargon2-0-dev cython3 python3-dev python3-setuptools && apt-get clean
@@ -1,23 +1,7 @@
-FROM ubuntu:16.04
+FROM ubuntu:17.04
 MAINTAINER Adrien Béraud <adrien.beraud@savoirfairelinux.com>
 RUN apt-get update \
-	&& apt-get install -y llvm llvm-dev clang make cmake git wget libncurses5-dev libreadline-dev nettle-dev libgnutls28-dev libuv1-dev libmsgpack-dev libjsoncpp-dev libasio-dev cython3 python3-dev python3-setuptools libcppunit-dev \
+	&& apt-get install -y llvm llvm-dev clang make cmake git wget libncurses5-dev libreadline-dev nettle-dev libgnutls28-dev libuv1-dev libmsgpack-dev libargon2-0-dev cython3 python3-dev python3-setuptools \
 	&& apt-get remove -y gcc g++ && apt-get autoremove -y && apt-get clean
-
 ENV CC cc
-ENV CXX c++
-
-# build restbed from sources
-RUN git clone --recursive https://github.com/corvusoft/restbed.git \
-	&& cd restbed && mkdir build && cd build \
-	&& cmake -DBUILD_TESTS=NO -DBUILD_EXAMPLES=NO -DBUILD_SSL=NO -DBUILD_SHARED=YES -DCMAKE_INSTALL_PREFIX=/usr -DCMAKE_INSTALL_LIBDIR=lib .. \
-	&& make -j8 install \
-	&& cd .. && rm -rf restbed
-
-#build msgpack from source
-RUN wget https://github.com/msgpack/msgpack-c/releases/download/cpp-2.1.5/msgpack-2.1.5.tar.gz \
-	&& tar -xzf msgpack-2.1.5.tar.gz \
-	&& cd msgpack-2.1.5 && mkdir build && cd build \
-	&& cmake -DMSGPACK_CXX11=ON -DMSGPACK_BUILD_EXAMPLES=OFF -DCMAKE_INSTALL_PREFIX=/usr .. \
-	&& make -j8 && make install \
-	&& cd ../.. && rm -rf msgpack-2.1.5 msgpack-2.1.5.tar.gz
+ENV CXX c++
@@ -0,0 +1,16 @@
+FROM aberaud/opendht-deps
+
+# install jsoncpp
+RUN apt-get install libjsoncpp-dev -y
+
+# install restbed dependencies
+RUN sed -i -e 's/archive.ubuntu.com\|security.ubuntu.com/old-releases.ubuntu.com/g' /etc/apt/sources.list # LTS... avoid 404 for libasio-dev
+RUN apt-get update -y
+RUN apt-get install libasio-dev -y
+
+# build restbed from sources
+RUN git clone --recursive https://github.com/corvusoft/restbed.git
+WORKDIR restbed/build
+RUN cmake -DBUILD_TESTS=NO -DBUILD_EXAMPLES=NO -DBUILD_SSL=NO -DBUILD_SHARED=YES -DCMAKE_INSTALL_PREFIX=/usr ..
+RUN make -j 8 install
+RUN mv /usr/library/librestbed* /usr/lib/
@@ -1,7 +1,5 @@
 FROM aberaud/opendht-deps
 MAINTAINER Adrien Béraud <adrien.beraud@savoirfairelinux.com>
-
 COPY . /root/opendht
 RUN cd /root/opendht && mkdir build && cd build \
-	&& cmake -DCMAKE_INSTALL_PREFIX=/usr -DOPENDHT_PYTHON=On -DOPENDHT_LTO=On -DOPENDHT_TESTS=ON ..  \
-	&& make -j8 && ./opendht_unit_tests && make install
+	&& cmake -DCMAKE_INSTALL_PREFIX=/usr -DOPENDHT_PYTHON=On -DOPENDHT_LTO=On .. && make -j8 && make install
@@ -1,7 +1,5 @@
 FROM aberaud/opendht-deps-llvm
 MAINTAINER Adrien Béraud <adrien.beraud@savoirfairelinux.com>
-
 COPY . /root/opendht
 RUN cd /root/opendht && mkdir build && cd build \
-	&& cmake -DCMAKE_INSTALL_PREFIX=/usr -DOPENDHT_PYTHON=On -DOPENDHT_TESTS=ON .. \
-	&& make -j8 && ./opendht_unit_tests && make install
+	&& cmake -DCMAKE_INSTALL_PREFIX=/usr -DOPENDHT_PYTHON=On .. && make -j8 && make install
@@ -1,4 +1,3 @@
-FROM aberaud/opendht-deps
+FROM opendht-deps-proxy
 MAINTAINER Adrien Béraud <adrien.beraud@savoirfairelinux.com>
-
 COPY . /root/opendht
@@ -98,7 +98,6 @@ using ValuesExport = std::pair<InfoHash, Blob>;

 using QueryCallback = std::function<bool(const std::vector<std::shared_ptr<FieldValueIndex>>& fields)>;
 using GetCallback = std::function<bool(const std::vector<std::shared_ptr<Value>>& values)>;
-using ValueCallback = std::function<bool(const std::vector<std::shared_ptr<Value>>& values, bool expired)>;
 using GetCallbackSimple = std::function<bool(std::shared_ptr<Value> value)>;
 using ShutdownCallback = std::function<void()>;

@@ -99,10 +99,11 @@ public:
    bool isRunning(sa_family_t af = 0) const;

    virtual void registerType(const ValueType& type) {
-        types.registerType(type);
+        types[type.id] = type;
    }
    const ValueType& getType(ValueType::Id type_id) const {
-        return types.getType(type_id);
+        const auto& t_it = types.find(type_id);
+        return (t_it == types.end()) ? ValueType::USER_DATA : t_it->second;
    }

    /**
@@ -230,16 +231,8 @@ public:
     *
     * @return a token to cancel the listener later.
     */
-    virtual size_t listen(const InfoHash&, ValueCallback, Value::Filter={}, Where={});
-
-    virtual size_t listen(const InfoHash& key, GetCallback cb, Value::Filter f={}, Where w={}) {
-        return listen(key, [cb](const std::vector<Sp<Value>>& vals, bool expired){
-            if (not expired)
-                return cb(vals);
-            return true;
-        }, std::forward<Value::Filter>(f), std::forward<Where>(w));
-    }
-    virtual size_t listen(const InfoHash& key, GetCallbackSimple cb, Value::Filter f={}, Where w={}) {
+    virtual size_t listen(const InfoHash&, GetCallback, Value::Filter={}, Where w = {});
+    virtual size_t listen(const InfoHash& key, GetCallbackSimple cb, Value::Filter f={}, Where w = {}) {
        return listen(key, bindGetCb(cb), std::forward<Value::Filter>(f), std::forward<Where>(w));
    }

@@ -357,7 +350,7 @@ private:
    uint64_t oldsecret {};

    // registred types
-    TypeStore types;
+    std::map<ValueType::Id, ValueType> types;

    // are we a bootstrap node ?
    // note: Any running node can be used as a bootstrap node.
@@ -414,7 +407,7 @@ private:
    void expireStorage(InfoHash h);
    void expireStore(decltype(store)::iterator);

-    void storageChanged(const InfoHash& id, Storage& st, ValueStorage&, bool newValue);
+    void storageChanged(const InfoHash& id, Storage& st, ValueStorage&);
    std::string printStorageLog(const decltype(store)::value_type&) const;

    /**
@@ -467,7 +460,7 @@ private:
    Sp<Search> search(const InfoHash& id, sa_family_t af, GetCallback = {}, QueryCallback = {}, DoneCallback = {}, Value::Filter = {}, Query q = {});

    void announce(const InfoHash& id, sa_family_t af, Sp<Value> value, DoneCallback callback, time_point created=time_point::max(), bool permanent = false);
-    size_t listenTo(const InfoHash& id, sa_family_t af, ValueCallback cb, Value::Filter f = Value::AllFilter(), const Sp<Query>& q = {});
+    size_t listenTo(const InfoHash& id, sa_family_t af, GetCallback cb, Value::Filter f = Value::AllFilter(), const Sp<Query>& q = {});

    /**
     * Refill the search with good nodes if possible.
@@ -535,9 +528,7 @@ private:
     *
     * @param sr  The search to execute its operations.
     */
-    void searchStep(Sp<Search>);
-    void searchSynchedNodeListen(const Sp<Search>&, SearchNode&);
-
+    void searchStep(Sp<Search> sr);
    void dumpSearch(const Search& sr, std::ostream& out) const;

    bool neighbourhoodMaintenance(RoutingTable&);
@@ -165,7 +165,6 @@ public:
     */
    virtual size_t listen(const InfoHash&, GetCallback, Value::Filter={}, Where w = {}) = 0;
    virtual size_t listen(const InfoHash& key, GetCallbackSimple cb, Value::Filter f={}, Where w = {}) = 0;
-    virtual size_t listen(const InfoHash&, ValueCallback, Value::Filter={}, Where w = {}) = 0;

    virtual bool cancelListen(const InfoHash&, size_t token) = 0;

@@ -236,6 +235,11 @@ public:
     * @param notification to process
     */
    virtual void pushNotificationReceived(const std::map<std::string, std::string>& data) = 0;
+    /**
+     * Refresh a listen via a token
+     * @param token
+     */
+    virtual void resubscribe(unsigned token) = 0;

 protected:
    bool logFilerEnable_ {};
@@ -29,7 +29,6 @@
 #include "def.h"
 #include "dht_interface.h"
 #include "scheduler.h"
-#include "proxy.h"

 namespace restbed {
    class Request;
@@ -41,12 +40,10 @@ namespace Json {

 namespace dht {

-class SearchCache;
-
 class OPENDHT_PUBLIC DhtProxyClient final : public DhtInterface {
 public:

-    DhtProxyClient();
+    DhtProxyClient() : scheduler(DHT_LOG) {}

    explicit DhtProxyClient(std::function<void()> loopSignal, const std::string& serverHost, const std::string& pushClientId = "");

@@ -105,6 +102,8 @@ public:
        get(key, bindGetCb(cb), bindDoneCb(donecb), std::forward<Value::Filter>(f), std::forward<Where>(w));
    }

+    void get(const InfoHash& key, const GetCallback& cb, DoneCallback donecb, const Value::Filter& filterChain);
+
    /**
     * Announce a value on all available protocols (IPv4, IPv6).
     *
@@ -162,25 +161,22 @@ public:
     *
     * @return a token to cancel the listener later.
     */
-    virtual size_t listen(const InfoHash&, ValueCallback, Value::Filter={}, Where={});
-
-    virtual size_t listen(const InfoHash& key, GetCallback cb, Value::Filter f={}, Where w={}) {
-        return listen(key, [cb](const std::vector<Sp<Value>>& vals, bool expired){
-            if (not expired)
-                return cb(vals);
-            return true;
-        }, std::forward<Value::Filter>(f), std::forward<Where>(w));
-    }
-    virtual size_t listen(const InfoHash& key, GetCallbackSimple cb, Value::Filter f={}, Where w={}) {
+    virtual size_t listen(const InfoHash&, GetCallback, Value::Filter={}, Where={});
+    virtual size_t listen(const InfoHash& key, GetCallbackSimple cb, Value::Filter f={}, Where w = {}) {
        return listen(key, bindGetCb(cb), std::forward<Value::Filter>(f), std::forward<Where>(w));
    }
-    virtual bool cancelListen(const InfoHash& key, size_t token);
+    virtual bool cancelListen(const InfoHash&, size_t token);

    /**
     * Call linked callback with a push notification
     * @param notification to process
     */
    void pushNotificationReceived(const std::map<std::string, std::string>& notification);
+    /**
+     * Refresh a listen via a token
+     * @param token
+     */
+    void resubscribe(const unsigned token);

    time_point periodic(const uint8_t*, size_t, const SockAddr&);
    time_point periodic(const uint8_t *buf, size_t buflen, const sockaddr* from, socklen_t fromlen) {
@@ -206,31 +202,21 @@ public:
    /**
     * Get data currently being put at the given hash.
     */
-    std::vector<Sp<Value>> getPut(const InfoHash&);
+    std::vector<Sp<Value>> getPut(const InfoHash&) { return {}; }

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

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

    void pingNode(const sockaddr*, socklen_t, DoneCallbackSimple&& /*cb*/={}) { }

-    virtual void registerType(const ValueType& type) {
-        types.registerType(type);
-    }
-    const ValueType& getType(ValueType::Id type_id) const {
-        return types.getType(type_id);
-    }
-
-    std::vector<Sp<Value>> getLocal(const InfoHash& k, Value::Filter filter) const;
-    Sp<Value> getLocalById(const InfoHash& k, Value::Id id) const;
-
    /**
     * NOTE: The following methods will not be implemented because the
     * DhtProxyClient doesn't have any storage nor synchronization process
@@ -239,6 +225,10 @@ public:
    void insertNode(const InfoHash&, const sockaddr*, socklen_t) { }
    void insertNode(const NodeExport&) { }
    std::pair<size_t, size_t> getStoreSize() const { return {}; }
+    virtual void registerType(const ValueType&) { }
+    const ValueType& getType(ValueType::Id) const { return NO_VALUE; }
+    std::vector<Sp<Value>> getLocal(const InfoHash&, Value::Filter) const { return {}; }
+    Sp<Value> getLocalById(const InfoHash&, Value::Id) const { return {}; }
    std::vector<NodeExport> exportNodes() { return {}; }
    std::vector<ValuesExport> exportValues() const { return {}; }
    void importValues(const std::vector<ValuesExport>&) {}
@@ -250,16 +240,12 @@ public:
    void dumpTables() const {}
    std::vector<unsigned> getNodeMessageStats(bool) { return {}; }
    void setStorageLimit(size_t) {}
-    void connectivityChanged(sa_family_t) {
-        restartListeners();
-    }
-    void connectivityChanged() {
-        getProxyInfos();
-        restartListeners();
-        loopSignal_();
-    }
+    void connectivityChanged(sa_family_t) {}
+    void connectivityChanged() { }

 private:
+    const ValueType NO_VALUE;
+
    /**
     * Start the connection with a server.
     */
@@ -270,16 +256,11 @@ private:
     * @return the JSON returned by the proxy
     */
    void getProxyInfos();
-    void onProxyInfos(const Json::Value& val, sa_family_t family);
+    void onProxyInfos(const Json::Value& val);
    SockAddr parsePublicAddress(const Json::Value& val);

    void opFailed();

-    size_t doListen(const InfoHash& key, ValueCallback, Value::Filter);
-    bool doCancelListen(const InfoHash& key, size_t token);
-
-    void doPut(const InfoHash&, Sp<Value>, DoneCallback, time_point created, bool permanent);
-
    /**
     * Initialize statusIpvX_
     */
@@ -292,7 +273,6 @@ private:
     * cancel all Operations
     */
    void cancelAllOperations();
-
    std::string serverHost_;
    std::string pushClientId_;

@@ -301,23 +281,27 @@ private:
    NodeStatus statusIpv6_ {NodeStatus::Disconnected};
    NodeStats stats4_ {};
    NodeStats stats6_ {};
-    SockAddr publicAddressV4_;
-    SockAddr publicAddressV6_;
+    SockAddr publicAddress_;

    InfoHash myid {};

-    // registred types
-    TypeStore types;
-
    /**
     * Store listen requests.
     */
-    struct Listener;
-    struct ProxySearch;
-
-    size_t listenerToken_ {0};
-    std::map<InfoHash, ProxySearch> searches_;
-    mutable std::mutex searchLock_;
+    struct Listener
+    {
+        size_t token;
+        std::shared_ptr<restbed::Request> req;
+        std::string key;
+        GetCallback cb;
+        Value::Filter filterChain;
+        std::thread thread;
+        unsigned callbackId;
+        std::shared_ptr<unsigned> pushNotifToken; // NOTE: unused if not using push notifications
+    };
+    std::vector<Listener> listeners_;
+    size_t listener_token_ {0};
+    std::mutex lockListener_;

    /**
     * Store current put and get requests.
@@ -350,26 +334,20 @@ private:
     */
    void restartListeners();

-    /**
-     * Refresh a listen via a token
-     * @param token
-     */
-    void resubscribe(const InfoHash& key, Listener& listener);
-
    /**
     * If we want to use push notifications by default.
     * NOTE: empty by default to avoid to use services like FCM or APN.
     */
    std::string deviceKey_ {};
+    unsigned callbackId_ {0};
+    std::mutex lockCallback_;

    const std::function<void()> loopSignal_;

 #if OPENDHT_PUSH_NOTIFICATIONS
-    void fillBodyToGetToken(std::shared_ptr<restbed::Request> request, unsigned token = 0);
-    void getPushRequest(Json::Value&) const;
+    void fillBodyToGetToken(std::shared_ptr<restbed::Request> request, unsigned callbackId);
 #endif // OPENDHT_PUSH_NOTIFICATIONS

-    bool isDestroying_ {false};
 };

 }
@@ -1,7 +1,6 @@
 /*
 *  Copyright (C) 2017-2018 Savoir-faire Linux Inc.
 *  Author: Sébastien Blin <sebastien.blin@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
@@ -24,9 +23,6 @@
 #include "def.h"
 #include "sockaddr.h"
 #include "infohash.h"
-#include "scheduler.h"
-#include "value.h"
-#include "proxy.h"

 #include <thread>
 #include <memory>
@@ -63,27 +59,6 @@ public:
    DhtProxyServer& operator=(const DhtProxyServer& other) = delete;
    DhtProxyServer& operator=(DhtProxyServer&& other) = delete;

-    struct ServerStats {
-        /** Current number of listen operations */
-        size_t listenCount;
-        /** Current number of permanent put operations */
-        size_t putCount;
-        /** Current number of push tokens with at least one listen operation */
-        size_t pushListenersCount;
-        /** Average requests per second */
-        double requestRate;
-
-        std::string toString() const {
-            std::ostringstream ss;
-            ss << "Listens: " << listenCount << " Puts: " << putCount << " PushListeners: " << pushListenersCount << std::endl;
-            ss << "Requests: " << requestRate << " per second." << std::endl;
-            return ss.str();
-        }
-    };
-    ServerStats getStats() const;
-
-    std::shared_ptr<DhtRunner> getNode() const { return dht_; }
-
    /**
     * Stop the DhtProxyServer
     */
@@ -121,7 +96,7 @@ private:
     * On error: HTTP 503, body: {"err":"xxxx"}
     * @param session
     */
-    void listen(const std::shared_ptr<restbed::Session>& session);
+    void listen(const std::shared_ptr<restbed::Session>& session) const;

    /**
     * Put a value on the DHT
@@ -132,9 +107,7 @@ private:
     * HTTP 400, body: {"err":"xxxx"} if bad json or HTTP 502 if put fails
     * @param session
     */
-    void put(const std::shared_ptr<restbed::Session>& session);
-
-    void cancelPut(const InfoHash& key, Value::Id vid);
+    void put(const std::shared_ptr<restbed::Session>& session) const;

 #if OPENDHT_PROXY_SERVER_IDENTITY
    /**
@@ -181,63 +154,43 @@ private:
     */
    void handleOptionsMethod(const std::shared_ptr<restbed::Session>& session) const;

-    /**
-     * Remove finished listeners
-     * @param testSession if we remove the listener only if the session is closed
-     */
-    void removeClosedListeners(bool testSession = true);
-
 #if OPENDHT_PUSH_NOTIFICATIONS
    /**
     * Subscribe to push notifications for an iOS or Android device.
     * Method: SUBSCRIBE "/{InfoHash: .*}"
-     * Body: {"key": "device_key", (optional)"isAndroid":false (default true)}"
+     * Body: {"key": "device_key", (optional)"callback_id":y,
+     * (optional)"isAndroid":false (default true)}"
     * Return: {"token": x}" where x if a token to save
     * @note: the listen will timeout after six hours (and send a push notification).
     * so you need to refresh the operation each six hours.
+     * @note: callback_id is used to add the possibility to have multiple listen
+     * on same hash for same device and must be > 0
     * @param session
     */
-    void subscribe(const std::shared_ptr<restbed::Session>& session);
+    void subscribe(const std::shared_ptr<restbed::Session>& session) const;
    /**
     * Unsubscribe to push notifications for an iOS or Android device.
     * Method: UNSUBSCRIBE "/{InfoHash: .*}"
-     * Body: {"key": "device_key", "token": x} where x if the token to cancel
+     * Body: {"key": "device_key", "token": x, (optional)"callback_id":y"
+     * where x if the token to cancel
     * Return: nothing
+     * @note: callback id is used to add the possibility to have multiple listen
+     * on same hash for same device
     * @param session
     */
-    void unsubscribe(const std::shared_ptr<restbed::Session>& session);
+    void unsubscribe(const std::shared_ptr<restbed::Session>& session) const;
    /**
     * Send a push notification via a gorush push gateway
     * @param key of the device
     * @param json, the content to send
     */
    void sendPushNotification(const std::string& key, const Json::Value& json, bool isAndroid) const;
-
-    /**
-     * Remove a push listener between a client and a hash
-     * @param pushToken
-     * @param key
-     * @param clientId
-     */
-    void cancelPushListen(const std::string& pushToken, const InfoHash& key, const std::string& clientId);
-
-
 #endif //OPENDHT_PUSH_NOTIFICATIONS

-    using clock = std::chrono::steady_clock;
-    using time_point = clock::time_point;
-
    std::thread server_thread {};
    std::unique_ptr<restbed::Service> service_;
    std::shared_ptr<DhtRunner> dht_;

-    std::mutex schedulerLock_;
-    std::condition_variable schedulerCv_;
-    Scheduler scheduler_;
-    std::thread schedulerThread_;
-
-    Sp<Scheduler::Job> printStatsJob_;
-
    // Handle client quit for listen.
    // NOTE: can be simplified when we will supports restbed 5.0
    std::thread listenThread_;
@@ -246,26 +199,37 @@ private:
        InfoHash hash;
        std::future<size_t> token;
    };
-    std::vector<SessionToHashToken> currentListeners_;
-    std::mutex lockListener_;
+    mutable std::vector<SessionToHashToken> currentListeners_;
+    mutable std::mutex lockListener_;
    std::atomic_bool stopListeners {false};

-    struct PermanentPut;
-    struct SearchPuts;
-    std::map<InfoHash, SearchPuts> puts_;
-
-    mutable std::atomic<size_t> requestNum_ {0};
-    mutable std::atomic<time_point> lastStatsReset_ {time_point::min()};
-
+#if OPENDHT_PUSH_NOTIFICATIONS
+    struct PushListener {
+        std::string pushToken;
+        InfoHash hash;
+        unsigned token;
+        std::future<size_t> internalToken;
+        std::chrono::steady_clock::time_point deadline;
+        bool started {false};
+        unsigned callbackId {0};
+        std::string clientId {};
+        bool isAndroid {true};
+    };
+    mutable std::mutex lockPushListeners_;
+    mutable std::vector<PushListener> pushListeners_;
+    mutable unsigned tokenPushNotif_ {0};
+#endif //OPENDHT_PUSH_NOTIFICATIONS
    const std::string pushServer_;

-#if OPENDHT_PUSH_NOTIFICATIONS
-    struct Listener;
-    struct PushListener;
-    std::mutex lockPushListeners_;
-    std::map<std::string, PushListener> pushListeners_;
-    proxy::ListenToken tokenPushNotif_ {0};
-#endif //OPENDHT_PUSH_NOTIFICATIONS
+    /**
+     * Remove finished listeners
+     * @param testSession if we remove the listener only if the session is closed
+     */
+    void removeClosedListeners(bool testSession = true);
+    /**
+     * Launch or remove push listeners if needed
+     */
+    void handlePushListeners();
 };

 }
@@ -136,15 +136,7 @@ public:
        query(hash, cb, bindDoneCb(done_cb), q);
    }

-    std::future<size_t> listen(InfoHash key, ValueCallback vcb, Value::Filter f = Value::AllFilter(), Where w = {});
-
-    std::future<size_t> listen(InfoHash key, GetCallback cb, Value::Filter f={}, Where w={}) {
-        return listen(key, [cb](const std::vector<Sp<Value>>& vals, bool expired){
-            if (not expired)
-                return cb(vals);
-            return true;
-        }, std::forward<Value::Filter>(f), std::forward<Where>(w));
-    }
+    std::future<size_t> listen(InfoHash key, GetCallback vcb, Value::Filter f = Value::AllFilter(), Where w = {});
    std::future<size_t> listen(const std::string& key, GetCallback vcb, Value::Filter f = Value::AllFilter(), Where w = {});
    std::future<size_t> listen(InfoHash key, GetCallbackSimple cb, Value::Filter f = Value::AllFilter(), Where w = {}) {
        return listen(key, bindGetCb(cb), f, w);
@@ -315,7 +307,7 @@ public:
     * @param threaded: If false, ::loop() must be called periodically. Otherwise a thread is launched.
     * @param cb: Optional callback to receive general state information.
     */
-    void run(in_port_t port = 4222, const crypto::Identity identity = {}, bool threaded = false, NetId network = 0) {
+    void run(in_port_t port, const crypto::Identity identity, bool threaded = false, NetId network = 0) {
        run(port, {
            /*.dht_config = */{
                /*.node_config = */{
@@ -407,7 +399,12 @@ public:
    /**
     * Insert a push notification to process for OpenDHT
     */
-    void pushNotificationReceived(const std::map<std::string, std::string>& data);
+    void pushNotificationReceived(const std::map<std::string, std::string>& data) const;
+    /**
+     * Refresh a listen via a token
+     * @param token
+     */
+    void resubscribe(unsigned token);

    /* Proxy server mothods */
    void forwardAllMessages(bool forward);
@@ -454,8 +451,15 @@ private:
    /**
     * Store current listeners and translates global tokens for each client.
     */
-    struct Listener;
-    std::map<size_t, Listener> listeners_;
+    struct Listener {
+        size_t tokenClassicDht;
+        size_t tokenProxyDht;
+        GetCallback gcb;
+        InfoHash hash;
+        Value::Filter f;
+        Where w;
+    };
+    std::map<size_t, Listener> listeners_ {};
    size_t listener_token_ {1};

    mutable std::mutex dht_mtx {};
@@ -129,7 +129,7 @@ public:
     * Find the lowest 1 bit in an id.
     * Result will allways be lower than 8*N
     */
-    inline int lowbit() const {
+    inline unsigned lowbit() const {
        int i, j;
        for(i = N-1; i >= 0; i--)
            if(data_[i] != 0)
@@ -1,5 +1,5 @@
 /*
- *  Copyright (C) 2014-2018 Savoir-faire Linux Inc.
+ *  Copyright (C) 2014-2017 Savoir-faire Linux Inc.
 *  Author(s) : Adrien Béraud <adrien.beraud@savoirfairelinux.com>
 *              Simon Désaulniers <simon.desaulniers@savoirfairelinux.com>
 *
@@ -87,8 +87,6 @@ struct RequestAnswer {
    Blob ntoken {};
    Value::Id vid {};
    std::vector<Sp<Value>> values {};
-    std::vector<Value::Id> refreshed_values {};
-    std::vector<Value::Id> expired_values {};
    std::vector<Sp<FieldValueIndex>> fields {};
    std::vector<Sp<Node>> nodes4 {};
    std::vector<Sp<Node>> nodes6 {};
@@ -237,9 +235,6 @@ public:
            std::vector<Sp<Node>>&& nodes, std::vector<Sp<Node>>&& nodes6,
            std::vector<Sp<Value>>&& values, const Query& q);

-    void tellListenerRefreshed(Sp<Node> n, Tid socket_id, const InfoHash& hash, const Blob& ntoken, const std::vector<Value::Id>& values);
-    void tellListenerExpired(Sp<Node> n, Tid socket_id, const InfoHash& hash, const Blob& ntoken, const std::vector<Value::Id>& values);
-
    bool isRunning(sa_family_t af) const;
    inline want_t want () const { return dht_socket >= 0 && dht_socket6 >= 0 ? (WANT4 | WANT6) : -1; }

@@ -1,31 +0,0 @@
-/*
- *  Copyright (C) 2017-2018 Savoir-faire Linux Inc.
- *  Author: Adrien Béraud <adrien.beraud@savoirfairelinux.com>
- *
- *  This program is free software; you can redistribute it and/or modify
- *  it under the terms of the GNU General Public License as published by
- *  the Free Software Foundation; either version 3 of the License, or
- *  (at your option) any later version.
- *
- *  This program is distributed in the hope that it will be useful,
- *  but WITHOUT ANY WARRANTY; without even the implied warranty of
- *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- *  GNU General Public License for more details.
- *
- *  You should have received a copy of the GNU General Public License
- *  along with this program. If not, see <https://www.gnu.org/licenses/>.
- */
-#pragma once
-
-#include <chrono>
-
-namespace dht {
-namespace proxy {
-
-constexpr const std::chrono::minutes OP_TIMEOUT {1 * 10}; // one hour
-constexpr const std::chrono::minutes OP_MARGIN {5}; // 5 minutes
-constexpr const char* const HTTP_PROTO {"http://"};
-using ListenToken = uint64_t;
-
-}
-}
@@ -36,10 +36,11 @@ namespace dht {
 */
 class Scheduler {
 public:
+    Scheduler(const Logger& l) : DHT_LOG(l) {}
+
    struct Job {
        Job(std::function<void()>&& f) : do_(std::move(f)) {}
        std::function<void()> do_;
-        void cancel() { do_ = {}; }
    };

    /**
@@ -57,19 +58,17 @@ public:
        return job;
    }

-    void add(const Sp<Scheduler::Job>& job, time_point t) {
-        if (t != time_point::max())
-            timers.emplace(std::move(t), job);
-    }
-
    /**
     * Reschedules a job.
     *
+     * @param time  The time at which the job shall be rescheduled.
     * @param job  The job to edit.
-     * @param t  The time at which the job shall be rescheduled.
+     *
+     * @return pointer to the newly scheduled job.
     */
    void edit(Sp<Scheduler::Job>& job, time_point t) {
        if (not job) {
+            DHT_LOG.ERR("editing an empty job");
            return;
        }
        // std::function move doesn't garantee to leave the object empty.
@@ -119,6 +118,7 @@ public:
 private:
    time_point now {clock::now()};
    std::multimap<time_point, Sp<Job>> timers {}; /* the jobs ordered by time */
+    const Logger& DHT_LOG;
 };

 }
@@ -100,6 +100,8 @@ public:
        get(key, bindGetCb(cb), bindDoneCb(donecb), std::forward<Value::Filter>(f), std::forward<Where>(w));
    }

+    size_t listen(const InfoHash& id, GetCallback cb, Value::Filter = {}, Where w = {});
+
    /**
     * Will take ownership of the value, sign it using our private key and put it in the DHT.
     */
@@ -278,11 +280,8 @@ public:
    bool cancelPut(const InfoHash& h, const Value::Id& vid) {
        return dht_->cancelPut(h, vid);
    }
-
-    size_t listen(const InfoHash& key, ValueCallback, Value::Filter={}, Where={});
-    size_t listen(const InfoHash& key, GetCallback cb, Value::Filter = {}, Where w = {});
    size_t listen(const InfoHash& key, GetCallbackSimple cb, Value::Filter f={}, Where w = {}) {
-        return listen(key, bindGetCb(cb), f, w);
+        return dht_->listen(key, cb, f, w);
    }
    bool cancelListen(const InfoHash& h, size_t token) {
        return dht_->cancelListen(h, token);
@@ -309,6 +308,13 @@ public:
    void pushNotificationReceived(const std::map<std::string, std::string>& notification) {
        dht_->pushNotificationReceived(notification);
    }
+    /**
+     * Refresh a listen via a token
+     * @param token
+     */
+    void resubscribe(const unsigned token) {
+        dht_->resubscribe(token);
+    }

    void setLoggers(LogMethod error = NOLOG, LogMethod warn = NOLOG, LogMethod debug = NOLOG)
    {
@@ -332,8 +338,6 @@ private:
    SecureDht(const SecureDht&) = delete;
    SecureDht& operator=(const SecureDht&) = delete;

-    Sp<Value> checkValue(const Sp<Value>& v);
-    ValueCallback getCallbackFilter(ValueCallback, Value::Filter&&);
    GetCallback getCallbackFilter(GetCallback, Value::Filter&&);

    Sp<crypto::PrivateKey> key_ {};
@@ -346,6 +350,8 @@ private:
    std::map<InfoHash, Sp<crypto::Certificate>> nodesCertificates_ {};
    std::map<InfoHash, Sp<const crypto::PublicKey>> nodesPubKeys_ {};

+    std::uniform_int_distribution<Value::Id> rand_id {};
+
    std::atomic_bool forward_all_ {false};
 };

@@ -54,12 +54,6 @@ void erase_if(std::map<Key, Item>& map, const Condition& condition)
    }
 }

-/**
- * Split "[host]:port" or "host:port" to pair<"host", "port">.
- */
-OPENDHT_PUBLIC std::pair<std::string, std::string>
-splitPort(const std::string& s);
-
 class OPENDHT_PUBLIC DhtException : public std::runtime_error {
 public:
    DhtException(const std::string &str = "") :
@@ -109,19 +109,6 @@ struct OPENDHT_PUBLIC ValueType {
    EditPolicy editPolicy {DEFAULT_EDIT_POLICY};
 };

-class TypeStore {
-public:
-    void registerType(const ValueType& type) {
-        types[type.id] = type;
-    }
-    const ValueType& getType(ValueType::Id type_id) const {
-        const auto& t_it = types.find(type_id);
-        return (t_it == types.end()) ? ValueType::USER_DATA : t_it->second;
-    }
-private:
-    std::map<ValueType::Id, ValueType> types {};
-};
-
 /**
 * A "value" is data potentially stored on the Dht, with some metadata.
 *
@@ -187,15 +174,6 @@ struct OPENDHT_PUBLIC Value
                return f1(v) or f2(v);
            };
        }
-        std::vector<Sp<Value>> filter(const std::vector<Sp<Value>>& values) {
-            if (not (*this))
-                return values;
-            std::vector<Sp<Value>> ret;
-            for (const auto& v : values)
-                if ((*this)(v))
-                    ret.emplace_back(v);
-            return ret;
-        }
    };

    /* Sneaky functions disguised in classes */
@@ -842,10 +820,6 @@ struct OPENDHT_PUBLIC Where
        return ss.str();
    }

-    bool empty() const {
-        return filters_.empty();
-    }
-
    OPENDHT_PUBLIC friend std::ostream& operator<<(std::ostream& s, const dht::Where& q);

 private:
@@ -996,10 +970,6 @@ unpackVector(const std::vector<std::shared_ptr<Value>>& vals) {
    return ret;
 }

-#if OPENDHT_PROXY_SERVER || OPENDHT_PROXY_CLIENT
-uint64_t unpackId(const Json::Value& json, const std::string& key);
-#endif
-
 }

 MSGPACK_ADD_ENUM(dht::Value::Field)
@@ -9,9 +9,6 @@ libopendht_la_SOURCES  = \
        listener.h \
        request.h \
        search.h \
-        value_cache.h \
-        op_cache.h \
-        op_cache.cpp \
        net.h \
        parsed_message.h \
        node_cache.cpp \
@@ -56,7 +53,7 @@ nobase_include_HEADERS = \
        ../include/opendht/rng.h

 if ENABLE_PROXY_SERVER
-libopendht_la_SOURCES += dht_proxy_server.cpp
+libopendht_la_SOURCES += base64.h base64.cpp dht_proxy_server.cpp
 nobase_include_HEADERS += ../include/opendht/dht_proxy_server.h
 endif

@@ -65,11 +62,6 @@ libopendht_la_SOURCES += dht_proxy_client.cpp
 nobase_include_HEADERS += ../include/opendht/dht_proxy_client.h ../include/opendht/dht_interface.h
 endif

-if PROXY_CLIENT_OR_SERVER
-libopendht_la_SOURCES += base64.h base64.cpp
-nobase_include_HEADERS += ../include/opendht/proxy.h
-endif
-
 if ENABLE_INDEXATION
 libopendht_la_SOURCES += indexation/pht.cpp
 nobase_include_HEADERS += ../include/opendht/indexation/pht.h
@@ -411,8 +411,6 @@ PublicKey::operator=(PublicKey&& o) noexcept
 void
 PublicKey::pack(Blob& b) const
 {
-    if (not pk)
-        throw CryptoException(std::string("Could not export public key: null key"));
    std::vector<uint8_t> tmp(2048);
    size_t sz = tmp.size();
    int err = gnutls_pubkey_export(pk, GNUTLS_X509_FMT_DER, tmp.data(), &sz);
@@ -439,8 +437,6 @@ PublicKey::unpack(const uint8_t* data, size_t data_size)
 std::string
 PublicKey::toString() const
 {
-    if (not pk)
-        throw CryptoException(std::string("Could not print public key: null key"));
    std::string ret;
    size_t sz = ret.size();
    int err = gnutls_pubkey_export(pk, GNUTLS_X509_FMT_PEM, (void*)ret.data(), &sz);
@@ -180,7 +180,7 @@ Dht::expireBuckets(RoutingTable& list)
 {
    for (auto& b : list) {
        bool changed = false;
-        b.nodes.remove_if([&changed](const Sp<Node>& n) {
+        b.nodes.remove_if([this,&changed](const Sp<Node>& n) {
            if (n->isExpired()) {
                changed = true;
                return true;
@@ -387,7 +387,7 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr) {
        if (not n.isSynced(now))
            continue;
        if (std::find_if(sr->announce.cbegin(), sr->announce.cend(),
-            [&now,&n](const Announce& a) {
+            [this,&now,&n](const Announce& a) {
                return n.getAnnounceTime(a.value->id) <= now;
            }) == sr->announce.cend())
            continue;
@@ -482,72 +482,6 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr) {
    }
 }

-void
-Dht::searchSynchedNodeListen(const Sp<Search>& sr, SearchNode& n)
-{
-    std::weak_ptr<Search> ws = sr;
-    for (const auto& l : sr->listeners) {
-        const auto& query = l.second.query;
-        auto list_token = l.first;
-        if (n.getListenTime(query) > scheduler.time())
-            continue;
-        DHT_LOG.w(sr->id, n.node->id, "[search %s] [node %s] sending 'listen'",
-                sr->id.toString().c_str(), n.node->toString().c_str());
-
-        auto r = n.listenStatus.find(query);
-        if (r == n.listenStatus.end()) {
-            r = n.listenStatus.emplace(query, SearchNode::CachedListenStatus{
-                [ws,list_token](const std::vector<Sp<Value>>& values, bool expired){
-                    if (auto sr = ws.lock()) {
-                        auto l = sr->listeners.find(list_token);
-                        if (l != sr->listeners.end()) {
-                            l->second.get_cb(l->second.filter.filter(values), expired);
-                        }
-                    }
-                }
-            }).first;
-            auto node = n.node;
-            r->second.cacheExpirationJob = scheduler.add(time_point::max(), [this,ws,query,node]{
-                if (auto sr = ws.lock()) {
-                    if (auto sn = sr->getNode(node)) {
-                        sn->expireValues(query, scheduler);
-                    }
-                }
-            });
-        }
-        auto prev_req = r != n.listenStatus.end() ? r->second.req : nullptr;
-        r->second.req = network_engine.sendListen(n.node, sr->id, *query, n.token, prev_req,
-            [this,ws,query](const net::Request& req, net::RequestAnswer&& answer) mutable
-            { /* on done */
-                if (auto sr = ws.lock()) {
-                    scheduler.edit(sr->nextSearchStep, scheduler.time());
-                    if (auto sn = sr->getNode(req.node))
-                        scheduler.add(sn->getListenTime(query), std::bind(&Dht::searchStep, this, sr));
-                    onListenDone(req.node, answer, sr);
-                }
-            },
-            [this,ws,query](const net::Request& req, bool over) mutable
-            { /* on request expired */
-                if (auto sr = ws.lock()) {
-                    scheduler.edit(sr->nextSearchStep, scheduler.time());
-                    if (over)
-                        if (auto sn = sr->getNode(req.node))
-                            sn->listenStatus.erase(query);
-                }
-            },
-            [this,ws,query](const Sp<Node>& node, net::RequestAnswer&& answer) mutable
-            { /* on new values */
-                if (auto sr = ws.lock()) {
-                    scheduler.edit(sr->nextSearchStep, scheduler.time());
-                    if (auto sn = sr->getNode(node)) {
-                        sn->onValues(query, std::move(answer), types, scheduler);
-                    }
-                }
-            }
-        );
-    }
-}
-
 /* When a search is in progress, we periodically call search_step to send
   further requests. */
 void
@@ -604,7 +538,45 @@ Dht::searchStep(Sp<Search> sr)
            for (auto& n : sr->nodes) {
                if (not n.isSynced(now))
                    continue;
-                searchSynchedNodeListen(sr, n);
+                for (const auto& l : sr->listeners) {
+                    const auto& query = l.second.query;
+                    if (n.getListenTime(query) > now)
+                        continue;
+                    DHT_LOG.w(sr->id, n.node->id, "[search %s] [node %s] sending 'listen'",
+                            sr->id.toString().c_str(), n.node->toString().c_str());
+
+                    const auto& r = n.listenStatus.find(query);
+                    auto prev_req = r != n.listenStatus.end() ? r->second : nullptr;
+
+                    std::weak_ptr<Search> ws = sr;
+                    n.listenStatus[query] = network_engine.sendListen(n.node, sr->id, *query, n.token, prev_req,
+                        [this,ws,query](const net::Request& req, net::RequestAnswer&& answer) mutable
+                        { /* on done */
+                            if (auto sr = ws.lock()) {
+                                scheduler.edit(sr->nextSearchStep, scheduler.time());
+                                if (auto sn = sr->getNode(req.node))
+                                    scheduler.add(sn->getListenTime(query), std::bind(&Dht::searchStep, this, sr));
+                                onListenDone(req.node, answer, sr);
+                            }
+                        },
+                        [this,ws,query](const net::Request& req, bool over) mutable
+                        { /* on expired */
+                            if (auto sr = ws.lock()) {
+                                scheduler.edit(sr->nextSearchStep, scheduler.time());
+                                if (over)
+                                    if (auto sn = sr->getNode(req.node))
+                                        sn->listenStatus.erase(query);
+                            }
+                        },
+                        [this,ws,query](const Sp<Node>& node, net::RequestAnswer&& answer) mutable
+                        { /* on new values */
+                            if (auto sr = ws.lock()) {
+                                scheduler.edit(sr->nextSearchStep, scheduler.time());
+                                onGetValuesDone(node, answer, sr, query);
+                            }
+                        }
+                    );
+                }
                if (not n.candidate and ++i == LISTEN_NODES)
                    break;
            }
@@ -800,8 +772,9 @@ Dht::announce(const InfoHash& id,
 }

 size_t
-Dht::listenTo(const InfoHash& id, sa_family_t af, ValueCallback cb, Value::Filter f, const Sp<Query>& q)
+Dht::listenTo(const InfoHash& id, sa_family_t af, GetCallback cb, Value::Filter f, const Sp<Query>& q)
 {
+    const auto& now = scheduler.time();
    if (!isRunning(af))
        return 0;
       // DHT_LOG_ERR("[search %s IPv%c] search_time is now in %lfs", sr->id.toString().c_str(), (sr->af == AF_INET) ? '4' : '6', print_dt(tm-clock::now()));
@@ -813,11 +786,15 @@ Dht::listenTo(const InfoHash& id, sa_family_t af, ValueCallback cb, Value::Filte
    if (!sr)
        throw DhtException("Can't create search");
    DHT_LOG.e(id, "[search %s IPv%c] listen", id.toString().c_str(), (af == AF_INET) ? '4' : '6');
-    return sr->listen(cb, f, q, scheduler);
+    sr->done = false;
+    auto token = ++sr->listener_token;
+    sr->listeners.emplace(token, LocalListener{q, f, cb});
+    scheduler.edit(sr->nextSearchStep, now);
+    return token;
 }

 size_t
-Dht::listen(const InfoHash& id, ValueCallback cb, Value::Filter f, Where where)
+Dht::listen(const InfoHash& id, GetCallback cb, Value::Filter f, Where where)
 {
    scheduler.syncTime();

@@ -825,7 +802,26 @@ Dht::listen(const InfoHash& id, ValueCallback cb, Value::Filter f, Where where)
    auto vals = std::make_shared<std::map<Value::Id, Sp<Value>>>();
    auto token = ++listener_token;

-    auto gcb = OpValueCache::cacheCallback(std::move(cb));
+    auto gcb = [=](const std::vector<Sp<Value>>& values) {
+        std::vector<Sp<Value>> newvals;
+        for (const auto& v : values) {
+            auto it = vals->find(v->id);
+            if (it == vals->cend() || !(*it->second == *v))
+                newvals.push_back(v);
+        }
+        if (!newvals.empty()) {
+            if (!cb(newvals)) {
+                cancelListen(id, token);
+                return false;
+            }
+            for (const auto& v : newvals) {
+                auto it = vals->emplace(v->id, v);
+                if (not it.second)
+                    it.first->second = v;
+            }
+        }
+        return true;
+    };

    auto query = std::make_shared<Query>(q);
    auto filter = f.chain(q.where.getFilter());
@@ -837,7 +833,7 @@ Dht::listen(const InfoHash& id, ValueCallback cb, Value::Filter f, Where where)
        if (not st->second.empty()) {
            std::vector<Sp<Value>> newvals = st->second.get(filter);
            if (not newvals.empty()) {
-                if (!gcb(newvals, false))
+                if (!cb(newvals))
                    return 0;
                for (const auto& v : newvals) {
                    auto it = vals->emplace(v->id, v);
@@ -874,10 +870,10 @@ Dht::cancelListen(const InfoHash& id, size_t token)
    if (st != store.end() && tokenlocal)
        st->second.local_listeners.erase(tokenlocal);

-    auto searches_cancel_listen = [this,&id](std::map<InfoHash, Sp<Search>>& srs, size_t token) {
+    auto searches_cancel_listen = [&id](std::map<InfoHash, Sp<Search>>& srs, size_t token) {
        auto srp = srs.find(id);
        if (srp != srs.end() and token)
-            srp->second->cancelListen(token, scheduler);
+            srp->second->cancelListen(token);
    };
    searches_cancel_listen(searches4, std::get<1>(it->second));
    searches_cancel_listen(searches6, std::get<2>(it->second));
@@ -1139,27 +1135,25 @@ Dht::cancelPut(const InfoHash& id, const Value::Id& vid)
 // Storage

 void
-Dht::storageChanged(const InfoHash& id, Storage& st, ValueStorage& v, bool newValue)
+Dht::storageChanged(const InfoHash& id, Storage& st, ValueStorage& v)
 {
-    if (newValue) {
-        if (not st.local_listeners.empty()) {
-            DHT_LOG.d(id, "[store %s] %lu local listeners", id.toString().c_str(), st.local_listeners.size());
-            std::vector<std::pair<ValueCallback, std::vector<Sp<Value>>>> cbs;
-            for (const auto& l : st.local_listeners) {
-                std::vector<Sp<Value>> vals;
-                if (not l.second.filter or l.second.filter(*v.data))
-                    vals.push_back(v.data);
-                if (not vals.empty()) {
-                    DHT_LOG.d(id, "[store %s] sending update local listener with token %lu",
-                            id.toString().c_str(),
-                            l.first);
-                    cbs.emplace_back(l.second.get_cb, std::move(vals));
-                }
+    if (not st.local_listeners.empty()) {
+        DHT_LOG.d(id, "[store %s] %lu local listeners", id.toString().c_str(), st.local_listeners.size());
+        std::vector<std::pair<GetCallback, std::vector<Sp<Value>>>> cbs;
+        for (const auto& l : st.local_listeners) {
+            std::vector<Sp<Value>> vals;
+            if (not l.second.filter or l.second.filter(*v.data))
+                vals.push_back(v.data);
+            if (not vals.empty()) {
+                DHT_LOG.d(id, "[store %s] sending update local listener with token %lu",
+                        id.toString().c_str(),
+                        l.first);
+                cbs.emplace_back(l.second.get_cb, std::move(vals));
            }
-            // listeners are copied: they may be deleted by the callback
-            for (auto& cb : cbs)
-                cb.first(cb.second, false);
        }
+        // listeners are copied: they may be deleted by the callback
+        for (auto& cb : cbs)
+            cb.first(cb.second);
    }

    if (not st.listeners.empty()) {
@@ -1213,7 +1207,7 @@ Dht::storageStore(const InfoHash& id, const Sp<Value>& value, time_point created
        if (total_store_size > max_store_size) {
            expireStore();
        }
-        storageChanged(id, st->second, *vs, store.second.values_diff > 0);
+        storageChanged(id, st->second, *vs);
    }

    return std::get<0>(store);
@@ -1247,36 +1241,11 @@ Dht::storageAddListener(const InfoHash& id, const Sp<Node>& node, size_t socket_
 void
 Dht::expireStore(decltype(store)::iterator i)
 {
-    const auto& id = i->first;
-    auto& st = i->second;
-    auto stats = st.expire(id, scheduler.time());
-    total_store_size += stats.first;
-    total_values -= stats.second.size();
-    if (not stats.second.empty()) {
-        DHT_LOG.d(id, "[store %s] discarded %ld expired values (%ld bytes)",
-            id.toString().c_str(), stats.second.size(), -stats.first);
-
-        if (not st.listeners.empty()) {
-            DHT_LOG.d(id, "[store %s] %lu remote listeners", id.toString().c_str(), st.listeners.size());
-
-            std::vector<Value::Id> ids;
-            ids.reserve(stats.second.size());
-            for (const auto& v : stats.second)
-                ids.emplace_back(v->id);
-
-            for (const auto& node_listeners : st.listeners) {
-                for (const auto& l : node_listeners.second) {
-                    DHT_LOG.w(id, node_listeners.first->id, "[store %s] [node %s] sending expired",
-                            id.toString().c_str(),
-                            node_listeners.first->toString().c_str());
-                    Blob ntoken = makeToken(node_listeners.first->getAddr(), false);
-                    network_engine.tellListenerExpired(node_listeners.first, l.first, id, ntoken, ids);
-                }
-            }
-        }
-        for (const auto& local_listeners : st.local_listeners) {
-            local_listeners.second.get_cb(stats.second, true);
-        }
+    auto stats = i->second.expire(i->first, scheduler.time());
+    total_store_size += stats.size_diff;
+    total_values += stats.values_diff;
+    if (stats.values_diff) {
+        DHT_LOG.d(i->first, "[store %s] discarded %ld expired values (%ld bytes)", i->first.toString().c_str(), -stats.values_diff, -stats.size_diff);
    }
 }

@@ -1349,7 +1318,7 @@ Dht::connectivityChanged(sa_family_t af)
    for (auto& sp : searches(af))
        for (auto& sn : sp.second->nodes) {
            for (auto& ls : sn.listenStatus)
-                sn.node->cancelRequest(ls.second.req);
+                sn.node->cancelRequest(ls.second);
            sn.listenStatus.clear();
        }
    reported_addr.erase(std::remove_if(reported_addr.begin(), reported_addr.end(), [&](const ReportedAddr& addr){
@@ -1684,12 +1653,13 @@ Dht::~Dht()
        s.second->clear();
 }

-Dht::Dht() : store(), network_engine(DHT_LOG, scheduler) {}
+Dht::Dht() : store(), scheduler(DHT_LOG), network_engine(DHT_LOG, scheduler) {}

 Dht::Dht(const int& s, const int& s6, Config config)
    : myid(config.node_id ? config.node_id : InfoHash::getRandom()),
    is_bootstrap(config.is_bootstrap),
    maintain_storage(config.maintain_storage), store(), store_quota(),
+    scheduler(DHT_LOG),
    network_engine(myid, config.network, s, s6, DHT_LOG, scheduler,
            std::bind(&Dht::onError, this, _1, _2),
            std::bind(&Dht::onNewNode, this, _1, _2),
@@ -1901,7 +1871,7 @@ Dht::periodic(const uint8_t *buf, size_t buflen, const SockAddr& from)
        try {
            network_engine.processMessage(buf, buflen, from);
        } catch (const std::exception& e) {
-            DHT_LOG.e("Can't process message from %s: %s", from.toString().c_str(), e.what());
+            DHT_LOG.e("Can't parse message from %s: %s", from.toString().c_str(), e.what());
        }
    }
    return scheduler.run();
@@ -2198,7 +2168,7 @@ void Dht::onGetValuesDone(const Sp<Node>& node,
            }

            /* callbacks for local search listeners */
-            /*std::vector<std::pair<ValueCallback, std::vector<Sp<Value>>>> tmp_lists;
+            std::vector<std::pair<GetCallback, std::vector<Sp<Value>>>> tmp_lists;
            for (auto& l : sr->listeners) {
                if (!l.second.get_cb or (orig_query and l.second.query and not l.second.query->isSatisfiedBy(*orig_query)))
                    continue;
@@ -2210,10 +2180,7 @@ void Dht::onGetValuesDone(const Sp<Node>& node,
                    tmp_lists.emplace_back(l.second.get_cb, std::move(tmp));
            }
            for (auto& l : tmp_lists)
-                l.first(l.second, false);*/
-        } else if (not a.expired_values.empty()) {
-            DHT_LOG.w(sr->id, node->id, "[search %s] [node %s] %u expired values",
-                      sr->id.toString().c_str(), node->toString().c_str(), a.expired_values.size());
+                l.first(l.second);
        }
    } else {
        DHT_LOG.w(sr->id, "[node %s] no token provided. Ignoring response content.", node->toString().c_str());
@@ -43,15 +43,6 @@ namespace dht {

 constexpr std::chrono::seconds DhtRunner::BOOTSTRAP_PERIOD;

-struct DhtRunner::Listener {
-    size_t tokenClassicDht;
-    size_t tokenProxyDht;
-    ValueCallback gcb;
-    InfoHash hash;
-    Value::Filter f;
-    Where w;
-};
-
 DhtRunner::DhtRunner() : dht_()
 #if OPENDHT_PROXY_CLIENT
 , dht_via_proxy_()
@@ -120,8 +111,7 @@ DhtRunner::run(const SockAddr& local4, const SockAddr& local6, DhtRunner::Config
            } catch (const dht::SocketException& e) {
                startNetwork(local4, local6);
            }
-
-            auto hasJobToDo = [this]() {
+            cv.wait_until(lk, wakeup, [this]() {
                if (not running)
                    return true;
                {
@@ -138,11 +128,7 @@ DhtRunner::run(const SockAddr& local4, const SockAddr& local6, DhtRunner::Config
                        return true;
                }
                return false;
-            };
-            if (wakeup == time_point::max())
-                cv.wait(lk, hasJobToDo);
-            else
-                cv.wait_until(lk, wakeup, hasJobToDo);
+            });
        }
    });
 }
@@ -556,7 +542,7 @@ DhtRunner::query(const InfoHash& hash, QueryCallback cb, DoneCallback done_cb, Q
 }

 std::future<size_t>
-DhtRunner::listen(InfoHash hash, ValueCallback vcb, Value::Filter f, Where w)
+DhtRunner::listen(InfoHash hash, GetCallback vcb, Value::Filter f, Where w)
 {
    auto ret_token = std::make_shared<std::promise<size_t>>();
    {
@@ -567,8 +553,8 @@ DhtRunner::listen(InfoHash hash, ValueCallback vcb, Value::Filter f, Where w)
            listener.hash = hash;
            listener.f = std::move(f);
            listener.w = std::move(w);
-            listener.gcb = [hash,vcb,tokenbGlobal,this](const std::vector<Sp<Value>>& vals, bool expired){
-                if (not vcb(vals, expired)) {
+            listener.gcb = [hash,vcb,tokenbGlobal,this](const std::vector<Sp<Value>>& vals){
+                if (not vcb(vals)) {
 #if OPENDHT_PROXY_CLIENT
                    cancelListen(hash, tokenbGlobal);
 #endif
@@ -600,8 +586,8 @@ void
 DhtRunner::cancelListen(InfoHash h, size_t token)
 {
    std::lock_guard<std::mutex> lck(storage_mtx);
-    pending_ops.emplace([=](SecureDht& dht) {
 #if OPENDHT_PROXY_CLIENT
+    pending_ops.emplace([=](SecureDht&) {
        auto it = listeners_.find(token);
        if (it == listeners_.end()) return;
        if (it->second.tokenClassicDht)
@@ -609,31 +595,23 @@ DhtRunner::cancelListen(InfoHash h, size_t token)
        if (it->second.tokenProxyDht and dht_via_proxy_)
            dht_via_proxy_->cancelListen(h, it->second.tokenProxyDht);
        listeners_.erase(it);
-#else
-        dht.cancelListen(h, token);
-#endif // OPENDHT_PROXY_CLIENT
    });
+#else
+    pending_ops.emplace([=](SecureDht& dht) {
+        dht.cancelListen(h, token);
+    });
+#endif // OPENDHT_PROXY_CLIENT
    cv.notify_all();
 }

 void
-DhtRunner::cancelListen(InfoHash h, std::shared_future<size_t> ftoken)
+DhtRunner::cancelListen(InfoHash h, std::shared_future<size_t> token)
 {
    {
        std::lock_guard<std::mutex> lck(storage_mtx);
        pending_ops.emplace([=](SecureDht& dht) {
-            auto token = ftoken.get();
-#if OPENDHT_PROXY_CLIENT
-            auto it = listeners_.find(token);
-            if (it == listeners_.end()) return;
-            if (it->second.tokenClassicDht)
-                dht_->cancelListen(h, it->second.tokenClassicDht);
-            if (it->second.tokenProxyDht and dht_via_proxy_)
-                dht_via_proxy_->cancelListen(h, it->second.tokenProxyDht);
-            listeners_.erase(it);
-#else
-            dht.cancelListen(h, token);
-#endif // OPENDHT_PROXY_CLIENT
+            auto tk = token.get();
+            dht.cancelListen(h, tk);
        });
    }
    cv.notify_all();
@@ -966,17 +944,20 @@ DhtRunner::setPushNotificationToken(const std::string& token) {
 }

 void
-DhtRunner::pushNotificationReceived(const std::map<std::string, std::string>& data)
+DhtRunner::pushNotificationReceived(const std::map<std::string, std::string>& data) const
 {
 #if OPENDHT_PROXY_CLIENT && OPENDHT_PUSH_NOTIFICATIONS
-    {
-        std::lock_guard<std::mutex> lck(storage_mtx);
-        pending_ops_prio.emplace([=](SecureDht& dht) {
-            if (dht_via_proxy_)
-                dht_via_proxy_->pushNotificationReceived(data);
-        });
-    }
-    cv.notify_all();
+    if (dht_via_proxy_)
+        dht_via_proxy_->pushNotificationReceived(data);
+#endif
+}
+
+void
+DhtRunner::resubscribe(unsigned token)
+{
+#if OPENDHT_PROXY_CLIENT && OPENDHT_PUSH_NOTIFICATIONS
+    if (dht_via_proxy_)
+        dht_via_proxy_->resubscribe(token);
 #endif
 }

@@ -20,7 +20,6 @@

 #include "value.h"
 #include "utils.h"
-#include "callbacks.h"

 namespace dht {

@@ -45,14 +44,7 @@ struct Listener {
 struct LocalListener {
    Sp<Query> query;
    Value::Filter filter;
-    ValueCallback get_cb;
-};
-
-
-struct SearchListener {
-    Sp<Query> query;
-    Value::Filter filter;
-    ValueCallback get_cb;
+    GetCallback get_cb;
 };

 }
@@ -122,25 +122,12 @@ serializeValues(const std::vector<Sp<Value>>& st)
    return svals;
 }

-void
-packToken(msgpack::packer<msgpack::sbuffer>& pk, const Blob& token)
-{
-    pk.pack_bin(token.size());
-    pk.pack_bin_body((char*)token.data(), token.size());
-}
-
 RequestAnswer::RequestAnswer(ParsedMessage&& msg)
- : ntoken(std::move(msg.token)),
-   values(std::move(msg.values)),
-   refreshed_values(std::move(msg.refreshed_values)),
-   expired_values(std::move(msg.expired_values)),
-   fields(std::move(msg.fields)),
-   nodes4(std::move(msg.nodes4)),
-   nodes6(std::move(msg.nodes6))
-{}
+ : ntoken(std::move(msg.token)), values(std::move(msg.values)), fields(std::move(msg.fields)),
+    nodes4(std::move(msg.nodes4)), nodes6(std::move(msg.nodes6)) {}

 NetworkEngine::NetworkEngine(Logger& log, Scheduler& scheduler, const int& s, const int& s6)
-    : myid(zeroes), dht_socket(s), dht_socket6(s6), DHT_LOG(log), scheduler(scheduler)
+    : myid(zeroes), DHT_LOG(log), scheduler(scheduler), dht_socket(s), dht_socket6(s6)
 {}
 NetworkEngine::NetworkEngine(InfoHash& myid, NetId net, const int& s, const int& s6, Logger& log, Scheduler& scheduler,
        decltype(NetworkEngine::onError) onError,
@@ -184,67 +171,6 @@ NetworkEngine::tellListener(Sp<Node> node, Tid socket_id, const InfoHash& hash,
    }
 }

-void
-NetworkEngine::tellListenerRefreshed(Sp<Node> n, Tid socket_id, const InfoHash& hash, const Blob& token, const std::vector<Value::Id>& values)
-{
-    msgpack::sbuffer buffer;
-    msgpack::packer<msgpack::sbuffer> pk(&buffer);
-    pk.pack_map(4+(network?1:0));
-
-    pk.pack(std::string("u"));
-    pk.pack_map(1 + (not values.empty()?1:0) + (not token.empty()?1:0));
-    pk.pack(std::string("id")); pk.pack(myid);
-    if (not token.empty()) {
-        pk.pack(std::string("token")); packToken(pk, token);
-    }
-    if (not values.empty()) {
-        pk.pack(std::string("re"));
-        pk.pack(values);
-        DHT_LOG.d(n->id, "[node %s] sending %zu refreshed values", n->toString().c_str(), values.size());
-    }
-
-    pk.pack(std::string("t")); pk.pack(socket_id);
-    pk.pack(std::string("y")); pk.pack(std::string("r"));
-    pk.pack(std::string("v")); pk.pack(my_v);
-    if (network) {
-        pk.pack(std::string("n")); pk.pack(network);
-    }
-
-    // send response
-    send(buffer.data(), buffer.size(), 0, n->getAddr());
-}
-
-void
-NetworkEngine::tellListenerExpired(Sp<Node> n, Tid socket_id, const InfoHash& hash, const Blob& token, const std::vector<Value::Id>& values)
-{
-    msgpack::sbuffer buffer;
-    msgpack::packer<msgpack::sbuffer> pk(&buffer);
-    pk.pack_map(4+(network?1:0));
-
-    pk.pack(std::string("u"));
-    pk.pack_map(1 + (not values.empty()?1:0) + (not token.empty()?1:0));
-    pk.pack(std::string("id")); pk.pack(myid);
-    if (not token.empty()) {
-        pk.pack(std::string("token")); packToken(pk, token);
-    }
-    if (not values.empty()) {
-        pk.pack(std::string("exp"));
-        pk.pack(values);
-        DHT_LOG.d(n->id, "[node %s] sending %zu expired values", n->toString().c_str(), values.size());
-    }
-
-    pk.pack(std::string("t")); pk.pack(socket_id);
-    pk.pack(std::string("y")); pk.pack(std::string("r"));
-    pk.pack(std::string("v")); pk.pack(my_v);
-    if (network) {
-        pk.pack(std::string("n")); pk.pack(network);
-    }
-
-    // send response
-    send(buffer.data(), buffer.size(), 0, n->getAddr());
-}
-
-
 bool
 NetworkEngine::isRunning(sa_family_t af) const
 {
@@ -292,24 +218,20 @@ NetworkEngine::requestStep(Sp<Request> sreq)
    auto err = send((char*)req.msg.data(), req.msg.size(),
            (node.getReplyTime() >= now - UDP_REPLY_TIME) ? 0 : MSG_CONFIRM,
            node.getAddr());
-    std::weak_ptr<Request> wreq = sreq;
    if (err == ENETUNREACH  ||
        err == EHOSTUNREACH ||
        err == EAFNOSUPPORT ||
        err == EPIPE)
    {
-        scheduler.add(now, [this,wreq]() mutable {
-            if (auto req = wreq.lock()) {
-                req->node->setExpired();
-                if (not req->node->id)
-                    requests.erase(req->tid);
-            }
-        });
+        node.setExpired();
+        if (not node.id)
+            requests.erase(req.tid);
    } else {
        if (err != EAGAIN) {
            ++req.attempt_count;
        }
        req.last_try = now;
+        std::weak_ptr<Request> wreq = sreq;
        scheduler.add(req.last_try + Node::MAX_RESPONSE_TIME, [this,wreq] {
            if (auto req = wreq.lock())
                requestStep(req);
@@ -415,7 +337,7 @@ NetworkEngine::processMessage(const uint8_t *buf, size_t buflen, const SockAddr&
        msgpack::unpacked msg_res = msgpack::unpack((const char*)buf, buflen);
        msg->msgpack_unpack(msg_res.get());
    } catch (const std::exception& e) {
-        DHT_LOG.w("Can't parse message of size %lu: %s", buflen, e.what());
+        DHT_LOG.w("Can't process message of size %lu: %s", buflen, e.what());
        DHT_LOG.DEBUG.logPrintable(buf, buflen);
        return;
    }
@@ -614,7 +536,6 @@ NetworkEngine::process(std::unique_ptr<ParsedMessage>&& msg, const SockAddr& fro
                DHT_LOG.d(msg->info_hash, node->id, "[node %s] got 'listen' request for %s", node->toString().c_str(), msg->info_hash.toString().c_str());
                ++in_stats.listen;
                RequestAnswer answer = onListen(node, msg->info_hash, msg->token, msg->socket_id, std::move(msg->query));
-                auto nnodes = bufferNodes(from.getFamily(), msg->info_hash, msg->want, answer.nodes4, answer.nodes6);
                sendListenConfirmation(from, msg->tid);
                break;
            }
@@ -629,6 +550,13 @@ NetworkEngine::process(std::unique_ptr<ParsedMessage>&& msg, const SockAddr& fro
    }
 }

+void
+packToken(msgpack::packer<msgpack::sbuffer>& pk, const Blob& token)
+{
+    pk.pack_bin(token.size());
+    pk.pack_bin_body((char*)token.data(), token.size());
+}
+
 void
 insertAddr(msgpack::packer<msgpack::sbuffer>& pk, const SockAddr& addr)
 {
@@ -1,204 +0,0 @@
-/*
- *  Copyright (C) 2018 Savoir-faire Linux Inc.
- *  Author(s) : Adrien Béraud <adrien.beraud@savoirfairelinux.com>
- *
- *  This program is free software; you can redistribute it and/or modify
- *  it under the terms of the GNU General Public License as published by
- *  the Free Software Foundation; either version 3 of the License, or
- *  (at your option) any later version.
- *
- *  This program is distributed in the hope that it will be useful,
- *  but WITHOUT ANY WARRANTY; without even the implied warranty of
- *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- *  GNU General Public License for more details.
- *
- *  You should have received a copy of the GNU General Public License
- *  along with this program. If not, see <https://www.gnu.org/licenses/>.
- */
-#include "op_cache.h"
-
-namespace dht {
-
-constexpr const std::chrono::seconds OpCache::EXPIRATION;
-
-bool
-OpValueCache::onValuesAdded(const std::vector<Sp<Value>>& vals) {
-    std::vector<Sp<Value>> newValues;
-    for (const auto& v : vals) {
-        auto viop = values.emplace(v->id, OpCacheValueStorage{v});
-        if (viop.second) {
-            newValues.emplace_back(v);
-        } else {
-            viop.first->second.refCount++;
-        }
-    }
-    return newValues.empty() ? true : callback(newValues, false);
-}
-bool
-OpValueCache::onValuesExpired(const std::vector<Sp<Value>>& vals) {
-    std::vector<Sp<Value>> expiredValues;
-    for (const auto& v : vals) {
-        auto vit = values.find(v->id);
-        if (vit != values.end()) {
-            vit->second.refCount--;
-            if (not vit->second.refCount) {
-                expiredValues.emplace_back(std::move(vit->second.data));
-                values.erase(vit);
-            }
-        }
-    }
-    return expiredValues.empty() ? true : callback(expiredValues, true);
-}
-std::vector<Sp<Value>>
-OpValueCache::get(Value::Filter& filter) const {
-    std::vector<Sp<Value>> ret;
-    if (not filter)
-        ret.reserve(values.size());
-    for (const auto& v : values)
-        if (not filter or filter(*v.second.data))
-            ret.emplace_back(v.second.data);
-    return ret;
-}
-
-Sp<Value>
-OpValueCache::get(Value::Id id) const {
-    auto v = values.find(id);
-    if (v == values.end())
-        return {};
-    return v->second.data;
-}
-
-std::vector<Sp<Value>>
-OpValueCache::getValues() const {
-    std::vector<Sp<Value>> ret;
-    ret.reserve(values.size());
-    for (const auto& v : values)
-        ret.emplace_back(v.second.data);
-    return ret;
-}
-
-void
-OpCache::onValuesAdded(const std::vector<Sp<Value>>& vals) {
-    if (not listeners.empty()) {
-        std::vector<LocalListener> list;
-        list.reserve(listeners.size());
-        for (const auto& l : listeners)
-            list.emplace_back(l.second);
-        for (auto& l : list)
-            l.get_cb(l.filter.filter(vals), false);
-    }
-}
-
-void
-OpCache::onValuesExpired(const std::vector<Sp<Value>>& vals) {
-    if (not listeners.empty()) {
-        std::vector<LocalListener> list;
-        list.reserve(listeners.size());
-        for (const auto& l : listeners)
-            list.emplace_back(l.second);
-        for (auto& l : list)
-            l.get_cb(l.filter.filter(vals), true);
-    }
-}
-
-time_point
-OpCache::getExpiration() const {
-    if (not listeners.empty())
-        return time_point::max();
-    return lastRemoved + EXPIRATION;
-}
-
-size_t
-SearchCache::listen(ValueCallback get_cb, Sp<Query> q, Value::Filter filter, std::function<size_t(Sp<Query>, ValueCallback)> onListen)
-{
-    // find exact match
-    auto op = ops.find(q);
-    if (op == ops.end()) {
-        // find satisfying query
-        for (auto it = ops.begin(); it != ops.end(); it++) {
-            if (q->isSatisfiedBy(*it->first)) {
-                op = it;
-                break;
-            }
-        }
-    }
-    if (op == ops.end()) {
-        // New query
-        op = ops.emplace(q, std::unique_ptr<OpCache>(new OpCache)).first;
-        auto& cache = *op->second;
-        cache.searchToken = onListen(q, [&](const std::vector<Sp<Value>>& values, bool expired){
-            return cache.onValue(values, expired);
-        });
-    }
-    auto token = nextToken_++;
-    if (nextToken_ == 0)
-        nextToken_++;
-    op->second->addListener(token, get_cb, q, filter);
-    return token;
-}
-
-bool
-SearchCache::cancelListen(size_t gtoken, const time_point& now) {
-    for (auto& op : ops) {
-        if (op.second->removeListener(gtoken, now)) {
-            nextExpiration_ = std::min(nextExpiration_, op.second->getExpiration());
-            return true;
-        }
-    }
-    return false;
-}
-
-void
-SearchCache::cancelAll(std::function<void(size_t)> onCancel) {
-    for (auto& op : ops) {
-        auto cache = std::move(op.second);
-        cache->removeAll();
-        onCancel(cache->searchToken);
-    }
-    ops.clear();
-}
-
-time_point
-SearchCache::expire(const time_point& now, std::function<void(size_t)> onCancel) {
-    nextExpiration_ = time_point::max();
-    auto ret = nextExpiration_;
-    for (auto it = ops.begin(); it != ops.end();) {
-        auto expiration = it->second->getExpiration();
-        if (expiration < now) {
-            auto cache = std::move(it->second);
-            it = ops.erase(it);
-            onCancel(cache->searchToken);
-        } else {
-            nextExpiration_ = std::min(nextExpiration_, expiration);
-            ret = nextExpiration_;
-            ++it;
-        }
-    }
-    return ret;
-}
-
-std::vector<Sp<Value>>
-SearchCache::get(Value::Filter& filter) const {
-    if (ops.size() == 1)
-        return ops.begin()->second->get(filter);
-    std::map<Value::Id, Sp<Value>> c;
-    for (const auto& op : ops) {
-        for (const auto& v : op.second->get(filter))
-            c.emplace(v->id, v);
-    }
-    std::vector<Sp<Value>> ret;
-    ret.reserve(c.size());
-    for (auto& v : c)
-        ret.emplace_back(std::move(v.second));
-    return ret;
-}
-
-Sp<Value>
-SearchCache::get(Value::Id id) const {
-    for (const auto& op : ops)
-        if (auto v = op.second->get(id))
-            return v;
-    return {};
-}
-
-}
@@ -1,155 +0,0 @@
-/*
- *  Copyright (C) 2018 Savoir-faire Linux Inc.
- *  Author(s) : Adrien Béraud <adrien.beraud@savoirfairelinux.com>
- *
- *  This program is free software; you can redistribute it and/or modify
- *  it under the terms of the GNU General Public License as published by
- *  the Free Software Foundation; either version 3 of the License, or
- *  (at your option) any later version.
- *
- *  This program is distributed in the hope that it will be useful,
- *  but WITHOUT ANY WARRANTY; without even the implied warranty of
- *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- *  GNU General Public License for more details.
- *
- *  You should have received a copy of the GNU General Public License
- *  along with this program. If not, see <https://www.gnu.org/licenses/>.
- */
-#pragma once
-
-#include "value.h"
-#include "value_cache.h"
-#include "listener.h"
-
-namespace dht {
-
-struct OpCacheValueStorage
-{
-    Sp<Value> data {};
-    unsigned refCount {1};
-    OpCacheValueStorage(Sp<Value> val = {}) : data(val) {}
-};
-
-class OpValueCache {
-public:
-    OpValueCache(ValueCallback&& cb) : callback(std::forward<ValueCallback>(cb)) {}
-    OpValueCache(OpValueCache&& o) : values(std::move(o.values)), callback(std::move(o.callback)) {
-        o.callback = {};
-    }
-
-    static ValueCallback cacheCallback(ValueCallback&& cb) {
-        auto cache = std::make_shared<OpValueCache>(std::forward<ValueCallback>(cb));
-        return [cache](const std::vector<Sp<Value>>& vals, bool expired){
-            return cache->onValue(vals, expired);
-        };
-    }
-
-    bool onValue(const std::vector<Sp<Value>>& vals, bool expired) {
-        if (expired)
-            return onValuesExpired(vals);
-        else
-            return onValuesAdded(vals);
-    }
-
-    bool onValuesAdded(const std::vector<Sp<Value>>& vals);
-    bool onValuesExpired(const std::vector<Sp<Value>>& vals);
-
-    std::vector<Sp<Value>> get(Value::Filter& filter) const;
-    Sp<Value> get(Value::Id id) const;
-    std::vector<Sp<Value>> getValues() const;
-
-private:
-    OpValueCache(const OpValueCache&) = delete;
-    OpValueCache& operator=(const OpValueCache&) = delete;
-
-    std::map<Value::Id, OpCacheValueStorage> values {};
-    ValueCallback callback;
-};
-
-class OpCache {
-public:
-    OpCache() : cache([this](const std::vector<Sp<Value>>& vals, bool expired){
-        if (expired)
-            onValuesExpired(vals);
-        else
-            onValuesAdded(vals);
-        return true;
-    }) {}
-
-    bool onValue(const std::vector<Sp<Value>>& vals, bool expired) {
-        cache.onValue(vals, expired);
-        return not listeners.empty();
-    }
-
-    void onValuesAdded(const std::vector<Sp<Value>>& vals);
-    void onValuesExpired(const std::vector<Sp<Value>>& vals);
-
-    void addListener(size_t token, ValueCallback cb, Sp<Query> q, Value::Filter filter) {
-        listeners.emplace(token, LocalListener{q, filter, cb});
-        cb(cache.get(filter), false);
-    }
-
-    bool removeListener(size_t token, const time_point& now) {
-        lastRemoved = now;
-        return listeners.erase(token) > 0;
-    }
-
-    void removeAll() {
-        listeners.clear();
-    }
-
-    bool isDone() {
-        return listeners.empty();
-    }
-
-    std::vector<Sp<Value>> get(Value::Filter& filter) const {
-        return cache.get(filter);
-    }
-
-    Sp<Value> get(Value::Id id) const {
-        return cache.get(id);
-    }
-
-    bool isExpired(const time_point& now) const {
-        return listeners.empty() and (lastRemoved + EXPIRATION < now);
-    }
-    time_point getExpiration() const;
-
-    size_t searchToken;
-private:
-    constexpr static const std::chrono::seconds EXPIRATION {60};
-    OpCache(const OpCache&) = delete;
-    OpCache& operator=(const OpCache&) = delete;
-
-    OpValueCache cache;
-    std::map<size_t, LocalListener> listeners;
-    time_point lastRemoved {clock::now()};
-};
-
-class SearchCache {
-public:
-    SearchCache() {}
-    SearchCache(SearchCache&&) = default;
-    size_t listen(ValueCallback get_cb, Sp<Query> q, Value::Filter filter, std::function<size_t(Sp<Query>, ValueCallback)> onListen);
-
-    bool cancelListen(size_t gtoken, const time_point& now);
-    void cancelAll(std::function<void(size_t)> onCancel);
-
-    time_point expire(const time_point& now, std::function<void(size_t)> onCancel);
-    time_point getExpiration() const {
-        return nextExpiration_;
-    }
-
-    std::vector<Sp<Value>> get(Value::Filter& filter) const;
-    Sp<Value> get(Value::Id id) const;
-
-private:
-    SearchCache(const SearchCache&) = delete;
-    SearchCache& operator=(const SearchCache&) = delete;
-
-    std::map<Sp<Query>, std::unique_ptr<OpCache>> ops {};
-    size_t nextToken_ {1};
-    time_point nextExpiration_ {time_point::max()};
-};
-
-}
@@ -62,8 +62,6 @@ struct ParsedMessage {
    std::vector<Sp<Node>> nodes4, nodes6;
    /* values to store or retreive request */
    std::vector<Sp<Value>> values;
-    std::vector<Value::Id> refreshed_values {};
-    std::vector<Value::Id> expired_values {};
    /* index for fields values */
    std::vector<Sp<FieldValueIndex>> fields;
    /** When part of the message header: {index -> (total size, {})}
@@ -284,10 +282,6 @@ ParsedMessage::msgpack_unpack(msgpack::object msg)
        } else {
            throw msgpack::type_error();
        }
-    } else if (auto raw_fields = findMapValue(req, "exp")) {
-        expired_values = raw_fields->as<decltype(expired_values)>();
-    } else if (auto raw_fields = findMapValue(req, "re")) {
-        refreshed_values = raw_fields->as<decltype(refreshed_values)>();
    }

    if (auto w = findMapValue(req, "w")) {
@@ -21,8 +21,6 @@
 #include "value.h"
 #include "request.h"
 #include "listener.h"
-#include "value_cache.h"
-#include "op_cache.h"

 namespace dht {

@@ -61,17 +59,6 @@ struct Dht::SearchNode {
     */
    using SyncStatus = std::map<Sp<Query>, Sp<net::Request>>;

-    struct CachedListenStatus {
-        ValueCache cache;
-        Sp<Scheduler::Job> cacheExpirationJob {};
-        Sp<net::Request> req {};
-        CachedListenStatus(ValueStateCallback&& cb) : cache(std::forward<ValueStateCallback>(cb)) {}
-        CachedListenStatus(CachedListenStatus&&) = default;
-        CachedListenStatus(const CachedListenStatus&) = delete;
-        CachedListenStatus& operator=(const CachedListenStatus&) = delete;
-    };
-    using NodeListenerStatus = std::map<Sp<Query>, CachedListenStatus>;
-
    Sp<Node> node {};                 /* the node info */

    /* queries sent for finding out values hosted by the node */
@@ -80,7 +67,7 @@ struct Dht::SearchNode {
    std::map<Sp<Query>, std::vector<Sp<Query>>> pagination_queries {};

    SyncStatus getStatus {};    /* get/sync status */
-    NodeListenerStatus listenStatus {}; /* listen status */
+    SyncStatus listenStatus {}; /* listen status */
    AnnounceStatus acked {};    /* announcement status for a given value id */

    Blob token {};                                 /* last token the node sent to us after a get request */
@@ -214,25 +201,6 @@ struct Dht::SearchNode {
        getStatus.clear();
    }

-    void onValues(const Sp<Query>& q, net::RequestAnswer&& answer, const TypeStore& types, Scheduler& scheduler)
-    {
-        auto l = listenStatus.find(q);
-        if (l != listenStatus.end()) {
-            auto next = l->second.cache.onValues(answer.values,
-                                     answer.refreshed_values,
-                                     answer.expired_values, types, scheduler.time());
-            scheduler.edit(l->second.cacheExpirationJob, next);
-        }
-    }
-
-    void expireValues(const Sp<Query>& q, Scheduler& scheduler) {
-        auto l = listenStatus.find(q);
-        if (l != listenStatus.end()) {
-            auto next = l->second.cache.expireValues(scheduler.time());
-            scheduler.edit(l->second.cacheExpirationJob, next);
-        }
-    }
-
    /**
     * Tells if a request in the status map is expired.
     *
@@ -260,12 +228,6 @@ struct Dht::SearchNode {
                return r.second and r.second->pending();
            }) != status.end();
    }
-    static bool pending(const NodeListenerStatus& status) {
-        return std::find_if(status.begin(), status.end(),
-            [](const NodeListenerStatus::value_type& r){
-                return r.second.req and r.second.req->pending();
-            }) != status.end();
-    }

    bool pendingGet() const { return pending(getStatus); }

@@ -301,20 +263,20 @@ struct Dht::SearchNode {
        else
            return isListening(now, ls);
    }
-    bool isListening(time_point now, NodeListenerStatus::const_iterator listen_status) const {
+    bool isListening(time_point now, SyncStatus::const_iterator listen_status) const {
        if (listen_status == listenStatus.end())
            return false;
-        return listen_status->second.req->reply_time + LISTEN_EXPIRE_TIME > now;
+        return listen_status->second->reply_time + LISTEN_EXPIRE_TIME > now;
    }
    void cancelListen() {
        for (const auto& status : listenStatus)
-            node->cancelRequest(status.second.req);
+            node->cancelRequest(status.second);
        listenStatus.clear();
    }
    void cancelListen(const Sp<Query>& query) {
        auto it = listenStatus.find(query);
        if (it != listenStatus.end()) {
-            node->cancelRequest(it->second.req);
+            node->cancelRequest(it->second);
            listenStatus.erase(it);
        }
    }
@@ -340,10 +302,10 @@ struct Dht::SearchNode {
     */
    time_point getListenTime(const Sp<Query>& q) const {
        auto listen_status = listenStatus.find(q);
-        if (listen_status == listenStatus.end() or not listen_status->second.req)
+        if (listen_status == listenStatus.end() or not listen_status->second)
            return time_point::min();
-        return listen_status->second.req->pending() ? time_point::max() :
-            listen_status->second.req->reply_time + LISTEN_EXPIRE_TIME - REANNOUNCE_MARGIN;
+        return listen_status->second->pending() ? time_point::max() :
+            listen_status->second->reply_time + LISTEN_EXPIRE_TIME - REANNOUNCE_MARGIN;
    }

    /**
@@ -378,16 +340,10 @@ struct Dht::Search {
    std::multimap<time_point, Get> callbacks {};

    /* listeners */
-    std::map<size_t, SearchListener> listeners {};
+    std::map<size_t, LocalListener> listeners {};
    size_t listener_token = 1;

-    /* Cache */
-    SearchCache cache;
-    Sp<Scheduler::Job> opExpirationJob;
-
    ~Search() {
-        if (opExpirationJob)
-            opExpirationJob->cancel();
        for (auto& get : callbacks) {
            get.second.done_cb(false, {});
            get.second.done_cb = {};
@@ -475,38 +431,19 @@ struct Dht::Search {
    bool isAnnounced(Value::Id id) const;
    bool isListening(time_point now) const;

-    size_t listen(ValueCallback cb, Value::Filter f, const Sp<Query>& q, Scheduler& scheduler) {
-        //DHT_LOG.e(id, "[search %s IPv%c] listen", id.toString().c_str(), (af == AF_INET) ? '4' : '6');
-        return cache.listen(cb, q, f, [&](const Sp<Query>& q, ValueCallback vcb){
-            done = false;
-            auto token = ++listener_token;
-            listeners.emplace(token, SearchListener{q, f, vcb});
-            scheduler.edit(nextSearchStep, scheduler.time());
-            return token;
-        });
-    }
-
-    void cancelListen(size_t token, Scheduler& scheduler) {
-        cache.cancelListen(token, scheduler.time());
-        if (not opExpirationJob)
-            opExpirationJob = scheduler.add(time_point::max(), [this,&scheduler]{
-                auto nextExpire = cache.expire(scheduler.time(), [&](size_t t){
-                    Sp<Query> query;
-                    const auto& ll = listeners.find(t);
-                    if (ll != listeners.cend()) {
-                        query = ll->second.query;
-                        listeners.erase(ll);
-                    }
-                    for (auto& sn : nodes) {
-                        if (listeners.empty())
-                            sn.cancelListen();
-                        else if (query)
-                            sn.cancelListen(query);
-                    }
-                });
-                scheduler.edit(opExpirationJob, nextExpire);
-            });
-        scheduler.edit(opExpirationJob, cache.getExpiration());
+    void cancelListen(size_t token) {
+        Sp<Query> query;
+        const auto& ll = listeners.find(token);
+        if (ll != listeners.cend()) {
+            query = ll->second.query;
+            listeners.erase(ll);
+        }
+        for (auto& sn : nodes) {
+            if (listeners.empty())
+                sn.cancelListen();
+            else if (query)
+                sn.cancelListen(query);
+        }
    }

    /**
@@ -795,7 +732,7 @@ Dht::Search::isListening(time_point now) const
    for (const auto& n : nodes) {
        if (n.isBad())
            continue;
-        SearchNode::NodeListenerStatus::const_iterator ls {};
+        SearchNode::SyncStatus::const_iterator ls {};
        for (ls = n.listenStatus.begin(); ls != n.listenStatus.end() ; ++ls) {
            if (n.isListening(now, ls))
                break;
@@ -213,84 +213,58 @@ SecureDht::findPublicKey(const InfoHash& node, std::function<void(const Sp<const
    findCertificate(node, [=](const Sp<crypto::Certificate> crt) {
        if (crt && *crt) {
            auto pk = std::make_shared<crypto::PublicKey>(crt->getPublicKey());
-            if (*pk) {
-                nodesPubKeys_[pk->getId()] = pk;
-                if (cb) cb(pk);
-                return;
-            }
+            nodesPubKeys_[pk->getId()] = pk;
+            if (cb) cb(pk);
+        } else {
+            if (cb) cb(nullptr);
        }
-        if (cb) cb(nullptr);
    });
 }

-Sp<Value>
-SecureDht::checkValue(const Sp<Value>& v)
-{
-    // Decrypt encrypted values
-    if (v->isEncrypted()) {
-        if (not key_) {
-#if OPENDHT_PROXY_SERVER
-            if (forward_all_) // We are currently a proxy, send messages to clients.
-                return v;
-#endif
-            return {};
-        }
-        try {
-            Value decrypted_val (decrypt(*v));
-            if (decrypted_val.recipient == getId()) {
-                if (decrypted_val.owner)
-                    nodesPubKeys_[decrypted_val.owner->getId()] = decrypted_val.owner;
-                return std::make_shared<Value>(std::move(decrypted_val));
-            }
-            // Ignore values belonging to other people
-        } catch (const std::exception& e) {
-            DHT_LOG.WARN("Could not decrypt value %s : %s", v->toString().c_str(), e.what());
-        }
-    }
-    // Check signed values
-    else if (v->isSigned()) {
-        if (v->owner and v->owner->checkSignature(v->getToSign(), v->signature)) {
-            nodesPubKeys_[v->owner->getId()] = v->owner;
-            return v;
-        }
-        else
-            DHT_LOG.WARN("Signature verification failed for %s", v->toString().c_str());
-    }
-    // Forward normal values
-    else {
-        return v;
-    }
-    return {};
-}
-
-ValueCallback
-SecureDht::getCallbackFilter(ValueCallback cb, Value::Filter&& filter)
-{
-    return [=](const std::vector<Sp<Value>>& values, bool expired) {
-        std::vector<Sp<Value>> tmpvals {};
-        for (const auto& v : values) {
-            if (auto nv = checkValue(v))
-                if (not filter or filter(*nv))
-                    tmpvals.emplace_back(std::move(nv));
-        }
-        if (cb and not tmpvals.empty())
-            return cb(tmpvals, expired);
-        return true;
-    };
-}
-
-
 GetCallback
 SecureDht::getCallbackFilter(GetCallback cb, Value::Filter&& filter)
 {
    return [=](const std::vector<Sp<Value>>& values) {
        std::vector<Sp<Value>> tmpvals {};
        for (const auto& v : values) {
-            if (auto nv = checkValue(v))
-                if (not filter or filter(*nv))
-                    tmpvals.emplace_back(std::move(nv));
+            // Decrypt encrypted values
+            if (v->isEncrypted()) {
+                if (not key_) {
+#if OPENDHT_PROXY_SERVER
+                    if (forward_all_) // We are currently a proxy, send messages to clients.
+                        tmpvals.push_back(v);
+#endif
+                    continue;
+                }
+                try {
+                    Value decrypted_val (decrypt(*v));
+                    if (decrypted_val.recipient == getId()) {
+                        nodesPubKeys_[decrypted_val.owner->getId()] = decrypted_val.owner;
+                        if (not filter or filter(decrypted_val))
+                            tmpvals.push_back(std::make_shared<Value>(std::move(decrypted_val)));
+                    }
+                    // Ignore values belonging to other people
+                } catch (const std::exception& e) {
+                    DHT_LOG.WARN("Could not decrypt value %s : %s", v->toString().c_str(), e.what());
+                }
+            }
+            // Check signed values
+            else if (v->isSigned()) {
+                if (v->owner and v->owner->checkSignature(v->getToSign(), v->signature)) {
+                    nodesPubKeys_[v->owner->getId()] = v->owner;
+                    if (not filter  or filter(*v))
+                        tmpvals.push_back(v);
+                }
+                else
+                    DHT_LOG.WARN("Signature verification failed for %s", v->toString().c_str());
+            }
+            // Forward normal values
+            else {
+                if (not filter or filter(*v))
+                    tmpvals.push_back(v);
+            }
        }
-        if (cb and not tmpvals.empty())
+        if (cb && not tmpvals.empty())
            return cb(tmpvals);
        return true;
    };
@@ -302,13 +276,6 @@ SecureDht::get(const InfoHash& id, GetCallback cb, DoneCallback donecb, Value::F
    dht_->get(id, getCallbackFilter(cb, std::forward<Value::Filter>(f)), donecb, {}, std::forward<Where>(w));
 }

-size_t
-SecureDht::listen(const InfoHash& id, ValueCallback cb, Value::Filter f, Where w)
-{
-    return dht_->listen(id, getCallbackFilter(cb, std::forward<Value::Filter>(f)), {}, std::forward<Where>(w));
-}
-
-
 size_t
 SecureDht::listen(const InfoHash& id, GetCallback cb, Value::Filter f, Where w)
 {
@@ -320,7 +287,6 @@ SecureDht::putSigned(const InfoHash& hash, Sp<Value> val, DoneCallback callback,
 {
    if (val->id == Value::INVALID_ID) {
        crypto::random_device rdev;
-        std::uniform_int_distribution<Value::Id> rand_id;
        val->id = rand_id(rdev);
    }

@@ -167,7 +167,7 @@ struct Storage {

    StoreDiff remove(const InfoHash& id, Value::Id);

-    std::pair<ssize_t, std::vector<Sp<Value>>> expire(const InfoHash& id, time_point now);
+    StoreDiff expire(const InfoHash& id, time_point now);

 private:
    Storage(const Storage&) = delete;
@@ -245,7 +245,7 @@ Storage::clear()
    return {-tot_size, -num_values, 0};
 }

-std::pair<ssize_t, std::vector<Sp<Value>>>
+Storage::StoreDiff
 Storage::expire(const InfoHash& id, time_point now)
 {
    // expire listeners
@@ -271,18 +271,16 @@ Storage::expire(const InfoHash& id, time_point now)
    auto r = std::partition(values.begin(), values.end(), [&](const ValueStorage& v) {
        return v.expiration > now;
    });
-    std::vector<Sp<Value>> ret;
-    ret.reserve(std::distance(r, values.end()));
+    ssize_t del_num = -std::distance(r, values.end());
    ssize_t size_diff {};
    std::for_each(r, values.end(), [&](const ValueStorage& v) {
        size_diff -= v.data->size();
        if (v.store_bucket)
            v.store_bucket->erase(id, *v.data, v.expiration);
-        ret.emplace_back(std::move(v.data));
    });
    total_size += size_diff;
    values.erase(r, values.end());
-    return {size_diff, std::move(ret)};
+    return {size_diff, del_num, del_listen};
 }

 }
@@ -29,26 +29,6 @@ namespace dht {

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

-std::pair<std::string, std::string>
-splitPort(const std::string& s) {
-    if (s.empty())
-        return {};
-    if (s[0] == '[') {
-        std::size_t closure = s.find_first_of(']');
-        std::size_t found = s.find_last_of(':');
-        if (closure == std::string::npos)
-            return {s, ""};
-        if (found == std::string::npos or found < closure)
-            return {s.substr(1,closure-1), ""};
-        return {s.substr(1,closure-1), s.substr(found+1)};
-    }
-    std::size_t found = s.find_last_of(':');
-    std::size_t first = s.find_first_of(':');
-    if (found == std::string::npos or found != first)
-        return {s, ""};
-    return {s.substr(0,found), s.substr(found+1)};
-}
-
 std::vector<SockAddr>
 SockAddr::resolve(const std::string& host, const std::string& service)
 {
@@ -175,37 +175,40 @@ Value::msgpack_unpack_body(const msgpack::object& o)
 #if OPENDHT_PROXY_SERVER || OPENDHT_PROXY_CLIENT
 Value::Value(Json::Value& json)
 {
-    id = Value::Id(unpackId(json, "id"));
-    if (json.isMember("cypher")) {
-        auto cypherStr = json["cypher"].asString();
-        cypherStr = base64_decode(cypherStr);
-        cypher = std::vector<unsigned char>(cypherStr.begin(), cypherStr.end());
-    }
-    if (json.isMember("sig")) {
-        auto sigStr = json["sig"].asString();
-        sigStr = base64_decode(sigStr);
-        signature = std::vector<unsigned char>(sigStr.begin(), sigStr.end());
-    }
-    if (json.isMember("seq"))
-        seq = json["seq"].asInt();
-    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);
-    }
-    if (json.isMember("to")) {
-        auto toStr = json["to"].asString();
-        recipient = InfoHash(toStr);
-    }
-    if (json.isMember("type"))
-        type = json["type"].asInt();
-    if (json.isMember("data")){
-        auto dataStr = json["data"].asString();
-        dataStr = base64_decode(dataStr);
-        data = std::vector<unsigned char>(dataStr.begin(), dataStr.end());
-    }
-    if (json.isMember("utype"))
-        user_type = json["utype"].asString();
+   try {
+       if (json.isMember("id"))
+           id = ValueType::Id(json["id"].asInt());
+   } catch (...) { }
+   if (json.isMember("cypher")) {
+       auto cypherStr = json["cypher"].asString();
+       cypherStr = base64_decode(cypherStr);
+       cypher = std::vector<unsigned char>(cypherStr.begin(), cypherStr.end());
+   }
+   if (json.isMember("sig")) {
+       auto sigStr = json["sig"].asString();
+       sigStr = base64_decode(sigStr);
+       signature = std::vector<unsigned char>(sigStr.begin(), sigStr.end());
+   }
+   if (json.isMember("seq"))
+       seq = json["seq"].asInt();
+   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);
+   }
+   if (json.isMember("to")) {
+       auto toStr = json["to"].asString();
+       recipient = InfoHash(toStr);
+   }
+   if (json.isMember("type"))
+       type = json["type"].asInt();
+   if (json.isMember("data")){
+       auto dataStr = json["data"].asString();
+       dataStr = base64_decode(dataStr);
+       data = std::vector<unsigned char>(dataStr.begin(), dataStr.end());
+   }
+   if (json.isMember("utype"))
+       user_type = json["utype"].asString();
 }

 Json::Value
@@ -222,7 +225,7 @@ Value::toJson() const
        if (has_owner) { // isSigned
            val["seq"] = seq;
            val["owner"] = owner->toString();
-            if (recipient)
+            if (recipient != InfoHash())
                val["to"] = recipient.toString();
        }
        val["type"] = type;
@@ -232,22 +235,6 @@ Value::toJson() const
    }
    return val;
 }
-
-uint64_t
-unpackId(const Json::Value& json, const std::string& key) {
-    uint64_t ret = 0;
-    try {
-        if (json.isMember(key)) {
-            const auto& t = json[key];
-            if (t.isString()) {
-                ret = std::stoull(t.asString());
-            } else {
-                ret = t.asLargestUInt();
-            }
-        }
-    } catch (...) {}
-    return ret;
-}
 #endif //OPENDHT_PROXY_SERVER

 bool
@@ -1,189 +0,0 @@
-/*
- *  Copyright (C) 2018 Savoir-faire Linux Inc.
- *  Author(s) : Adrien Béraud <adrien.beraud@savoirfairelinux.com>
- *
- *  This program is free software; you can redistribute it and/or modify
- *  it under the terms of the GNU General Public License as published by
- *  the Free Software Foundation; either version 3 of the License, or
- *  (at your option) any later version.
- *
- *  This program is distributed in the hope that it will be useful,
- *  but WITHOUT ANY WARRANTY; without even the implied warranty of
- *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- *  GNU General Public License for more details.
- *
- *  You should have received a copy of the GNU General Public License
- *  along with this program. If not, see <https://www.gnu.org/licenses/>.
- */
-#pragma once
-
-#include "value.h"
-
-namespace dht {
-
-using ValueStateCallback = std::function<void(const std::vector<Sp<Value>>&, bool)>;
-using CallbackQueue = std::list<std::function<void()>>;
-
-class ValueCache {
-public:
-    ValueCache(ValueStateCallback&& cb) : callback(std::forward<ValueStateCallback>(cb)) {}
-    ValueCache(ValueCache&& o) : values(std::move(o.values)), callback(std::move(o.callback)) {
-        o.callback = {};
-    }
-
-    ~ValueCache() {
-        auto q = clear();
-        for (auto& cb: q)
-            cb();
-    }
-
-    CallbackQueue clear() {
-        std::vector<Sp<Value>> expired_values;
-        expired_values.reserve(values.size());
-        for (const auto& v : values)
-            expired_values.emplace_back(std::move(v.second.data));
-        values.clear();
-        CallbackQueue ret;
-        if (not expired_values.empty() and callback) {
-            auto cb = callback;
-            ret.emplace_back([expired_values, cb]{
-                cb(expired_values, true);
-            });
-        }
-        return ret;
-    }
-
-    time_point expireValues(const time_point& now) {
-        time_point ret = time_point::max();
-        auto cbs = expireValues(now, ret);
-        while (not cbs.empty()) {
-            cbs.front()();
-            cbs.pop_front();
-        }
-        return ret;
-    }
-
-    CallbackQueue expireValues(const time_point& now, time_point& next) {
-        std::vector<Sp<Value>> expired_values;
-        for (auto it = values.begin(); it != values.end();) {
-            if (it->second.expiration < now) {
-                expired_values.emplace_back(std::move(it->second.data));
-                it = values.erase(it);
-            } else {
-                next = std::min(next, it->second.expiration);
-                ++it;
-            }
-        }
-        while (values.size() >= MAX_VALUES) {
-            // too many values, remove oldest values
-            time_point oldest_creation = time_point::max();
-            auto oldest_value = values.end();
-            for (auto it = values.begin(); it != values.end(); ++it)
-                if (it->second.created < oldest_creation) {
-                    oldest_value = it;
-                    oldest_creation = it->second.created;
-                }
-            if (oldest_value != values.end()) {
-                expired_values.emplace_back(std::move(oldest_value->second.data));
-                values.erase(oldest_value);
-            }
-        }
-        CallbackQueue ret;
-        if (not expired_values.empty() and callback) {
-            auto cb = callback;
-            ret.emplace_back([cb, expired_values]{
-                if (cb) cb(expired_values, true);
-            });
-        }
-        return ret;
-    }
-
-    time_point onValues
-        (const std::vector<Sp<Value>>& values,
-        const std::vector<Value::Id>& refreshed_values,
-        const std::vector<Value::Id>& expired_values,
-        const TypeStore& types, const time_point& now)
-    {
-        CallbackQueue cbs;
-        time_point ret = time_point::max();
-        if (not values.empty())
-            cbs.splice(cbs.end(), addValues(values, types, now));
-        for (const auto& vid : refreshed_values)
-            refreshValue(vid, types, now);
-        for (const auto& vid : expired_values)
-            cbs.splice(cbs.end(), expireValue(vid));
-        cbs.splice(cbs.end(), expireValues(now, ret));
-        while (not cbs.empty()) {
-            cbs.front()();
-            cbs.pop_front();
-        }
-        return ret;
-    }
-
-private:
-    // prevent copy
-    ValueCache(const ValueCache&) = delete;
-    ValueCache& operator=(const ValueCache&) = delete;
-    ValueCache& operator=(ValueCache&&) = delete;
-
-    /* The maximum number of values we store in the cache. */
-    static constexpr unsigned MAX_VALUES {1024};
-
-    struct CacheValueStorage {
-        Sp<Value> data {};
-        time_point created {};
-        time_point expiration {};
-
-        CacheValueStorage() {}
-        CacheValueStorage(const Sp<Value>& v, time_point t, time_point e)
-         : data(v), created(t), expiration(e) {}
-    };
-
-    std::map<Value::Id, CacheValueStorage> values;
-    ValueStateCallback callback;
-
-    CallbackQueue addValues(const std::vector<Sp<Value>>& new_values, const TypeStore& types, const time_point& now) {
-        std::vector<Sp<Value>> nvals;
-        for (const auto& value : new_values) {
-            auto v = values.find(value->id);
-            if (v == values.end()) {
-                // new value
-                nvals.emplace_back(value);
-                values.emplace(value->id, CacheValueStorage(value, now, now + types.getType(value->type).expiration));
-            } else {
-                // refreshed value
-                v->second.created = now;
-                v->second.expiration = now + types.getType(v->second.data->type).expiration;
-            }
-        }
-        auto cb = callback;
-        CallbackQueue ret;
-        if (not nvals.empty())
-            ret.emplace_back([cb, nvals]{
-                if (cb) cb(nvals, false);
-            });
-        return ret;
-    }
-    CallbackQueue expireValue(Value::Id vid) {
-        auto v = values.find(vid);
-        if (v == values.end())
-            return {};
-        const std::vector<Sp<Value>> val {std::move(v->second.data)};
-        values.erase(v);
-        auto cb = callback;
-        CallbackQueue ret;
-        ret.emplace_back([cb, val]{
-            if (cb) cb(val, true);
-        });
-        return ret;
-    }
-    void refreshValue(Value::Id vid, const TypeStore& types, const time_point& now) {
-        auto v = values.find(vid);
-        if (v == values.end())
-            return;
-        v->second.created = now;
-        v->second.expiration = now + types.getType(v->second.data->type).expiration;
-    }
-};
-
-}
@@ -1,9 +0,0 @@
-if ENABLE_TESTS
-bin_PROGRAMS = opendht_unit_tests
-
-AM_CPPFLAGS = -I../include
-
-nobase_include_HEADERS = infohashtester.h dhtrunnertester.h
-opendht_unit_tests_SOURCES = tests_runner.cpp infohashtester.cpp dhtrunnertester.cpp
-opendht_unit_tests_LDFLAGS = -lopendht -lcppunit -L@top_builddir@/src/.libs @GnuTLS_LIBS@
-endif
@@ -1,64 +0,0 @@
-/*
- *  Copyright (C) 2018 Savoir-faire Linux Inc.
- *
- *  Author: Adrien Béraud <adrien.beraud@savoirfairelinux.com>
- *
- *  This program is free software; you can redistribute it and/or modify
- *  it under the terms of the GNU General Public License as published by
- *  the Free Software Foundation; either version 3 of the License, or
- *  (at your option) any later version.
- *
- *  This program is distributed in the hope that it will be useful,
- *  but WITHOUT ANY WARRANTY; without even the implied warranty of
- *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- *  GNU General Public License for more details.
- *
- *  You should have received a copy of the GNU General Public License
- *  along with this program. If not, see <https://www.gnu.org/licenses/>.
- */
-
-#include "dhtrunnertester.h"
-
-// std
-#include <iostream>
-#include <string>
-
-namespace test {
-CPPUNIT_TEST_SUITE_REGISTRATION(DhtRunnerTester);
-
-void
-DhtRunnerTester::setUp() {
-    node1.run(42222, {}, true);
-    node2.run(42232, {}, true);
-    node2.bootstrap(node1.getBound());
-}
-
-void
-DhtRunnerTester::tearDown() {
-    node1.join();
-    node2.join();
-}
-
-void
-DhtRunnerTester::testConstructors() {
-    CPPUNIT_ASSERT(node1.getBoundPort() == 42222);
-    CPPUNIT_ASSERT(node2.getBoundPort() == 42232);
-}
-
-void
-DhtRunnerTester::testGetPut() {
-    auto key = dht::InfoHash::get("123");
-    dht::Value val {"hey"};
-    auto val_data = val.data;
-    node1.put(key, std::move(val));
-    auto vals = node2.get(key).get();
-    CPPUNIT_ASSERT(not vals.empty());
-    CPPUNIT_ASSERT(vals.front()->data == val_data);
-}
-
-void
-DhtRunnerTester::testListen() {
-    // TODO
-}
-
-}  // namespace test
@@ -1,62 +0,0 @@
-/*
- *  Copyright (C) 2018 Savoir-faire Linux Inc.
- *
- *  Author: Adrien Béraud <adrien.beraud@savoirfairelinux.com>
- *
- *  This program is free software; you can redistribute it and/or modify
- *  it under the terms of the GNU General Public License as published by
- *  the Free Software Foundation; either version 3 of the License, or
- *  (at your option) any later version.
- *
- *  This program is distributed in the hope that it will be useful,
- *  but WITHOUT ANY WARRANTY; without even the implied warranty of
- *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- *  GNU General Public License for more details.
- *
- *  You should have received a copy of the GNU General Public License
- *  along with this program. If not, see <https://www.gnu.org/licenses/>.
- */
-
-#pragma once
-
-// cppunit
-#include <cppunit/TestFixture.h>
-#include <cppunit/extensions/HelperMacros.h>
-
-#include <opendht/dhtrunner.h>
-
-namespace test {
-
-class DhtRunnerTester : public CppUnit::TestFixture {
-    CPPUNIT_TEST_SUITE(DhtRunnerTester);
-    CPPUNIT_TEST(testConstructors);
-    CPPUNIT_TEST(testGetPut);
-    CPPUNIT_TEST(testListen);
-    CPPUNIT_TEST_SUITE_END();
-
-    dht::DhtRunner node1 {};
-    dht::DhtRunner node2 {};
- public:
-    /**
-     * Method automatically called before each test by CppUnit
-     */
-    void setUp();
-    /**
-     * Method automatically called after each test CppUnit
-     */
-    void tearDown();
-    /**
-     * Test the differents behaviors of constructors
-     */
-    void testConstructors();
-    /**
-     * Test get and put methods
-     */
-    void testGetPut();
-    /**
-     * Test listen method
-     */
-    void testListen();
-};
-
-}  // namespace test
@@ -1,144 +0,0 @@
-/*
- *  Copyright (C) 2018 Savoir-faire Linux Inc.
- *
- *  Author: Sébastien Blin <sebastien.blin@savoirfairelinux.com>
- *
- *  This program is free software; you can redistribute it and/or modify
- *  it under the terms of the GNU General Public License as published by
- *  the Free Software Foundation; either version 3 of the License, or
- *  (at your option) any later version.
- *
- *  This program is distributed in the hope that it will be useful,
- *  but WITHOUT ANY WARRANTY; without even the implied warranty of
- *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- *  GNU General Public License for more details.
- *
- *  You should have received a copy of the GNU General Public License
- *  along with this program. If not, see <https://www.gnu.org/licenses/>.
- */
-
-#include "infohashtester.h"
-
-// std
-#include <iostream>
-#include <string>
-
-// opendht
-#include "opendht/infohash.h"
-
-namespace test {
-CPPUNIT_TEST_SUITE_REGISTRATION(InfoHashTester);
-
-void
-InfoHashTester::setUp() {
-
-}
-
-void
-InfoHashTester::testConstructors() {
-    // Default constructor creates a null infohash
-    auto nullHash = dht::InfoHash();
-    CPPUNIT_ASSERT(nullHash.size() == 20);
-    CPPUNIT_ASSERT(!nullHash);
-    // Build from a uint8_t. if length to short, should get a null infohash
-    uint8_t to_short[] = {0x0, 0x1, 0x2, 0x3, 0x4, 0x5, 0x6, 0x7, 0x8};
-    auto infohash = dht::InfoHash(to_short, 8);
-    CPPUNIT_ASSERT(infohash.size() == 20);
-    CPPUNIT_ASSERT_EQUAL(infohash.toString(),
-        std::string("0000000000000000000000000000000000000000"));
-    // Build from a uint8_t. if length is enough, data should contains the uint8_t
-    uint8_t enough[] = {0x1, 0x2, 0x3, 0x4, 0x5, 0x6, 0x7, 0x8, 0x9, 0xa,
-                        0x1, 0x2, 0x3, 0x4, 0x5, 0x6, 0x7, 0x8, 0x9, 0xa};
-    infohash = dht::InfoHash(enough, 20);
-    CPPUNIT_ASSERT(infohash.size() == 20);
-    const auto* data = infohash.data();
-    for (auto i = 0; i < 20; ++i) {
-        CPPUNIT_ASSERT_EQUAL(enough[i], data[i]);
-    }
-    // if too long, should be cutted to 20
-    uint8_t tooLong[] = {0x1, 0x2, 0x3, 0x4, 0x5, 0x6, 0x7, 0x8, 0x9, 0xa,
-                        0x1, 0x2, 0x3, 0x4, 0x5, 0x6, 0x7, 0x8, 0x9, 0xa, 0xb0};
-    infohash = dht::InfoHash(tooLong, 21);
-    CPPUNIT_ASSERT(infohash.size() == 20);
-    const auto* data2 = infohash.data();
-    for (auto i = 0; i < 20; ++i) {
-        CPPUNIT_ASSERT_EQUAL(enough[i], data2[i]);
-    }
-    // Build from string
-    auto infohashFromStr = dht::InfoHash("0102030405060708090A0102030405060708090A");
-    CPPUNIT_ASSERT(infohashFromStr.size() == 20);
-    const auto* dataStr = infohashFromStr.data();
-    for (auto i = 0; i < 20; ++i) {
-        CPPUNIT_ASSERT_EQUAL((int)dataStr[i], (int)data[i]);
-    }
-}
-
-void
-InfoHashTester::testComperators() {
-    auto nullHash = dht::InfoHash();
-    auto minHash = dht::InfoHash("0000000000000000000000000000000000111110");
-    auto maxHash = dht::InfoHash("0111110000000000000000000000000000000000");
-    // operator ==
-    CPPUNIT_ASSERT_EQUAL(minHash, minHash);
-    CPPUNIT_ASSERT_EQUAL(minHash, dht::InfoHash("0000000000000000000000000000000000111110"));
-	CPPUNIT_ASSERT(!(minHash == maxHash));
-    // operator !=
-    CPPUNIT_ASSERT(!(minHash != minHash));
-    CPPUNIT_ASSERT(!(minHash != dht::InfoHash("0000000000000000000000000000000000111110")));
-	CPPUNIT_ASSERT(minHash != maxHash);
-    // operator<
-    CPPUNIT_ASSERT(nullHash < minHash);
-    CPPUNIT_ASSERT(nullHash < maxHash);
-    CPPUNIT_ASSERT(minHash < maxHash);
-    CPPUNIT_ASSERT(!(minHash < nullHash));
-    CPPUNIT_ASSERT(!(maxHash < nullHash));
-    CPPUNIT_ASSERT(!(maxHash < minHash));
-    CPPUNIT_ASSERT(!(minHash < minHash));
-    // bool()
-    CPPUNIT_ASSERT(maxHash);
-    CPPUNIT_ASSERT(!nullHash);
-
-}
-
-void
-InfoHashTester::testLowBit() {
-    auto nullHash = dht::InfoHash();
-    auto minHash = dht::InfoHash("0000000000000000000000000000000000000010");
-    auto maxHash = dht::InfoHash("0100000000000000000000000000000000000000");
-    CPPUNIT_ASSERT_EQUAL(nullHash.lowbit(), -1);
-    CPPUNIT_ASSERT_EQUAL(minHash.lowbit(), 155);
-    CPPUNIT_ASSERT_EQUAL(maxHash.lowbit(), 7);
-}
-
-void
-InfoHashTester::testCommonBits() {
-    auto nullHash = dht::InfoHash();
-    auto minHash = dht::InfoHash("0000000000000000000000000000000000000010");
-    auto maxHash = dht::InfoHash("0100000000000000000000000000000000000000");
-    CPPUNIT_ASSERT_EQUAL(dht::InfoHash::commonBits(nullHash, nullHash), (unsigned)160);
-    CPPUNIT_ASSERT_EQUAL(dht::InfoHash::commonBits(nullHash, minHash), (unsigned)155);
-    CPPUNIT_ASSERT_EQUAL(dht::InfoHash::commonBits(nullHash, maxHash), (unsigned)7);
-    CPPUNIT_ASSERT_EQUAL(dht::InfoHash::commonBits(minHash, maxHash), (unsigned)7);
-}
-
-void
-InfoHashTester::testXorCmp() {
-    auto nullHash = dht::InfoHash();
-    auto minHash = dht::InfoHash("0000000000000000000000000000000000000010");
-    auto maxHash = dht::InfoHash("0100000000000000000000000000000000000000");
-    CPPUNIT_ASSERT_EQUAL(minHash.xorCmp(nullHash, maxHash), -1);
-    CPPUNIT_ASSERT_EQUAL(minHash.xorCmp(maxHash, nullHash), 1);
-    CPPUNIT_ASSERT_EQUAL(minHash.xorCmp(minHash, maxHash), -1);
-    CPPUNIT_ASSERT_EQUAL(minHash.xorCmp(maxHash, minHash), 1);
-    CPPUNIT_ASSERT_EQUAL(nullHash.xorCmp(minHash, maxHash), -1);
-    CPPUNIT_ASSERT_EQUAL(nullHash.xorCmp(maxHash, minHash), 1);
-    // Because hashes are circular in distance.
-    CPPUNIT_ASSERT_EQUAL(maxHash.xorCmp(nullHash, minHash), -1);
-    CPPUNIT_ASSERT_EQUAL(maxHash.xorCmp(minHash, nullHash), 1);
-}
-
-void
-InfoHashTester::tearDown() {
-
-}
-}  // namespace test
@@ -1,69 +0,0 @@
-/*
- *  Copyright (C) 2018 Savoir-faire Linux Inc.
- *
- *  Author: Sébastien Blin <sebastien.blin@savoirfairelinux.com>
- *
- *  This program is free software; you can redistribute it and/or modify
- *  it under the terms of the GNU General Public License as published by
- *  the Free Software Foundation; either version 3 of the License, or
- *  (at your option) any later version.
- *
- *  This program is distributed in the hope that it will be useful,
- *  but WITHOUT ANY WARRANTY; without even the implied warranty of
- *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- *  GNU General Public License for more details.
- *
- *  You should have received a copy of the GNU General Public License
- *  along with this program. If not, see <https://www.gnu.org/licenses/>.
- */
-
-#pragma once
-
-// cppunit
-#include <cppunit/TestFixture.h>
-#include <cppunit/extensions/HelperMacros.h>
-
-namespace test {
-
-class InfoHashTester : public CppUnit::TestFixture {
-    CPPUNIT_TEST_SUITE(InfoHashTester);
-    CPPUNIT_TEST(testConstructors);
-    CPPUNIT_TEST(testComperators);
-    CPPUNIT_TEST(testLowBit);
-    CPPUNIT_TEST(testCommonBits);
-    CPPUNIT_TEST(testXorCmp);
-    CPPUNIT_TEST_SUITE_END();
-
- public:
-    /**
-     * Method automatically called before each test by CppUnit
-     */
-    void setUp();
-    /**
-     * Method automatically called after each test CppUnit
-     */
-    void tearDown();
-    /**
-     * Test the differents behaviors of constructors
-     */
-    void testConstructors();
-    /**
-     * Test compare operators
-     */
-    void testComperators();
-    /**
-     * Test lowbit method
-     */
-    void testLowBit();
-    /**
-     * Test commonBits method
-     */
-    void testCommonBits();
-    /**
-     * Test xorCmp operators
-     */
-    void testXorCmp();
-
-};
-
-}  // namespace test
@@ -1,35 +0,0 @@
-/*
- *  Copyright (C) 2017-2018 Savoir-faire Linux Inc.
- *  Author: Sébastien Blin <sebastien.blin@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, write to the Free Software
- *  Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301 USA.
- */
-#include <cppunit/extensions/TestFactoryRegistry.h>
-#include <cppunit/ui/text/TestRunner.h>
-#include <cppunit/CompilerOutputter.h>
-#include <iostream>
-
-int main(int argc, char** argv) {
-    CppUnit::TestFactoryRegistry &registry = CppUnit::TestFactoryRegistry::getRegistry();
-    CppUnit::Test *suite = registry.makeTest();
-    if (suite->countTestCases() == 0) {
-        std::cout << "No test cases specified for suite" << std::endl;
-        return 1;
-    }
-    CppUnit::TextUi::TestRunner runner;
-    runner.addTest(suite);
-    auto result = runner.run() ? 0 : 1;
-    return result;
-}
@@ -73,7 +73,7 @@ void print_help() {
 #if OPENDHT_PUSH_NOTIFICATIONS
              << "  stt [server_address] <device_key> Start the proxy client." << std::endl
              << "  rs  [token]                       Resubscribe to opendht." << std::endl
-              << "  rp  [token]                       Inject a push notification in Opendht." << std::endl
+              << "  rp  [push_notification]           Inject a push notification in Opendht." << std::endl
 #else
              << "  stt [server_address]              Start the proxy client." << std::endl
 #endif // OPENDHT_PUSH_NOTIFICATIONS
@@ -87,10 +87,8 @@ void print_help() {
              << "  cl <key> <token>      Cancel listen for <token> and <key>." << std::endl
              << "  p <key> <str>         Put string value at <key>." << std::endl
              << "  pp <key> <str>        Put string value at <key> (persistent version)." << std::endl
-              << "  cpp <key> <id>        Cancel persistent put operation for <key> and value <id>." << std::endl
              << "  s <key> <str>         Put string value at <key>, signed with our generated private key." << std::endl
-              << "  e <key> <dest> <str>  Put string value at <key>, encrypted for <dest> with its public key (if found)." << std::endl
-              << "  cc                    Trigger connectivity changed signal." << std::endl;
+              << "  e <key> <dest> <str>  Put string value at <key>, encrypted for <dest> with its public key (if found)." << std::endl;

 #ifdef OPENDHT_INDEXATION
    std::cout << std::endl << "Indexation operations on the DHT:" << std::endl
@@ -141,12 +139,6 @@ void cmd_loop(std::shared_ptr<DhtRunner>& dht, dht_params& params
            std::cout << dht->getNodesStats(AF_INET).toString() << std::endl;
            std::cout << "IPv6 stats:" << std::endl;
            std::cout << dht->getNodesStats(AF_INET6).toString() << std::endl;
-#if OPENDHT_PROXY_SERVER
-            for (const auto& proxy : proxies) {
-                std::cout << "Stats for proxy on port " << proxy.first << std::endl;
-                std::cout << "  " << proxy.second->getStats().toString() << std::endl;
-            }
-#endif
            continue;
        } else if (op == "lr") {
            std::cout << "IPv4 routing table:" << std::endl;
@@ -180,8 +172,11 @@ void cmd_loop(std::shared_ptr<DhtRunner>& dht, dht_params& params
            iss >> idstr;
            try {
                auto addr = splitPort(idstr);
-                if (not addr.first.empty() and addr.second.empty())
-                    addr.second = std::to_string(DHT_DEFAULT_PORT);
+                if (not addr.first.empty() and addr.second.empty()){
+                    std::stringstream ss;
+                    ss << DHT_DEFAULT_PORT;
+                    addr.second = ss.str();
+                }
                dht->bootstrap(addr.first.c_str(), addr.second.c_str());
            } catch (const std::exception& e) {
                std::cerr << e.what() << std::endl;
@@ -197,8 +192,6 @@ void cmd_loop(std::shared_ptr<DhtRunner>& dht, dht_params& params
                log::disableLogging(*dht);
            dht->setLogFilter(filter);
            continue;
-        } else if (op == "cc") {
-            dht->connectivityChanged();
        }
 #if OPENDHT_PROXY_SERVER
        else if (op == "pst") {
@@ -237,7 +230,14 @@ void cmd_loop(std::shared_ptr<DhtRunner>& dht, dht_params& params
 #if OPENDHT_PUSH_NOTIFICATIONS
        else if (op == "rp") {
            iss >> value;
-            dht->pushNotificationReceived({{"to", "dhtnode"}, {"token", value}});
+            dht->pushNotificationReceived({{"token", value}});
+            continue;
+        } else if (op == "re") {
+            iss >> value;
+            try {
+                unsigned token = std::stoul(value);
+                dht->resubscribe(token);
+            } catch (...) { }
            continue;
        }
 #endif // OPENDHT_PUSH_NOTIFICATIONS
@@ -254,7 +254,11 @@ void cmd_loop(std::shared_ptr<DhtRunner>& dht, dht_params& params
        }
        dht::InfoHash id;

-        if (false) {}
+        if (op == "cl") {
+            std::string hash, rem;
+            iss >> hash >> rem;
+            dht->cancelListen(dht::InfoHash(hash), std::stoul(rem));
+        }
 #ifdef OPENDHT_INDEXATION
        else if (op == "il" or op == "ii") {
            // Pht syntax
@@ -264,7 +268,7 @@ void cmd_loop(std::shared_ptr<DhtRunner>& dht, dht_params& params
                        return i.first == index;
                    }) == indexes.end();
            if (not index.size()) {
-                std::cerr << "You must enter the index name." << std::endl;
+                std::cout << "You must enter the index name." << std::endl;
                continue;
            } else if (new_index) {
                using namespace dht::indexation;
@@ -326,27 +330,14 @@ void cmd_loop(std::shared_ptr<DhtRunner>& dht, dht_params& params
        else if (op == "l") {
            std::string rem;
            std::getline(iss, rem);
-            auto token = dht->listen(id, [](const std::vector<std::shared_ptr<Value>>& values, bool expired) {
-                std::cout << "Listen: found " << values.size() << " values" << (expired ? " expired" : "") << std::endl;
-                for (const auto& value : values)
-                    std::cout << "\t" << *value << std::endl;
+            auto token = dht->listen(id, [](std::shared_ptr<Value> value) {
+                std::cout << "Listen: found value:" << std::endl;
+                std::cout << "\t" << *value << std::endl;
                return true;
            }, {}, dht::Where {std::move(rem)});
            auto t = token.get();
            std::cout << "Listening, token: " << t << std::endl;
        }
-        if (op == "cl") {
-            std::string rem;
-            iss >> rem;
-            size_t token;
-            try {
-                token = std::stoul(rem);
-            } catch(...) {
-                std::cerr << "Syntax: cl [key] [token]" << std::endl;
-                continue;
-            }
-            dht->cancelListen(id, token);
-        }
        else if (op == "p") {
            std::string v;
            iss >> v;
@@ -498,7 +489,6 @@ main(int argc, char **argv)

        dht::DhtRunner::Config config;
        config.dht_config.node_config.network = params.network;
-        config.dht_config.node_config.maintain_storage = false;
        config.dht_config.id = crt;
        config.threaded = true;
        config.proxy_server = params.proxyclient;
@@ -45,6 +45,29 @@
 #include <sstream>
 #include <fstream>

+/**
+ * Split "[host]:port" or "host:port" to pair<"host", "port">.
+ */
+std::pair<std::string, std::string>
+splitPort(const std::string& s) {
+    if (s.empty())
+        return {};
+    if (s[0] == '[') {
+        std::size_t closure = s.find_first_of(']');
+        std::size_t found = s.find_last_of(':');
+        if (closure == std::string::npos)
+            return {s, ""};
+        if (found == std::string::npos or found < closure)
+            return {s.substr(1,closure-1), ""};
+        return {s.substr(1,closure-1), s.substr(found+1)};
+    }
+    std::size_t found = s.find_last_of(':');
+    std::size_t first = s.find_first_of(':');
+    if (found == std::string::npos or found != first)
+        return {s, ""};
+    return {s.substr(0,found), s.substr(found+1)};
+}
+
 /*
 * The mapString shall have the following format:
 *
@@ -158,7 +181,7 @@ parseArgs(int argc, char **argv) {
            params.network = strtoul(optarg, nullptr, 0);
            break;
        case 'b':
-            params.bootstrap = dht::splitPort((optarg[0] == '=') ? optarg+1 : optarg);
+            params.bootstrap = splitPort((optarg[0] == '=') ? optarg+1 : optarg);
            if (not params.bootstrap.first.empty() and params.bootstrap.second.empty()) {
                params.bootstrap.second = std::to_string(DHT_DEFAULT_PORT);
            }