python: expose value state and harden runner operations

This commit is contained in:
Adrien Beraud committed 2026-10-26 01:10:34 -04:00
1 parent 9501465293
commit fb90211157
4 files changed
+68 -20

No files matched your search

+21 -3
View File
@@ -161,6 +161,18 @@ value.pushType = "audioCall"
Set metadata before publishing the value. Numeric values outside the native
integer range raise `OverflowError`.
`value.isSigned()` reports whether a public key and signature are present;
it does not verify the signature. `value.isEncrypted()` reports whether the
value currently contains ciphertext. Values delivered by `get()` and `listen()`
are transparently decrypted when possible, so decrypted values return `False`
from `isEncrypted()`, even if they were published using `putEncrypted()`.
An uninitialized `PublicKey()` cannot be used as an encryption recipient and
raises `ValueError`.
Register custom `ValueType` instances after `run()` and before `join()`.
`registerType()` raises `RuntimeError` when the runner is stopped, `TypeError`
for `None`, and `ValueError` for an uninitialized type.
## Observability
`getNodeInfo()` returns a read-only `NodeInfo` snapshot with node identifiers,
@@ -184,9 +196,15 @@ 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.
thread while native locks are held: do not call snapshot getters, type
registration, logging controls, `run()`, blocking operations, or `join()` from
them. The asyncio wrapper schedules status callbacks on its event-loop thread,
where snapshot getters are safe to use.
Call `node.connectivityChanged()` after a network-interface or IP-address
change. It queues native connectivity recovery, including recontacting peers
and renewing subscriptions. It is also available on the asyncio runner as a
synchronous, nonblocking method.
## Queries And Filtered Listeners
+32 -10
View File
@@ -441,6 +441,10 @@ cdef class Value(object):
self._value.reset(new cpp.Value(id, val, len(val)))
def __str__(self):
return self._value.get().toString().decode()
def isSigned(self) -> bool:
return self._value.get().isSigned()
def isEncrypted(self) -> bool:
return self._value.get().isEncrypted()
@property
def owner(self) -> PublicKey:
h = PublicKey()
@@ -963,8 +967,13 @@ cdef class DhtRunner(_WithID):
host_bytes = host.encode()
port_bytes = port.encode() if port else b'4222'
self.thisptr.get().bootstrap(<cpp.const_char*>host_bytes, <cpp.const_char*>port_bytes)
def connectivityChanged(self):
with nogil:
self.thisptr.get().connectivityChanged()
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()
cdef cpp.string bind4, bind6, service
cdef object status_ref = _weakref.ref(self._status_callback)
if id is not None:
run_config.dht_config.id = id._id
if is_bootstrap:
@@ -972,12 +981,15 @@ cdef class DhtRunner(_WithID):
if ipv4 or ipv6:
bind4 = ipv4.encode() if ipv4 else b''
bind6 = ipv6.encode() if ipv6 else b''
cpp.runPythonRunnerAddress(deref(self.thisptr.get()), bind4, bind6,
str(port).encode(), run_config,
status_callback, _weakref.ref(self._status_callback))
service = str(port).encode()
with nogil:
cpp.runPythonRunnerAddress(deref(self.thisptr.get()), bind4.c_str(),
bind6.c_str(), service.c_str(), run_config,
status_callback, status_ref)
else:
cpp.runPythonRunner(deref(self.thisptr.get()), port, run_config,
status_callback, _weakref.ref(self._status_callback))
with nogil:
cpp.runPythonRunner(deref(self.thisptr.get()), port, run_config,
status_callback, status_ref)
def join(self):
with nogil:
self.thisptr.get().join()
@@ -986,14 +998,22 @@ cdef class DhtRunner(_WithID):
cb_obj = {'shutdown':shutdown_cb}
ref.Py_INCREF(cb_obj)
self.thisptr.get().shutdown(cpp.bindShutdownCb(shutdown_callback, <void*>cb_obj))
def registerType(self, ValueType type):
self.thisptr.get().registerType(deref(type._value.get()))
def registerType(self, ValueType type not None):
if not type._value:
raise ValueError("ValueType is not initialized")
cdef cpp.ValueType run_type = deref(type._value.get())
with nogil:
self.thisptr.get().registerType(run_type)
def enableLogging(self):
cpp.enableLogging(self.thisptr.get()[0])
with nogil:
cpp.enableLogging(self.thisptr.get()[0])
def disableLogging(self):
cpp.disableLogging(self.thisptr.get()[0])
with nogil:
cpp.disableLogging(self.thisptr.get()[0])
def enableFileLogging(self, str path):
cpp.enableFileLogging(self.thisptr.get()[0], path.encode())
cdef cpp.string log_path = path.encode()
with nogil:
cpp.enableFileLogging(self.thisptr.get()[0], log_path)
def isRunning(self):
return self.thisptr.get().isRunning()
def getBound(self, cpp.sa_family_t af = 0):
@@ -1135,6 +1155,8 @@ cdef class DhtRunner(_WithID):
def putEncrypted(self, InfoHash key, Value val, to, done_cb=None, bool permanent=False):
if to is None:
raise ValueError("'to' must be provided")
if isinstance(to, PublicKey) and not (<PublicKey>to)._key:
raise ValueError("PublicKey is not initialized")
if done_cb:
cb_obj = {'done':done_cb}
ref.Py_INCREF(cb_obj)
+13 -7
View File
@@ -168,6 +168,8 @@ cdef extern from "opendht/value.h" namespace "dht":
Value(const uint16_t t, const uint8_t* dat_ptr, size_t dat_len) except +
string toString() const
size_t size() const
bool isSigned() const
bool isEncrypted() const
uint64_t id
shared_ptr[PublicKey] owner
InfoHash recipient
@@ -179,6 +181,7 @@ cdef extern from "opendht/value.h" namespace "dht":
string pushType
cdef cppclass ValueType:
ValueType() except +
ValueType(uint16_t id, string name, seconds expiration) except +
uint16_t id
string name
@@ -200,7 +203,7 @@ cdef extern from "opendht/value.h" namespace "dht":
cdef cppclass Where:
Where() except +
Where(const string& q_str)
Where(const string& q_str) except +
bool isSatisfiedBy(const Where& where) const
Where& id(uint64_t id)
Where& valueType(uint16_t type)
@@ -291,7 +294,9 @@ cdef extern from *:
static std::shared_ptr<PyObject> opendht_python_owner(PyObject* data)
{
auto state = PyGILState_Ensure();
Py_INCREF(data);
PyGILState_Release(state);
return std::shared_ptr<PyObject>(data, [](PyObject* object) {
auto state = PyGILState_Ensure();
Py_DECREF(object);
@@ -383,13 +388,14 @@ cdef extern from "opendht/dhtrunner.h" namespace "dht":
InfoHash getNodeId() except + nogil
NodeInfo getNodeInfo() except + nogil
vector[SockAddr] getPublicAddress(sa_family_t af) except + nogil
void connectivityChanged() except + nogil
void bootstrap(const_char*, const_char*)
void bootstrap(const SockAddr&, DoneCallbackSimple done_cb)
void run(in_port_t, Config config) except +
void run(const_char*, const_char*, const_char*, Config config) except +
void join() except + nogil
void shutdown(ShutdownCallback)
void registerType(ValueType& value);
void registerType(const ValueType& value) except + nogil
bool isRunning()
SockAddr getBound(sa_family_t af) except + nogil
string getStorageLog() except + nogil
@@ -410,14 +416,14 @@ 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 runPythonRunner "opendht_python_run"(DhtRunner& runner, in_port_t port, DhtRunnerConfig& config, StatusCallbackRaw callback, object data) except + nogil
void runPythonRunnerAddress "opendht_python_run"(DhtRunner& runner, const_char* ipv4, const_char* ipv6, const_char* port, DhtRunnerConfig& config, StatusCallbackRaw callback, object data) except + nogil
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)
void enableFileLogging(DhtRunner& dht, const string& path)
void enableLogging(DhtRunner& dht) except + nogil
void disableLogging(DhtRunner& dht) except + nogil
void enableFileLogging(DhtRunner& dht, const string& path) except + nogil
cdef extern from "opendht/indexation/pht.h" namespace "dht::indexation":
size_t PHT_MAX_NODE_ENTRY_COUNT "dht::indexation::Pht::MAX_NODE_ENTRY_COUNT"