python: add cancelPut

.github/workflows/ccpp.yml

@@ -67,9 +67,9 @@ jobs:
     - name: restinio
       run: |
         mkdir restinio && cd restinio
-        wget https://github.com/aberaud/restinio/archive/bbaa034dbcc7555ce67df0f8a1475591a7441733.tar.gz
-        ls -l && tar -xzf bbaa034dbcc7555ce67df0f8a1475591a7441733.tar.gz
-        cd restinio-bbaa034dbcc7555ce67df0f8a1475591a7441733/dev
+        wget https://github.com/aberaud/restinio/archive/6fd08b65f6f15899dd0de3c801f6a5462b811c64.tar.gz
+        ls -l && tar -xzf 6fd08b65f6f15899dd0de3c801f6a5462b811c64.tar.gz
+        cd restinio-6fd08b65f6f15899dd0de3c801f6a5462b811c64/dev
         cmake -DCMAKE_INSTALL_PREFIX=/usr/local -DRESTINIO_TEST=OFF -DRESTINIO_SAMPLE=OFF \
                 -DRESTINIO_INSTALL_SAMPLES=OFF -DRESTINIO_BENCH=OFF -DRESTINIO_INSTALL_BENCHES=OFF \
                 -DRESTINIO_FIND_DEPS=ON -DRESTINIO_ALLOW_SOBJECTIZER=Off -DRESTINIO_USE_BOOST_ASIO=none .

.github/workflows/clang-analyzer.yml

@@ -18,9 +18,9 @@ jobs:
       - name: restinio
         run: |
           mkdir restinio && cd restinio \
-          && wget https://github.com/aberaud/restinio/archive/bbaa034dbcc7555ce67df0f8a1475591a7441733.tar.gz \
-          && ls -l && tar -xzf bbaa034dbcc7555ce67df0f8a1475591a7441733.tar.gz \
-          && cd restinio-bbaa034dbcc7555ce67df0f8a1475591a7441733/dev \
+          && wget https://github.com/aberaud/restinio/archive/6fd08b65f6f15899dd0de3c801f6a5462b811c64.tar.gz \
+          && ls -l && tar -xzf 6fd08b65f6f15899dd0de3c801f6a5462b811c64.tar.gz \
+          && cd restinio-6fd08b65f6f15899dd0de3c801f6a5462b811c64/dev \
           && cmake -DCMAKE_INSTALL_PREFIX=/usr -DRESTINIO_TEST=OFF -DRESTINIO_SAMPLE=OFF \
                   -DRESTINIO_INSTALL_SAMPLES=OFF -DRESTINIO_BENCH=OFF -DRESTINIO_INSTALL_BENCHES=OFF \
                   -DRESTINIO_FIND_DEPS=ON -DRESTINIO_ALLOW_SOBJECTIZER=Off -DRESTINIO_USE_BOOST_ASIO=none . \

CMakeLists.txt

@@ -14,8 +14,8 @@ include(FindPkgConfig)
 include(cmake/CheckAtomic.cmake)
 include(CTest)
 
-set (opendht_VERSION_MAJOR 2)
-set (opendht_VERSION_MINOR 6.0)
+set (opendht_VERSION_MAJOR 3)
+set (opendht_VERSION_MINOR 0.0)
 set (opendht_VERSION ${opendht_VERSION_MAJOR}.${opendht_VERSION_MINOR})
 set (PACKAGE_VERSION ${opendht_VERSION})
 set (VERSION "${opendht_VERSION}")

configure.ac

@@ -1,6 +1,6 @@
 dnl define macros
-m4_define([opendht_major_version], 2)
-m4_define([opendht_minor_version], 6)
+m4_define([opendht_major_version], 3)
+m4_define([opendht_minor_version], 0)
 m4_define([opendht_patch_version], 0)
 m4_define([opendht_version],
 		  [opendht_major_version.opendht_minor_version.opendht_patch_version])

