From d716fae4bce2203e5b15bb91f42efabd2dacbcfe Mon Sep 17 00:00:00 2001 From: Adrien Beraud Date: Mon, 26 Oct 2026 20:51:01 -0400 Subject: [PATCH] nodejs: expose per-family connectivity status and events --- nodejs/addon.cpp | 39 +++++++++++++++++++++++++++++++++++++-- nodejs/index.d.ts | 15 ++++++++++++++- nodejs/index.js | 11 +++++++---- 3 files changed, 58 insertions(+), 7 deletions(-) diff --git a/nodejs/addon.cpp b/nodejs/addon.cpp index a52607dc..6aec54ac 100644 --- a/nodejs/addon.cpp +++ b/nodejs/addon.cpp @@ -111,11 +111,13 @@ struct ValueData struct Event { - enum class Kind { Value, Done, Overflow }; + enum class Kind { Value, Done, Overflow, Status }; Kind kind; uint64_t operation; bool flag; std::optional value; + dht::NodeStatus ipv4 = dht::NodeStatus::Disconnected; + dht::NodeStatus ipv6 = dht::NodeStatus::Disconnected; size_t size() const { return sizeof(Event) + (value ? value->size() : 0); } }; @@ -247,6 +249,13 @@ struct RunnerState { if (closing) return; + if (event.kind == Event::Kind::Status) { + status4 = event.ipv4; + status6 = event.ipv6; + if (!onStatus.IsEmpty()) + onStatus.Call({status()}); + return; + } if (event.kind == Event::Kind::Overflow) { overflowed = true; startClose(); @@ -291,6 +300,14 @@ struct RunnerState } } + Napi::Object status() const + { + auto result = Napi::Object::New(env); + result.Set("ipv4", Napi::String::New(env, std::string(dht::statusToStr(closing ? dht::NodeStatus::Disconnected : status4)))); + result.Set("ipv6", Napi::String::New(env, std::string(dht::statusToStr(closing ? dht::NodeStatus::Disconnected : status6)))); + return result; + } + static void dispatch(napi_env environment, napi_value, void* context, void* data) { auto state = static_cast(context); @@ -380,6 +397,7 @@ struct RunnerState state->closeDeferred.reset(); state->closePromise.Reset(); state->onError.Reset(); + state->onStatus.Reset(); state->owner.Reset(); napi_delete_async_work(environment, state->closeWork); napi_remove_async_cleanup_hook(state->cleanupHook); @@ -407,7 +425,10 @@ struct RunnerState std::optional closeDeferred; Napi::Reference closePromise; Napi::FunctionReference onError; + Napi::FunctionReference onStatus; Napi::ObjectReference owner; + dht::NodeStatus status4 = dht::NodeStatus::Disconnected; + dht::NodeStatus status6 = dht::NodeStatus::Disconnected; }; class Runner : public Napi::ObjectWrap @@ -419,6 +440,8 @@ public: { if (info.Length() && info[0].IsFunction()) state_->onError = Napi::Persistent(info[0].As()); + if (info.Length() > 1 && info[1].IsFunction()) + state_->onStatus = Napi::Persistent(info[1].As()); } ~Runner() override { state_->startClose(); } @@ -433,6 +456,7 @@ public: InstanceMethod("isRunning", &Runner::isRunning), InstanceMethod("getBoundPort", &Runner::getBoundPort), InstanceMethod("getNodeId", &Runner::getNodeId), + InstanceMethod("getStatus", &Runner::getStatus), InstanceMethod("bootstrap", &Runner::bootstrap), InstanceMethod("get", &Runner::get), InstanceMethod("put", &Runner::put), @@ -466,8 +490,14 @@ private: config.dht_config.node_config.persist_path = text(options.Get("persistPath"), "persistPath"); if (options.Has("identity")) config.dht_config.id = opendht_node::readIdentity(options.Get("identity")); + dht::DhtRunner::Context context; + const std::weak_ptr state = state_; + context.statusChangedCallback = [state](dht::NodeStatus ipv4, dht::NodeStatus ipv6) { + if (auto owner = state.lock()) + owner->send(std::make_unique(Event {Event::Kind::Status, 0, false, {}, ipv4, ipv6})); + }; const auto service = std::to_string(port); - state_->runner.run(ipv4.c_str(), ipv6.c_str(), service.c_str(), config); + state_->runner.run(ipv4.c_str(), ipv6.c_str(), service.c_str(), config, std::move(context)); state_->started = true; state_->hasIdentity = bool(config.dht_config.id.first); state_->dispatcher.Ref(info.Env()); @@ -500,6 +530,11 @@ private: return Napi::Boolean::New(info.Env(), !state_->closing && !state_->overflowed && state_->runner.isRunning()); } + Napi::Value getStatus(const Napi::CallbackInfo& info) + { + return guarded(info, [&]() -> Napi::Value { return state_->status(); }); + } + Napi::Value getBoundPort(const Napi::CallbackInfo& info) { return guarded(info, [&]() -> Napi::Value { diff --git a/nodejs/index.d.ts b/nodejs/index.d.ts index bd95b1ca..059512a4 100644 --- a/nodejs/index.d.ts +++ b/nodejs/index.d.ts @@ -73,11 +73,24 @@ export interface Listener { cancel(): void; } -export class DhtRunner extends EventEmitter implements AsyncDisposable { +export type NodeStatus = 'disconnected' | 'connecting' | 'connected'; + +export interface ConnectivityStatus { + ipv4: NodeStatus; + ipv6: NodeStatus; +} + +export interface RunnerEvents { + statusChanged: [status: ConnectivityStatus]; + error: [error: Error]; +} + +export class DhtRunner extends EventEmitter implements AsyncDisposable { run(options?: RunOptions): void; isRunning(): boolean; getBoundPort(family?: 4 | 6): number; getNodeId(): string; + getStatus(): ConnectivityStatus; bootstrap(host: string, port: number): void; get(key: string): Promise; put(key: string, data: Uint8Array, options?: PutOptions): Promise; diff --git a/nodejs/index.js b/nodejs/index.js index 211fcdae..bb93b805 100644 --- a/nodejs/index.js +++ b/nodejs/index.js @@ -6,8 +6,8 @@ const native = process.env.OPENDHT_NODE_BINDING const { EventEmitter } = require('node:events'); const { PublicKey, PrivateKey, Certificate, Identity, handle } = require('./crypto')(native); -function errorReporter(owner) { - return error => { owner.deref()?.emit('error', error); }; +function eventReporter(owner, event) { + return value => { owner.deref()?.emit(event, value); }; } class DhtRunner extends EventEmitter { @@ -18,11 +18,13 @@ class DhtRunner extends EventEmitter { constructor() { super(); - this.#native = new native.NativeRunner(errorReporter(new WeakRef(this))); + const owner = new WeakRef(this); + this.#native = new native.NativeRunner(eventReporter(owner, 'error'), eventReporter(owner, 'statusChanged')); } #updateOwner() { - this.#native.setOwner(this.#listeners || this.#pending || this.listenerCount('error') ? this : null); + this.#native.setOwner(this.#listeners || this.#pending || this.listenerCount('error') + || this.listenerCount('statusChanged') ? this : null); } #track(operation) { @@ -48,6 +50,7 @@ class DhtRunner extends EventEmitter { isRunning() { return this.#native.isRunning(); } getBoundPort(family = 4) { return this.#native.getBoundPort(family); } getNodeId() { return this.#native.getNodeId(); } + getStatus() { return this.#native.getStatus(); } bootstrap(host, port) { this.#native.bootstrap(host, port); } get(key) { return this.#track(this.#native.get(key)); } put(key, data, options) { return this.#track(this.#native.put(key, data, options)); }