diff --git a/python/opendht.pyx b/python/opendht.pyx index 1c27f947..47b63540 100644 --- a/python/opendht.pyx +++ b/python/opendht.pyx @@ -539,7 +539,7 @@ cdef class TrustList(object): cdef class ListenToken(object): cdef cpp.InfoHash _h cdef cpp.shared_future[size_t] _t - _cb = dict() + cdef bool _cancelled cdef class Identity(object): cdef cpp.Identity _id @@ -673,15 +673,18 @@ cdef class DhtRunner(_WithID): host_bytes = host.encode() port_bytes = port.encode() if port else b'4222' self.thisptr.get().bootstrap(host_bytes, port_bytes) - def run(self, Identity id=None, is_bootstrap=False, cpp.in_port_t port=0, str ipv4="", str ipv6="", DhtConfig config=DhtConfig()): - if id: - config.setIdentity(id) + def run(self, Identity id=None, is_bootstrap=False, cpp.in_port_t port=0, str ipv4="", str ipv6="", DhtConfig config=None): + cdef cpp.DhtRunnerConfig run_config = config._config if config is not None else cpp.DhtRunnerConfig() + if id is not None: + run_config.dht_config.id = id._id + if is_bootstrap: + run_config.dht_config.node_config.is_bootstrap = True if ipv4 or ipv6: bind4 = ipv4.encode() if ipv4 else b'' bind6 = ipv6.encode() if ipv6 else b'' - self.thisptr.get().run(bind4, bind6, str(port).encode(), config._config) + self.thisptr.get().run(bind4, bind6, str(port).encode(), run_config) else: - self.thisptr.get().run(port, config._config) + self.thisptr.get().run(port, run_config) def join(self): self.thisptr.get().join() def shutdown(self, shutdown_cb=None): @@ -849,15 +852,14 @@ cdef class DhtRunner(_WithID): t = ListenToken() t._h = key._infohash cb_obj = {'valcb':value_cb} - t._cb['cb'] = cb_obj - # avoid the callback being destructed if the token is destroyed - ref.Py_INCREF(cb_obj) - t._t = self.thisptr.get().listen(t._h, cpp.bindValueCb(value_callback, cb_obj)).share() + t._t = self.thisptr.get().listen(t._h, cpp.bindPythonValueCb(value_callback, cb_obj)).share() return t def cancelListen(self, ListenToken token): - self.thisptr.get().cancelListen(token._h, token._t) - ref.Py_DECREF(token._cb['cb']) - # fixme: not thread safe + if token is None: + raise TypeError("token must be a ListenToken") + if not token._cancelled and token._t.valid(): + self.thisptr.get().cancelListen(token._h, token._t) + token._cancelled = True cdef class IndexValue(object): cdef cpp.shared_ptr[cpp.IndexValue] _value diff --git a/python/opendht_cpp.pxd b/python/opendht_cpp.pxd index 5953c18e..fef884fb 100644 --- a/python/opendht_cpp.pxd +++ b/python/opendht_cpp.pxd @@ -243,9 +243,35 @@ cdef extern from "opendht/callbacks.h" namespace "dht": Config node_config Identity id +cdef extern from *: + """ + #include + #include + + static dht::ValueCallback opendht_bind_python_value_cb( + dht::ValueCallbackRaw callback, PyObject* data) + { + Py_INCREF(data); + auto owner = std::shared_ptr(data, [](PyObject* object) { + auto state = PyGILState_Ensure(); + Py_DECREF(object); + PyGILState_Release(state); + }); + return [callback, owner = std::move(owner)]( + const std::vector>& values, bool expired) { + for (const auto& value : values) + if (!callback(value, expired, owner.get())) + return false; + return true; + }; + } + """ + ValueCallback bindPythonValueCb "opendht_bind_python_value_cb"(ValueCallbackRaw callback, object data) except + + +ctypedef future[size_t] ListenToken +ctypedef shared_future[size_t] SharedListenToken + cdef extern from "opendht/dhtrunner.h" namespace "dht": - ctypedef future[size_t] ListenToken - ctypedef shared_future[size_t] SharedListenToken cdef cppclass DhtRunner: DhtRunner() except + cppclass Config: @@ -285,8 +311,8 @@ cdef extern from "opendht/dhtrunner.h" namespace "dht": void putEncrypted(InfoHash key, PkId to, shared_ptr[Value] val, DoneCallback done_cb, bool permanent) void putEncrypted(InfoHash key, shared_ptr[PublicKey] to, shared_ptr[Value] val, DoneCallback done_cb, bool permanent) void cancelPut(InfoHash key, shared_ptr[Value] val) - ListenToken listen(InfoHash key, ValueCallback get_cb) - void cancelListen(InfoHash key, SharedListenToken token) + ListenToken listen(InfoHash key, ValueCallback get_cb) except + + void cancelListen(InfoHash key, SharedListenToken token) except + vector[unsigned] getNodeMessageStats(bool i) ctypedef DhtRunner.Config DhtRunnerConfig