docker/DockerfileDeps

@@ -18,9 +18,9 @@ RUN apt-get update && apt-get install -y \
 
 RUN echo "*** Downloading RESTinio ***" \
     && mkdir restinio && cd restinio \
-    && wget https://github.com/aberaud/restinio/archive/bbaa034dbcc7555ce67df0f8a1475591a7441733.tar.gz \
-    && ls -l && tar -xzf bbaa034dbcc7555ce67df0f8a1475591a7441733.tar.gz \
-    && cd restinio-bbaa034dbcc7555ce67df0f8a1475591a7441733/dev \
+    && wget https://github.com/aberaud/restinio/archive/6fd08b65f6f15899dd0de3c801f6a5462b811c64.tar.gz \
+    && ls -l && tar -xzf 6fd08b65f6f15899dd0de3c801f6a5462b811c64.tar.gz \
+    && cd restinio-6fd08b65f6f15899dd0de3c801f6a5462b811c64/dev \
     && cmake -DCMAKE_INSTALL_PREFIX=/usr -DRESTINIO_TEST=OFF -DRESTINIO_SAMPLE=OFF \
              -DRESTINIO_INSTALL_SAMPLES=OFF -DRESTINIO_BENCH=OFF -DRESTINIO_INSTALL_BENCHES=OFF \
              -DRESTINIO_FIND_DEPS=ON -DRESTINIO_ALLOW_SOBJECTIZER=Off -DRESTINIO_USE_BOOST_ASIO=none . \

docker/DockerfileDepsAlpine

@@ -13,9 +13,9 @@ RUN apk add --no-cache \
 
 RUN echo "*** Downloading RESTinio ***" \
     && mkdir restinio && cd restinio \
-    && wget https://github.com/aberaud/restinio/archive/bbaa034dbcc7555ce67df0f8a1475591a7441733.tar.gz \
-    && tar -xzf bbaa034dbcc7555ce67df0f8a1475591a7441733.tar.gz \
-    && cd restinio-bbaa034dbcc7555ce67df0f8a1475591a7441733/dev \
+    && wget https://github.com/aberaud/restinio/archive/6fd08b65f6f15899dd0de3c801f6a5462b811c64.tar.gz \
+    && tar -xzf 6fd08b65f6f15899dd0de3c801f6a5462b811c64.tar.gz \
+    && cd restinio-6fd08b65f6f15899dd0de3c801f6a5462b811c64/dev \
     && cmake -DCMAKE_INSTALL_PREFIX=/usr -DRESTINIO_TEST=OFF -DRESTINIO_SAMPLE=OFF \
              -DRESTINIO_INSTALL_SAMPLES=OFF -DRESTINIO_BENCH=OFF -DRESTINIO_INSTALL_BENCHES=OFF \
              -DRESTINIO_FIND_DEPS=ON -DRESTINIO_ALLOW_SOBJECTIZER=Off -DRESTINIO_USE_BOOST_ASIO=none . \

docker/DockerfileDepsLlvm

@@ -22,9 +22,9 @@ ENV CXX c++
 
 RUN echo "*** Downloading RESTinio ***" \
     && mkdir restinio && cd restinio \
-    && wget https://github.com/aberaud/restinio/archive/bbaa034dbcc7555ce67df0f8a1475591a7441733.tar.gz \
-    && ls -l && tar -xzf bbaa034dbcc7555ce67df0f8a1475591a7441733.tar.gz \
-    && cd restinio-bbaa034dbcc7555ce67df0f8a1475591a7441733/dev \
+    && wget https://github.com/aberaud/restinio/archive/6fd08b65f6f15899dd0de3c801f6a5462b811c64.tar.gz \
+    && ls -l && tar -xzf 6fd08b65f6f15899dd0de3c801f6a5462b811c64.tar.gz \
+    && cd restinio-6fd08b65f6f15899dd0de3c801f6a5462b811c64/dev \
     && cmake -DCMAKE_INSTALL_PREFIX=/usr -DRESTINIO_TEST=OFF -DRESTINIO_SAMPLE=OFF \
              -DRESTINIO_INSTALL_SAMPLES=OFF -DRESTINIO_BENCH=OFF -DRESTINIO_INSTALL_BENCHES=OFF \
              -DRESTINIO_FIND_DEPS=ON -DRESTINIO_ALLOW_SOBJECTIZER=Off -DRESTINIO_USE_BOOST_ASIO=none . \

