diff --git a/include/opendht/callbacks.h b/include/opendht/callbacks.h index bbf3a2f1..683755aa 100644 --- a/include/opendht/callbacks.h +++ b/include/opendht/callbacks.h @@ -130,9 +130,7 @@ using ShutdownCallback = std::function; using CertificateStoreQuery = std::function>(const InfoHash& pk_id)>; typedef bool (*GetCallbackRaw)(std::shared_ptr, void *user_data); - -OPENDHT_PUBLIC GetCallbackSimple bindGetCb(const GetCallbackRaw& raw_cb, void* user_data); -OPENDHT_PUBLIC GetCallback bindGetCb(const GetCallbackSimple& cb); +typedef bool (*ValueCallbackRaw)(std::shared_ptr, bool expired, void *user_data); using DoneCallback = std::function>& nodes)>; typedef void (*DoneCallbackRaw)(bool, std::vector>*, void *user_data); @@ -142,6 +140,9 @@ typedef bool (*FilterRaw)(const Value&, void *user_data); using DoneCallbackSimple = std::function; +OPENDHT_PUBLIC GetCallbackSimple bindGetCb(const GetCallbackRaw& raw_cb, void* user_data); +OPENDHT_PUBLIC GetCallback bindGetCb(const GetCallbackSimple& cb); +OPENDHT_PUBLIC ValueCallback bindValueCb(const ValueCallbackRaw& raw_cb, void* user_data); OPENDHT_PUBLIC ShutdownCallback bindShutdownCb(const ShutdownCallbackRaw& shutdown_cb_raw, void* user_data); OPENDHT_PUBLIC DoneCallback bindDoneCb(DoneCallbackSimple donecb); OPENDHT_PUBLIC DoneCallback bindDoneCb(const DoneCallbackRaw& raw_cb, void* user_data); diff --git a/python/opendht.pyx b/python/opendht.pyx index 538276b8..c0cc8d53 100644 --- a/python/opendht.pyx +++ b/python/opendht.pyx @@ -65,6 +65,14 @@ cdef inline bool get_callback(shared_ptr[cpp.Value] value, void *user_data) with pv._value = value return cb(pv) if not f or f(pv) else True +cdef inline bool value_callback(shared_ptr[cpp.Value] value, bool expired, void *user_data) with gil: + cbs = user_data + cb = cbs['valcb'] + f = cbs['filter'] if 'filter' in cbs else None + pv = Value() + pv._value = value + return cb(pv, expired) if not f or f(pv) else True + cdef inline void done_callback(bool done, cpp.vector[shared_ptr[cpp.Node]]* nodes, void *user_data) with gil: node_ids = [] for n in deref(nodes): @@ -591,14 +599,14 @@ cdef class DhtRunner(_WithID): while pending > 0: lock.wait() return ok - def listen(self, InfoHash key, get_cb): + def listen(self, InfoHash key, value_cb): t = ListenToken() t._h = key._infohash - cb_obj = {'get':get_cb} + 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.bindGetCb(get_callback, cb_obj)).share() + t._t = self.thisptr.get().listen(t._h, cpp.bindValueCb(value_callback, cb_obj)).share() return t def cancelListen(self, ListenToken token): self.thisptr.get().cancelListen(token._h, token._t) diff --git a/python/opendht_cpp.pxd b/python/opendht_cpp.pxd index fa5c69f9..34843eb1 100644 --- a/python/opendht_cpp.pxd +++ b/python/opendht_cpp.pxd @@ -185,6 +185,7 @@ cdef extern from "opendht/node.h" namespace "dht": cdef extern from "opendht/callbacks.h" namespace "dht": ctypedef void (*ShutdownCallbackRaw)(void *user_data) ctypedef bool (*GetCallbackRaw)(shared_ptr[Value] values, void *user_data) + ctypedef bool (*ValueCallbackRaw)(shared_ptr[Value] values, bool expired, void *user_data) ctypedef void (*DoneCallbackRaw)(bool done, vector[shared_ptr[Node]]* nodes, void *user_data) ctypedef void (*DoneCallbackSimpleRaw)(bool done, void *user_data) @@ -192,6 +193,8 @@ cdef extern from "opendht/callbacks.h" namespace "dht": ShutdownCallback() except + cppclass GetCallback: GetCallback() except + + cppclass ValueCallback: + ValueCallback() except + cppclass DoneCallback: DoneCallback() except + cppclass DoneCallbackSimple: @@ -199,6 +202,7 @@ cdef extern from "opendht/callbacks.h" namespace "dht": cdef ShutdownCallback bindShutdownCb(ShutdownCallbackRaw cb, void *user_data) cdef GetCallback bindGetCb(GetCallbackRaw cb, void *user_data) + cdef ValueCallback bindValueCb(ValueCallbackRaw cb, void *user_data) cdef DoneCallback bindDoneCb(DoneCallbackRaw cb, void *user_data) cdef DoneCallbackSimple bindDoneCbSimple(DoneCallbackSimpleRaw cb, void *user_data) @@ -234,7 +238,7 @@ cdef extern from "opendht/dhtrunner.h" namespace "dht": 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) - ListenToken listen(InfoHash key, GetCallback get_cb) + ListenToken listen(InfoHash key, ValueCallback get_cb) void cancelListen(InfoHash key, SharedListenToken token) vector[unsigned] getNodeMessageStats(bool i) diff --git a/python/tools/network_monitor.py b/python/tools/network_monitor.py index 627ba883..074bee22 100755 --- a/python/tools/network_monitor.py +++ b/python/tools/network_monitor.py @@ -43,7 +43,7 @@ loop = asyncio.get_event_loop() pending_tests = {} keys = [dht.InfoHash.getRandom() for _ in range(args.num_ops)] -def listen_cb(key, val): +def listen_cb(key, val, expired): global pending_tests kstr = str(key) if kstr in pending_tests: @@ -54,7 +54,7 @@ def listen_cb(key, val): return True def listen(key): - node1.listen(key, lambda v: loop.call_soon_threadsafe(listen_cb, key, v)) + node1.listen(key, lambda v, e: loop.call_soon_threadsafe(listen_cb, key, v, e)) for key in keys: listen(key) diff --git a/python/tools/pingpong.py b/python/tools/pingpong.py index f8dca3c3..23b5b145 100755 --- a/python/tools/pingpong.py +++ b/python/tools/pingpong.py @@ -54,8 +54,8 @@ def pong(node, h): loop.call_soon_threadsafe(ping, node, h) return True -ping_node.listen(loc_ping, lambda v: pong(pong_node, loc_pong)) -pong_node.listen(loc_pong, lambda v: pong(ping_node, loc_ping)) +ping_node.listen(loc_ping, lambda v, e: pong(pong_node, loc_pong) if not e else True) +pong_node.listen(loc_pong, lambda v, e: pong(ping_node, loc_ping) if not e else True) ping(pong_node, loc_ping) diff --git a/src/callbacks.cpp b/src/callbacks.cpp index 7dd96361..8d85155f 100644 --- a/src/callbacks.cpp +++ b/src/callbacks.cpp @@ -24,6 +24,18 @@ bindGetCb(const GetCallbackSimple& cb) }; } +ValueCallback +bindValueCb(const ValueCallbackRaw& raw_cb, void* user_data) +{ + if (not raw_cb) return {}; + return [=](const std::vector>& values, bool expired) { + for (const auto& v : values) + if (not raw_cb(v, expired, user_data)) + return false; + return true; + }; +} + ShutdownCallback bindShutdownCb(const ShutdownCallbackRaw& shutdown_cb_raw, void* user_data) {