From af3fd618bf814e8e44d749a1605aaf4bdce8dd32 Mon Sep 17 00:00:00 2001 From: Adrien Beraud Date: Mon, 26 Oct 2026 00:17:33 -0400 Subject: [PATCH] python: expose observability and projected query APIs --- python/README.md | 70 ++++++++++++ python/opendht.pyx | 235 ++++++++++++++++++++++++++++++++++++++--- python/opendht/aio.py | 90 ++++++++++++---- python/opendht_cpp.pxd | 115 +++++++++++++++++--- 4 files changed, 463 insertions(+), 47 deletions(-) diff --git a/python/README.md b/python/README.md index 84cc6fcd..fcbddf17 100644 --- a/python/README.md +++ b/python/README.md @@ -161,6 +161,76 @@ value.pushType = "audioCall" Set metadata before publishing the value. Numeric values outside the native integer range raise `OverflowError`. +## Observability + +`getNodeInfo()` returns a read-only `NodeInfo` snapshot with node identifiers, +IPv4/IPv6 routing statistics, bound ports, pending-operation count, and local +and remote storage counters. `getNodesStats(socket.AF_INET)` and +`getNodesStats(socket.AF_INET6)` return `NodeStats` snapshots. Snapshots remain +valid after the runner stops. `getPublicAddress()` returns known public +`SockAddr` objects; pass an address family to restrict the results, or use +`getPublicAddressStr()` for strings. These lists may be empty before discovery. +`getNodeMessageStats(incoming=True)` selects incoming message counters. + +```python +info = node.getNodeInfo() +print(info.node_id, info.bound4, info.ipv4.good_nodes, info.storage_size) +print(node.getPublicAddressStr()) + +node.setOnStatusChanged(lambda ipv4, ipv6: print(ipv4.name, ipv6.name)) +node.setOnStatusChanged(None) +``` + +The status callback receives two `NodeStatus` values: `Disconnected`, +`Connecting`, or `Connected`. Registration replaces the previous callback and +persists across restarts; `None` disables it. Core callbacks run on the DHT +thread while native locks are held: do not call snapshot getters, blocking +operations, or `join()` from them. The asyncio wrapper schedules status +callbacks on its event-loop thread, where snapshot getters are safe to use. + +## Queries And Filtered Listeners + +`query()` retrieves metadata rather than complete value payloads. Each row is +a dictionary keyed by `ValueField`. Network projections support `Id`, +`ValueType`, `OwnerPk`, and `UserType`; values are integers, an `InfoHash` owner +fingerprint, or a UTF-8 string respectively. Explicit sequence-number +projections are not supported by the native wire format. + +```python +selection = dht.Select().field(dht.ValueField.Id).field(dht.ValueField.UserType) +query = dht.Query() +query.buildFrom(selection, dht.Where().userType("record")) +rows = node.query(key, query=query) + +query = dht.Query("SELECT id,user_type WHERE user_type=record") +node.query(key, query_cb=lambda row: print(row) or True, + done_cb=lambda ok, nodes: print(ok), query=query) +``` + +Without an explicit query, all four supported metadata fields are selected. +Supplying `query_cb` makes the core call nonblocking; returning `False` stops +delivery. The blocking core API returns the collected rows, as `get()` does. +The asyncio API offers `query()` as a context-managed async iterator and +`await queryAll()` as a collector; these raise `RuntimeError` on failure. + +```python +rows = await node.queryAll(key, query=query) +with node.query(key, query=query) as results: + async for row in results: + print(row[dht.ValueField.Id]) +``` + +Both core and asyncio `listen()` accept optional `where` and `filter` arguments. +`Where` is sent to the DHT to narrow the subscription; the Python predicate +runs locally. Existing two-argument core listeners are unchanged. + +```python +token = node.listen(key, lambda value, expired: True, + where=dht.Where().userType("record"), + filter=lambda value: len(value.data) < 1024) +node.cancelListen(token) +``` + ## Documentation For more information, examples and documentation, or to repport issues, visit: https://opendht.net diff --git a/python/opendht.pyx b/python/opendht.pyx index bb0a4d79..a348ce49 100644 --- a/python/opendht.pyx +++ b/python/opendht.pyx @@ -17,6 +17,7 @@ from libcpp.memory cimport shared_ptr, unique_ptr from cython.operator cimport dereference as deref, preincrement as inc, predecrement as dec from cpython cimport ref from datetime import timedelta +from enum import IntEnum as _IntEnum from operator import index as _index cimport opendht_cpp as cpp @@ -24,6 +25,8 @@ cimport opendht_cpp as cpp ctypedef cpp.Pht* _PhtPtr import threading +import socket as _socket +import weakref as _weakref cdef inline void _initialize_gnutls(): cdef int err = cpp.gnutls_global_init() @@ -42,6 +45,58 @@ _initialize_gnutls() def version() -> str: return cpp.version().decode() +class NodeStatus(_IntEnum): + Disconnected = 0 + Connecting = 1 + Connected = 2 + +class ValueField(_IntEnum): + Id = 1 + ValueType = 2 + OwnerPk = 3 + SeqNum = 4 + UserType = 5 + +class _StatusCallbackState: + __slots__ = ("callback", "__weakref__") + def __init__(self): + self.callback = None + +cdef inline void status_callback(int ipv4, int ipv6, void* user_data) noexcept with gil: + state = (user_data)() + if state is None: + return + callback = state.callback + if callback is not None: + callback(NodeStatus(ipv4), NodeStatus(ipv6)) + +cdef inline bool query_callback(shared_ptr[cpp.FieldValueIndex] value, void* user_data) noexcept with gil: + cdef cpp.Blob blob + row = {} + for entry in value.get().index: + field = ValueField(entry.first) + if field == ValueField.OwnerPk: + owner = InfoHash() + owner._infohash = entry.second.getHash() + row[field] = owner + elif field == ValueField.UserType: + blob = entry.second.getBlob() + row[field] = (blob.data())[:blob.size()].decode() + else: + row[field] = entry.second.getInt() + return (user_data)['query'](row) + +cdef inline void query_done_callback(bool done, cpp.vector[shared_ptr[cpp.Node]]* nodes, void* user_data) noexcept with gil: + callback = (user_data)['done'] + if callback is not None: + entries = [] + for node in deref(nodes): + entry = NodeEntry() + entry._v.first = node.get().getId() + entry._v.second = node + entries.append(entry) + callback(done, entries) + cdef inline void lookup_callback(cpp.vector[cpp.shared_ptr[cpp.IndexValue]]* values, cpp.Prefix* p, void *user_data) noexcept with gil: cbs = user_data if 'lookup' in cbs and cbs['lookup']: @@ -241,6 +296,78 @@ cdef class Node(_WithID): raise ValueError("Node is not initialized") return self._node.get().isExpired() +cdef class NodeStats(object): + cdef cpp.NodeStats _stats + @property + def good_nodes(self): + return self._stats.good_nodes + @property + def dubious_nodes(self): + return self._stats.dubious_nodes + @property + def cached_nodes(self): + return self._stats.cached_nodes + @property + def incoming_nodes(self): + return self._stats.incoming_nodes + @property + def table_depth(self): + return self._stats.table_depth + @property + def searches(self): + return self._stats.searches + @property + def node_cache_size(self): + return self._stats.node_cache_size + def getKnownNodes(self): + return self._stats.getKnownNodes() + def getNetworkSizeEstimation(self): + return self._stats.getNetworkSizeEstimation() + +cdef class NodeInfo(object): + cdef cpp.NodeInfo _info + @property + def id(self): + result = InfoHash() + result._infohash = self._info.id + return result + @property + def node_id(self): + result = InfoHash() + result._infohash = self._info.node_id + return result + @property + def ipv4(self): + result = NodeStats() + result._stats = self._info.ipv4 + return result + @property + def ipv6(self): + result = NodeStats() + result._stats = self._info.ipv6 + return result + @property + def ongoing_ops(self): + return self._info.ongoing_ops + @property + def storage_values(self): + return self._info.storage_values + @property + def storage_size(self): + return self._info.storage_size + @property + def local_storage_values(self): + return self._info.local_storage_values + @property + def local_storage_size(self): + return self._info.local_storage_size + @property + def bound4(self): + return self._info.bound4 + @property + def bound6(self): + return self._info.bound6 + cdef class NodeEntry(_WithID): cdef cpp.pair[cpp.InfoHash, shared_ptr[cpp.Node]] _v def getId(self): @@ -274,8 +401,11 @@ cdef class Select(object): return self._select.toString().decode() def isSatisfiedBy(self, Select os): return self._select.isSatisfiedBy(os._select) - def field(self, int field): - self._select.field( field) + def field(self, field): + cdef int field_id = ValueField(_index(field)) + if field_id == ValueField.SeqNum: + raise ValueError("The native wire format does not support sequence-number projections") + self._select.field(field_id) return self cdef class Where(object): @@ -750,8 +880,10 @@ cdef class DhtConfig(object): cdef class DhtRunner(_WithID): cdef cpp.shared_ptr[cpp.DhtRunner] thisptr cdef map[_PhtPtr, shared_ptr[cpp.Pht]] _indexes + cdef object _status_callback def __cinit__(self): self.thisptr.reset(new cpp.DhtRunner()) + self._status_callback = _StatusCallbackState() def __dealloc__(self): with nogil: if self.thisptr: @@ -761,19 +893,51 @@ cdef class DhtRunner(_WithID): def getId(self) -> InfoHash: h = InfoHash() if self.thisptr: - h._infohash = self.thisptr.get().getId() + with nogil: + h._infohash = self.thisptr.get().getId() return h def getPublicKey(self) -> PublicKey: if not self.thisptr: raise ValueError("DhtRunner is not initialized") pk = PublicKey() - pk._key = self.thisptr.get().getPublicKey() + with nogil: + pk._key = self.thisptr.get().getPublicKey() return pk def getNodeId(self) -> InfoHash: h = InfoHash() if self.thisptr: - h._infohash = self.thisptr.get().getNodeId() + with nogil: + h._infohash = self.thisptr.get().getNodeId() return h + def getNodeInfo(self) -> NodeInfo: + result = NodeInfo() + with nogil: + result._info = self.thisptr.get().getNodeInfo() + return result + def getNodesStats(self, af=_socket.AF_INET) -> NodeStats: + if af not in (_socket.AF_INET, _socket.AF_INET6): + raise ValueError("af must be AF_INET or AF_INET6") + info = self.getNodeInfo() + return info.ipv4 if af == _socket.AF_INET else info.ipv6 + def getPublicAddress(self, cpp.sa_family_t af=0): + if af not in (0, _socket.AF_INET, _socket.AF_INET6): + raise ValueError("af must be AF_UNSPEC, AF_INET or AF_INET6") + cdef cpp.vector[cpp.SockAddr] addresses + with nogil: + addresses = self.thisptr.get().getPublicAddress(af) + result = [] + for address in addresses: + value = SockAddr() + value._addr = address + result.append(value) + return result + def getPublicAddressStr(self, cpp.sa_family_t af=0): + return [str(address) for address in self.getPublicAddress(af)] + def setOnStatusChanged(self, callback=None): + """Replace the DHT-thread status callback, or disable it with None.""" + if callback is not None and not callable(callback): + raise TypeError("callback must be callable or None") + self._status_callback.callback = callback def ping(self, SockAddr addr, done_cb=None): if done_cb: cb_obj = {'done':done_cb} @@ -808,9 +972,12 @@ cdef class DhtRunner(_WithID): 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(), run_config) + cpp.runPythonRunnerAddress(deref(self.thisptr.get()), bind4, bind6, + str(port).encode(), run_config, + status_callback, _weakref.ref(self._status_callback)) else: - self.thisptr.get().run(port, run_config) + cpp.runPythonRunner(deref(self.thisptr.get()), port, run_config, + status_callback, _weakref.ref(self._status_callback)) def join(self): with nogil: self.thisptr.get().join() @@ -831,17 +998,29 @@ cdef class DhtRunner(_WithID): return self.thisptr.get().isRunning() def getBound(self, cpp.sa_family_t af = 0): s = SockAddr() - s._addr = self.thisptr.get().getBound(af) + with nogil: + s._addr = self.thisptr.get().getBound(af) return s def getStorageLog(self): - return self.thisptr.get().getStorageLog().decode() + cdef cpp.string result + with nogil: + result = self.thisptr.get().getStorageLog() + return result.decode() def getRoutingTablesLog(self, cpp.sa_family_t af): - return self.thisptr.get().getRoutingTablesLog(af).decode() + cdef cpp.string result + with nogil: + result = self.thisptr.get().getRoutingTablesLog(af) + return result.decode() def getSearchesLog(self, cpp.sa_family_t af): - return self.thisptr.get().getSearchesLog(af).decode() - def getNodeMessageStats(self): + cdef cpp.string result + with nogil: + result = self.thisptr.get().getSearchesLog(af) + return result.decode() + def getNodeMessageStats(self, bool incoming=False): stats = [] - cdef cpp.vector[unsigned] res = self.thisptr.get().getNodeMessageStats(False) + cdef cpp.vector[unsigned] res + with nogil: + res = self.thisptr.get().getNodeMessageStats(incoming) for n in res: stats.append(n) return stats @@ -884,6 +1063,25 @@ cdef class DhtRunner(_WithID): lock.wait() return res + def query(self, InfoHash key not None, query_cb=None, done_cb=None, Query query=None): + """Return projected metadata rows, or stream them through query_cb.""" + if query is None: + query = Query("SELECT id,value_type,owner_pk,user_type") + if query_cb is None: + completed = threading.Event() + rows = [] + def on_row(row): + rows.append(row) + return True + self.query(key, on_row, lambda ok, nodes: completed.set(), query) + completed.wait() + return rows + if not callable(query_cb) or (done_cb is not None and not callable(done_cb)): + raise TypeError("query callbacks must be callable") + callbacks = {'query': query_cb, 'done': done_cb} + cpp.queryPython(deref(self.thisptr.get()), key._infohash, query_callback, + query_done_callback, callbacks, query._query) + def put(self, InfoHash key, Value val, done_cb=None, permanent=False): """Publish a new value on the DHT at key. @@ -976,11 +1174,16 @@ cdef class DhtRunner(_WithID): def cancelPut(self, InfoHash key, Value val): self.thisptr.get().cancelPut(key._infohash, val._value) - def listen(self, InfoHash key, value_cb): + def listen(self, InfoHash key not None, value_cb, filter=None, Where where=None): + if not callable(value_cb) or (filter is not None and not callable(filter)): + raise TypeError("listener callbacks must be callable") + if where is None: + where = Where() t = ListenToken() t._h = key._infohash - cb_obj = {'valcb':value_cb} - t._t = self.thisptr.get().listen(t._h, cpp.bindPythonValueCb(value_callback, cb_obj)).share() + cb_obj = {'valcb':value_cb, 'filter':filter} + t._t = self.thisptr.get().listen(t._h, cpp.bindPythonValueCb(value_callback, cb_obj), + cpp.nullptr, where._where).share() return t def cancelListen(self, ListenToken token): if token is None: diff --git a/python/opendht/aio.py b/python/opendht/aio.py index 6db9b605..f4dd997c 100644 --- a/python/opendht/aio.py +++ b/python/opendht/aio.py @@ -3,12 +3,15 @@ from __future__ import annotations import asyncio -from typing import Any, Callable, Iterable, List, Optional, Tuple +from typing import Any, Callable, Generic, Iterable, List, Optional, Tuple, TypeVar from . import _core as _dht # Re-export core types for convenience from ._core import InfoHash, Value, ValueType, Where, SockAddr, PublicKey, PkId, DhtConfig, Identity +from ._core import NodeInfo, NodeStats, NodeStatus, Query, Select, ValueField + +_Result = TypeVar("_Result") def _set_result_if_pending(future: asyncio.Future[Any], result: Any) -> None: @@ -24,7 +27,11 @@ class _Listener: __slots__ = ("_runner", "_token", "_loop", "_queue", "_closed") - def __init__(self, runner: DhtRunner, key: InfoHash) -> None: + def __init__( + self, runner: DhtRunner, key: InfoHash, + where: Optional[Where] = None, + filter: Optional[Callable[[Value], bool]] = None, + ) -> None: self._runner = runner self._loop = runner._loop self._queue: asyncio.Queue[Tuple[Value, bool]] = asyncio.Queue() @@ -36,7 +43,7 @@ class _Listener: self._loop.call_soon_threadsafe(self._queue.put_nowait, (v, expired)) return True # continue listening - self._token = self._runner._dht.listen(key, _on_value) + self._token = self._runner._dht.listen(key, _on_value, filter, where) def __enter__(self) -> _Listener: return self @@ -67,10 +74,10 @@ class _Listener: self._queue.put_nowait(None) -class _GetStream: - """Async iterator around DHT get callback. +class _GetStream(Generic[_Result]): + """Async iterator around a DHT get or query callback. - Yields Value instances as they are found, then ends. + Yields results as they are found, then ends. Raises on failure. """ __slots__ = ("_runner", "_loop", "_queue", "_closed", "_error") @@ -81,14 +88,15 @@ class _GetStream: key: InfoHash, where: Optional[Where], filter: Optional[Callable[[Value], bool]], + query: Optional[Query] = None, ) -> None: self._runner = runner self._loop = runner._loop - self._queue: asyncio.Queue[Optional[Value]] = asyncio.Queue() + self._queue: asyncio.Queue[Optional[_Result]] = asyncio.Queue() self._closed = False self._error: Optional[BaseException] = None - def _on_value(v: Value) -> bool: + def _on_value(v: _Result) -> bool: # Called on DHT thread if self._closed: return False @@ -100,21 +108,24 @@ class _GetStream: if self._closed: return if not ok: - self._error = RuntimeError("DHT get failed") + self._error = RuntimeError("DHT query failed" if query is not None else "DHT get failed") self._loop.call_soon_threadsafe(self._queue.put_nowait, None) - self._runner._dht.get(key, _on_value, _on_done, filter, where) + if query is None: + self._runner._dht.get(key, _on_value, _on_done, filter, where) + else: + self._runner._dht.query(key, _on_value, _on_done, query) - def __enter__(self) -> _GetStream: + def __enter__(self) -> _GetStream[_Result]: return self def __exit__(self, exc_type, exc, tb) -> None: self.cancel() - def __aiter__(self) -> _GetStream: + def __aiter__(self) -> _GetStream[_Result]: return self - async def __anext__(self) -> Value: + async def __anext__(self) -> _Result: if self._closed: raise StopAsyncIteration item = await self._queue.get() @@ -137,7 +148,7 @@ class DhtRunner: """Async wrapper around opendht.DhtRunner. """ - __slots__ = ("_dht", "_loop", "_config", "_bootstrap") + __slots__ = ("_dht", "_loop", "_config", "_bootstrap", "_status_generation") def __init__( self, @@ -154,6 +165,7 @@ class DhtRunner: self._dht = core or _dht.DhtRunner() self._config = config self._bootstrap = list(bootstrap) if bootstrap is not None else None + self._status_generation = [0] def run( self, @@ -173,6 +185,29 @@ class DhtRunner: def bootstrap(self, host: str, port: Optional[str] = None) -> None: self._dht.bootstrap(host, port) + def setOnStatusChanged( + self, callback: Optional[Callable[[NodeStatus, NodeStatus], None]] = None, + ) -> None: + if callback is not None and not callable(callback): + raise TypeError("callback must be callable or None") + state = self._status_generation + state[0] += 1 + if callback is None: + self._dht.setOnStatusChanged(None) + return + generation = state[0] + loop = self._loop + + def _deliver(ipv4: NodeStatus, ipv6: NodeStatus) -> None: + if state[0] == generation: + callback(ipv4, ipv6) + + def _on_status(ipv4: NodeStatus, ipv6: NodeStatus) -> None: + if not loop.is_closed(): + loop.call_soon_threadsafe(_deliver, ipv4, ipv6) + + self._dht.setOnStatusChanged(_on_status) + async def ping(self, addr: SockAddr) -> bool: fut: asyncio.Future[bool] = self._loop.create_future() self._dht.ping(addr, lambda ok: self._loop.call_soon_threadsafe(_set_result_if_pending, fut, bool(ok))) @@ -184,9 +219,25 @@ class DhtRunner: *, where: Optional[Where] = None, filter: Optional[Callable[[Value], bool]] = None, - ): + ) -> _GetStream[Value]: # Return a streaming iterator of values; it ends on completion and raises on failure. - return _GetStream(self, key, where, filter) + return _GetStream[Value](self, key, where, filter) + + def query( + self, key: InfoHash, *, query: Optional[Query] = None, + ) -> _GetStream[dict[ValueField, Any]]: + if query is None: + query = Query("SELECT id,value_type,owner_pk,user_type") + return _GetStream[dict[ValueField, Any]](self, key, None, None, query) + + async def queryAll( + self, key: InfoHash, *, query: Optional[Query] = None, + ) -> List[dict[ValueField, Any]]: + rows: List[dict[ValueField, Any]] = [] + with self.query(key, query=query) as stream: + async for row in stream: + rows.append(row) + return rows async def getAll( self, @@ -255,8 +306,11 @@ class DhtRunner: self._dht.putEncrypted(key, val, to, _done, permanent) return await fut - def listen(self, key: InfoHash) -> _Listener: - return _Listener(self, key) + def listen( + self, key: InfoHash, *, where: Optional[Where] = None, + filter: Optional[Callable[[Value], bool]] = None, + ) -> _Listener: + return _Listener(self, key, where, filter) async def cancelListen(self, listener: _Listener) -> None: listener.cancel() diff --git a/python/opendht_cpp.pxd b/python/opendht_cpp.pxd index 4d530269..e6f03aef 100644 --- a/python/opendht_cpp.pxd +++ b/python/opendht_cpp.pxd @@ -155,6 +155,7 @@ cdef extern from "opendht/value.h" namespace "dht::Value": cdef extern from "opendht/value.h" namespace "dht::Value::Field": cdef Field None cdef Field Id + cdef Field ValueTypeField "dht::Value::Field::ValueType" cdef Field OwnerPk cdef Field SeqNum cdef Field UserType @@ -208,6 +209,14 @@ cdef extern from "opendht/value.h" namespace "dht": Where& userType(string user_type) string toString() const + cdef cppclass FieldValue: + uint64_t getInt() const + InfoHash getHash() const + Blob getBlob() const + + cdef cppclass FieldValueIndex: + map[Field, FieldValue] index + cdef extern from "opendht/node.h" namespace "dht": cdef cppclass Node: Node() except + @@ -216,6 +225,30 @@ cdef extern from "opendht/node.h" namespace "dht": bool isExpired() const cdef extern from "opendht/callbacks.h" namespace "dht": + cdef cppclass NodeStats: + unsigned good_nodes + unsigned dubious_nodes + unsigned cached_nodes + unsigned incoming_nodes + unsigned table_depth + unsigned searches + unsigned node_cache_size + unsigned getKnownNodes() const + unsigned long getNetworkSizeEstimation() const + + cdef cppclass NodeInfo: + InfoHash id + InfoHash node_id + NodeStats ipv4 + NodeStats ipv6 + size_t ongoing_ops + size_t storage_values + size_t storage_size + size_t local_storage_values + size_t local_storage_size + in_port_t bound4 + in_port_t bound6 + 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) @@ -254,17 +287,22 @@ cdef extern from "opendht/callbacks.h" namespace "dht": cdef extern from *: """ #include - #include + #include - static dht::ValueCallback opendht_bind_python_value_cb( - dht::ValueCallbackRaw callback, PyObject* data) + static std::shared_ptr opendht_python_owner(PyObject* data) { Py_INCREF(data); - auto owner = std::shared_ptr(data, [](PyObject* object) { + return std::shared_ptr(data, [](PyObject* object) { auto state = PyGILState_Ensure(); Py_DECREF(object); PyGILState_Release(state); }); + } + + static dht::ValueCallback opendht_bind_python_value_cb( + dht::ValueCallbackRaw callback, PyObject* data) + { + auto owner = opendht_python_owner(data); return [callback, owner = std::move(owner)]( const std::vector>& values, bool expired) { for (const auto& value : values) @@ -273,6 +311,48 @@ cdef extern from *: return true; }; } + + static dht::DhtRunner::Context opendht_python_context( + void (*callback)(int, int, void*), PyObject* data) + { + dht::DhtRunner::Context context; + auto owner = opendht_python_owner(data); + context.statusChangedCallback = [callback, owner = std::move(owner)]( + dht::NodeStatus ipv4, dht::NodeStatus ipv6) { + callback(static_cast(ipv4), static_cast(ipv6), owner.get()); + }; + return context; + } + + static void opendht_python_run(dht::DhtRunner& runner, in_port_t port, + dht::DhtRunner::Config& config, void (*callback)(int, int, void*), PyObject* data) + { + runner.run(port, config, opendht_python_context(callback, data)); + } + + static void opendht_python_run(dht::DhtRunner& runner, + const char* ipv4, const char* ipv6, const char* port, + dht::DhtRunner::Config& config, void (*callback)(int, int, void*), PyObject* data) + { + runner.run(ipv4, ipv6, port, config, opendht_python_context(callback, data)); + } + + static void opendht_python_query(dht::DhtRunner& runner, const dht::InfoHash& key, + bool (*callback)(std::shared_ptr, void*), + dht::DoneCallbackRaw done, PyObject* data, dht::Query query) + { + auto owner = opendht_python_owner(data); + runner.query(key, + [callback, owner](const std::vector>& rows) { + for (const auto& row : rows) + if (!callback(row, owner.get())) + return false; + return true; + }, + [done, owner](bool ok, const std::vector>& nodes) { + done(ok, const_cast>*>(&nodes), owner.get()); + }, std::move(query)); + } """ ValueCallback bindPythonValueCb "opendht_bind_python_value_cb"(ValueCallbackRaw callback, object data) except + @@ -298,9 +378,11 @@ cdef extern from "opendht/dhtrunner.h" namespace "dht": SockAddr bind4 SockAddr bind6 - InfoHash getId() const - shared_ptr[PublicKey] getPublicKey() const - InfoHash getNodeId() const + InfoHash getId() except + nogil + shared_ptr[PublicKey] getPublicKey() except + nogil + InfoHash getNodeId() except + nogil + NodeInfo getNodeInfo() except + nogil + vector[SockAddr] getPublicAddress(sa_family_t af) except + nogil void bootstrap(const_char*, const_char*) void bootstrap(const SockAddr&, DoneCallbackSimple done_cb) void run(in_port_t, Config config) except + @@ -309,22 +391,29 @@ cdef extern from "opendht/dhtrunner.h" namespace "dht": void shutdown(ShutdownCallback) void registerType(ValueType& value); bool isRunning() - SockAddr getBound(sa_family_t af) const - string getStorageLog() const - string getRoutingTablesLog(sa_family_t af) const - string getSearchesLog(sa_family_t af) const + SockAddr getBound(sa_family_t af) except + nogil + string getStorageLog() except + nogil + string getRoutingTablesLog(sa_family_t af) except + nogil + string getSearchesLog(sa_family_t af) except + nogil 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, time_point created, bool permanent) void putSigned(InfoHash key, shared_ptr[Value] val, DoneCallback done_cb, bool permanent) 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) except + + ListenToken listen(InfoHash key, ValueCallback get_cb, nullptr_t f, Where w) except + void cancelListen(InfoHash key, SharedListenToken token) except + - vector[unsigned] getNodeMessageStats(bool i) + vector[unsigned] getNodeMessageStats(bool i) except + nogil ctypedef DhtRunner.Config DhtRunnerConfig +cdef extern from *: + ctypedef void (*StatusCallbackRaw)(int ipv4, int ipv6, void* data) + ctypedef bool (*QueryCallbackRaw)(shared_ptr[FieldValueIndex] row, void* data) + void runPythonRunner "opendht_python_run"(DhtRunner& runner, in_port_t port, DhtRunnerConfig& config, StatusCallbackRaw callback, object data) except + + void runPythonRunnerAddress "opendht_python_run"(DhtRunner& runner, const_char* ipv4, const_char* ipv6, const_char* port, DhtRunnerConfig& config, StatusCallbackRaw callback, object data) except + + void queryPython "opendht_python_query"(DhtRunner& runner, InfoHash key, QueryCallbackRaw callback, DoneCallbackRaw done, object data, Query query) except + + cdef extern from "opendht/log.h" namespace "dht::log": void enableLogging(DhtRunner& dht) void disableLogging(DhtRunner& dht)