meson.build

@@ -1,5 +1,5 @@
 project('opendht', 'c', 'cpp',
-    version: '2.6.0',
+    version: '3.0.0',
     default_options: [
         'cpp_std=c++17',
         'warning_level=3'

python/opendht.pyx

@@ -158,6 +158,15 @@ cdef class SockAddr(object):
         return self.toString().decode()
     def __repr__(self):
         return "<%s '%s'>" % (self.__class__.__name__, str(self))
+    @staticmethod
+    def resolve(str host, str service=None):
+        vals = []
+        result = cpp.SockAddr.resolve(host.encode(), service.encode() if service else b'')
+        for val in result:
+            a = SockAddr()
+            a._addr = val
+            vals.append(a)
+        return vals
 
 cdef class Node(_WithID):
     cdef shared_ptr[cpp.Node] _node
@@ -457,7 +466,7 @@ cdef class DhtConfig(object):
     cdef cpp.DhtRunnerConfig _config
     def __init__(self):
         self._config = cpp.DhtRunnerConfig()
-        self._config.threaded = True;
+        self._config.threaded = True
     def setIdentity(self, Identity id):
         self._config.dht_config.id = id._id
     def setBootstrapMode(self, bool bootstrap):
@@ -471,6 +480,19 @@ cdef class DhtConfig(object):
     def setRateLimit(self, ssize_t max_req_per_sec, ssize_t max_peer_req_per_sec):
         self._config.dht_config.node_config.max_req_per_sec = max_req_per_sec
         self._config.dht_config.node_config.max_peer_req_per_sec = max_peer_req_per_sec
+    def setProxyInfo(self, str proxy_server, str push_node_id="", str push_token="", str push_topic="", str push_platform="", Certificate server_ca = Certificate()):
+        self._config.proxy_server = proxy_server.encode()
+        self._config.push_node_id = push_node_id.encode()
+        self._config.push_token = push_token.encode()
+        self._config.push_topic = push_topic.encode()
+        self._config.push_platform = push_platform.encode()
+        self._config.server_ca = server_ca._cert
+    def setPeerDiscovery(self, bool peer_discovery, bool peer_publish):
+        self._config.peer_discovery = peer_discovery
+        self._config.peer_publish = peer_publish
+    def setBound(self, SockAddr bind4, SockAddr bind6):
+        self._config.bind4 = bind4._addr
+        self._config.bind6 = bind6._addr
 
 cdef class DhtRunner(_WithID):
     cdef cpp.shared_ptr[cpp.DhtRunner] thisptr
@@ -587,7 +609,7 @@ cdef class DhtRunner(_WithID):
                 while pending > 0:
                     lock.wait()
             return res
-    def put(self, InfoHash key, Value val, done_cb=None):
+    def put(self, InfoHash key, Value val, done_cb=None, permanent=False):
         """Publish a new value on the DHT at key.
 
         key     -- the DHT key where to put the value
@@ -597,7 +619,7 @@ cdef class DhtRunner(_WithID):
         if done_cb:
             cb_obj = {'done':done_cb}
             ref.Py_INCREF(cb_obj)
-            self.thisptr.get().put(key._infohash, val._value, cpp.bindDoneCb(done_callback, <void*>cb_obj))
+            self.thisptr.get().put(key._infohash, val._value, cpp.bindDoneCb(done_callback, <void*>cb_obj), cpp.time_point.max(), permanent)
         else:
             lock = threading.Condition()
             pending = 0
@@ -614,6 +636,10 @@ cdef class DhtRunner(_WithID):
                 while pending > 0:
                     lock.wait()
             return ok
+
+    def cancelPut(self, InfoHash key, Value val):
+        self.thisptr.get().cancelPut(key._infohash, val._value)
+
     def listen(self, InfoHash key, value_cb):
         t = ListenToken()
         t._h = key._infohash

python/opendht_cpp.pxd

@@ -21,12 +21,13 @@ from libcpp.string cimport string
 from libcpp.vector cimport vector
 from libcpp.utility cimport pair
 from libcpp.map cimport map
+from libcpp.memory cimport shared_ptr, make_shared
 from libc.string cimport const_char, const_uchar
 
 ctypedef uint16_t in_port_t
 ctypedef unsigned short int sa_family_t;
 
-cdef extern from "<iostream>" namespace "std::chrono" nogil:
+cdef extern from "<chrono>" namespace "std::chrono" nogil:
     cdef cppclass duration[ulong]:
         duration() except +
 
@@ -34,16 +35,6 @@ cdef extern from "<iostream>" namespace "std::chrono" nogil:
         duration seconds(uint64_t) except +
         duration seconds()
 
-cdef extern from "<memory>" namespace "std" nogil:
-    cdef cppclass shared_ptr[T]:
-        shared_ptr() except +
-        shared_ptr(T*) except +
-        T* get()
-        T operator*()
-        bool operator bool() const
-        void reset(T*)
-    shared_ptr[T] make_shared[T](...) except +
-
 cdef extern from "<functional>" namespace "std" nogil:
     cdef cppclass hash[T]:
         hash() except +
@@ -59,6 +50,13 @@ cdef extern from "<future>" namespace "std" nogil:
         bool valid() const
         shared_future[T] share()
 
+cdef extern from "opendht/utils.h" namespace "dht":
+    cdef cppclass time_point:
+        @staticmethod
+        time_point min() const
+        @staticmethod
+        time_point max() const
+
 cdef extern from "opendht/infohash.h" namespace "dht":
     cdef cppclass InfoHash:
         InfoHash() except +
@@ -88,6 +86,8 @@ cdef extern from "opendht/sockaddr.h" namespace "dht":
         bool isLoopback() const
         bool isPrivate() const
         bool isUnspecified() const
+        @staticmethod
+        vector[SockAddr] resolve(string host, string service) except +
 
 ctypedef vector[uint8_t] Blob
 
@@ -239,6 +239,18 @@ cdef extern from "opendht/dhtrunner.h" namespace "dht":
         cppclass Config:
             SecureDhtConfig dht_config
             bool threaded
+            string proxy_server
+            string push_node_id
+            string push_token
+            string push_topic
+            string push_platform
+            bool peer_discovery
+            bool peer_publish
+            shared_ptr[Certificate] server_ca
+            Identity client_identity
+            SockAddr bind4
+            SockAddr bind6
+
         InfoHash getId() const
         InfoHash getNodeId() const
         void bootstrap(const_char*, const_char*)
@@ -254,7 +266,8 @@ cdef extern from "opendht/dhtrunner.h" namespace "dht":
         string getRoutingTablesLog(sa_family_t af) const
         string getSearchesLog(sa_family_t af) const
         void get(InfoHash key, GetCallback get_cb, DoneCallback done_cb, nullptr_t f, Where w)
-        void put(InfoHash key, shared_ptr[Value] val, DoneCallback done_cb)
+        void put(InfoHash key, shared_ptr[Value] val, DoneCallback done_cb, time_point created, bool permanent)
+        void cancelPut(InfoHash key, shared_ptr[Value] val)
         ListenToken listen(InfoHash key, ValueCallback get_cb)
         void cancelListen(InfoHash key, SharedListenToken token)
         vector[unsigned] getNodeMessageStats(bool i)

src/dht_proxy_client.cpp

@@ -491,7 +491,7 @@ void
 DhtProxyClient::doPut(const InfoHash& key, Sp<Value> val, DoneCallbackSimple cb, time_point /*created*/, bool permanent)
 {
     if (logger_)
-        logger_->d("[proxy:client] [put] [search %s] executing for %s", key.to_c_str(), val->toString().c_str());
+        logger_->debug("[proxy:client] [put] [search {}] executing for {}", key, val->toString());
 
     try {
         auto request = buildRequest("/key/" + key.toString());
@@ -539,12 +539,12 @@ DhtProxyClient::doPut(const InfoHash& key, Sp<Value> val, DoneCallbackSimple cb,
                         }
                     } else {
                         if (logger_)
-                            logger_->e("[proxy:client] [status] failed to parse value from  server", response.status_code);
+                            logger_->error("[proxy:client] [put] failed to parse value from server: {}", err);
                     }
                 }
             } else {
                 if (logger_)
-                    logger_->e("[proxy:client] [status] failed with code=%i", response.status_code);
+                    logger_->error("[proxy:client] [put] failed with code={:d}", response.status_code);
                 if (not response.aborted and response.status_code == 0)
                     opFailed();
             }
@@ -563,7 +563,8 @@ DhtProxyClient::doPut(const InfoHash& key, Sp<Value> val, DoneCallbackSimple cb,
     }
     catch (const std::exception &e){
         if (logger_)
-            logger_->e("[proxy:client] [put %s] error: %s", key.to_c_str(), e.what());
+            logger_->error("[proxy:client] [put {}] error: {}", key, e.what());
+        opFailed();
     }
 }
 
@@ -1154,7 +1155,7 @@ DhtProxyClient::restartListeners(const asio::error_code &ec)
 
     std::lock_guard<std::mutex> lock(searchLock_);
     for (auto& search : searches_) {
-        auto key = search.first;
+        const auto& key = search.first;
         for (auto& put : search.second.puts) {
             doPut(key, put.second.value, [ok = put.second.ok](bool result){
                 *ok = result;
@@ -1166,6 +1167,10 @@ DhtProxyClient::restartListeners(const asio::error_code &ec)
             put.second.refreshPutTimer->async_wait(std::bind(&DhtProxyClient::handleRefreshPut, this,
                                                    std::placeholders::_1, key, put.first));
         }
+        // Restart failed pending puts
+        for (auto& pput : search.second.pendingPuts) {
+            doPut(key, pput, {}, time_point::max(), true);
+        }
     }
     if (not deviceKey_.empty()) {
         if (logger_)

src/dht_proxy_server.cpp

@@ -1080,6 +1080,7 @@ DhtProxyServer::sendPushNotification(const std::string& token, Json::Value&& jso
             request->set_body(Json::writeString(jsonBuilder_, std::move(json)));
         } else {
             // NOTE: see https://github.com/appleboy/gorush
+            auto isResubscribe = json.isMember("timeout");
             Json::Value notification(Json::objectValue);
             Json::Value tokens(Json::arrayValue);
             tokens[0] = token;
@@ -1095,12 +1096,13 @@ DhtProxyServer::sendPushNotification(const std::string& token, Json::Value&& jso
                 notification["expiration"] = exp;
                 if (!topic.empty())
                     notification["topic"] = topic;
-                if (highPriority) {
+                if (highPriority || isResubscribe) {
                     Json::Value alert(Json::objectValue);
                     alert["title"]="hello";
                     notification["push_type"] = "alert";
                     notification["alert"] = alert;
                     notification["mutable_content"] = true;
+                    notification["priority"] = "high";
                 } else {
                     notification["push_type"] = "background";
                     notification["content_available"] = true;