From 3a3c421ef9a3093439d9a0a7bccbae67e485cdd7 Mon Sep 17 00:00:00 2001 From: Adrien Beraud Date: Mon, 26 Oct 2026 16:54:05 -0400 Subject: [PATCH] nodejs: add an interactive dhtnode example --- .github/workflows/nodejs-package.yml | 1 + nodejs/.gitignore | 3 +- nodejs/README.md | 43 ++++++- nodejs/addon.cpp | 8 +- nodejs/examples/dhtnode.ts | 172 +++++++++++++++++++++++++++ nodejs/index.js | 116 +++++++++--------- nodejs/package.json | 74 ++++++------ nodejs/test/basic.test.js | 6 +- 8 files changed, 323 insertions(+), 100 deletions(-) create mode 100644 nodejs/examples/dhtnode.ts diff --git a/.github/workflows/nodejs-package.yml b/.github/workflows/nodejs-package.yml index 44a06b2c..d763434e 100644 --- a/.github/workflows/nodejs-package.yml +++ b/.github/workflows/nodejs-package.yml @@ -142,6 +142,7 @@ jobs: - name: Pack all platforms working-directory: nodejs run: | + npm ci --ignore-scripts mkdir -p dist npm pack --pack-destination dist - uses: actions/upload-artifact@v7 diff --git a/nodejs/.gitignore b/nodejs/.gitignore index 327c9736..09a5bf66 100644 --- a/nodejs/.gitignore +++ b/nodejs/.gitignore @@ -1,4 +1,5 @@ node_modules/ build*/ prebuilds/ -dist/ \ No newline at end of file +dist/ +examples/dhtnode.js \ No newline at end of file diff --git a/nodejs/README.md b/nodejs/README.md index 3ed40201..d339f09b 100644 --- a/nodejs/README.md +++ b/nodejs/README.md @@ -31,6 +31,38 @@ Set `OPENDHT_NODE_BINDING` to an absolute addon path to select an explicit build Packaged binaries include OpenDHT and non-system native dependencies, so archive installation does not require a compiler or a system OpenDHT installation. +## Example Tool + +After building, start the TypeScript example from the repository root: + +```sh +npm --prefix nodejs run dhtnode -- --port 4222 --bootstrap bootstrap.jami.net +``` + +The source is [examples/dhtnode.ts](examples/dhtnode.ts). `npm run build` compiles +it to JavaScript; run `npm run build:examples` after editing just the example. +Bootstrap is optional; +`--bootstrap-port` selects its UDP port. Use `--port 0` for a free local port. + +```text +listen greeting +put greeting hello world +get greeting +sign greeting signed hello +cancel greeting +quit +``` + +Keys are text labels, hashed automatically; values are UTF-8 text. The tool +generates a fresh EC identity on each run and prints IPv4/IPv6 connectivity +changes. Operations run concurrently, leaving the prompt responsive; dependent +commands should be entered after their preceding operation completes. EOF waits +for pending operations, then cancels listeners and closes the runner. Ctrl-C, +SIGTERM, and `quit` close immediately, rejecting pending operations without +printing cancellation errors. `--help` lists all options. Archives include both +the TypeScript source and compiled `examples/dhtnode.js`, runnable without +TypeScript or development dependencies, including from inside `node_modules`. + ## Usage ```js @@ -67,6 +99,11 @@ main().catch(console.error); selects a free port. The defaults bind both IP families on all interfaces. `persistPath` is optional; use an empty string to disable persistence. - `getBoundPort(4 | 6)` defaults to IPv4; `getNodeId()` returns the node's hex ID. +- `getStatus()` returns the latest JS-dispatched `{ipv4, ipv6}` connectivity + snapshot. Each family is `'disconnected'`, `'connecting'`, or `'connected'`. + `runner.on('statusChanged', status => {})` receives changes on the JS thread. + New and closing/closed runners report both families as disconnected. No status + events are delivered after close begins; close does not emit a final event. - `bootstrap(host, port)` queues bootstrap discovery. Returning does not mean the peer is connected; subsequent DHT operations report their own completion. - `put(key, Uint8Array, {id, type, userType, permanent})` copies the input and @@ -87,7 +124,7 @@ main().catch(console.error); Callbacks run only on their owning JS thread. There is no polling timer or worker thread waiting on each get/put. Explicitly close runners; GC cleanup and Node environment cleanup hooks are fallbacks, including forced worker termination. -Pending operations, active listeners, and registered error handlers keep their +Pending operations, active listeners, and registered error/status handlers keep their runner alive until completion, cancellation, handler removal, or explicit close. Graceful persistence cannot be guaranteed if the process is killed. @@ -163,7 +200,9 @@ Bundling replaces only the current platform's generated prebuild directory. install scripts disabled and no dev dependencies, forces prebuild-only loading, and runs the full runtime suite. Pass an archive file or a directory containing one archive to test it directly, for example `npm run test:package -- dist`. -Only JS, declarations, documentation, and prebuilds are included in the archive. +Only runtime JS, declarations, the example's TypeScript source, documentation, +and prebuilds are included in the archive. `prepack` compiles the example, so +archive creation needs the development dependencies; archive installation does not. The [Node.js package workflow](../.github/workflows/nodejs-package.yml) builds: diff --git a/nodejs/addon.cpp b/nodejs/addon.cpp index 6aec54ac..d1b5db63 100644 --- a/nodejs/addon.cpp +++ b/nodejs/addon.cpp @@ -303,8 +303,12 @@ 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)))); + 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; } diff --git a/nodejs/examples/dhtnode.ts b/nodejs/examples/dhtnode.ts new file mode 100644 index 00000000..2973f3d5 --- /dev/null +++ b/nodejs/examples/dhtnode.ts @@ -0,0 +1,172 @@ +#!/usr/bin/env node +'use strict'; + +import { clearLine, createInterface, cursorTo, type Interface } from 'node:readline'; +import { parseArgs } from 'node:util'; +import type { ConnectivityStatus, Listener, Value } from '../index.js'; + +const commands = 'get KEY | put KEY TEXT | sign KEY TEXT | listen KEY | cancel KEY | help | quit'; + +function portNumber(value: string, name: string, minimum = 0): number { + const port = Number(value); + if (!Number.isInteger(port) || port < minimum || port > 65535) { + throw new Error(`${name} must be an integer between ${minimum} and 65535`); + } + return port; +} + +function errorMessage(error: unknown): string { + return error instanceof Error ? error.message : String(error); +} + +function report(error: unknown): void { + console.error(errorMessage(error)); + process.exitCode = 1; +} + +async function main(): Promise { + const { values } = parseArgs({ + options: { + help: { type: 'boolean', short: 'h' }, + port: { type: 'string', short: 'p', default: '4222' }, + ipv4: { type: 'string', default: '0.0.0.0' }, + ipv6: { type: 'string', default: '::' }, + bootstrap: { type: 'string', short: 'b' }, + 'bootstrap-port': { type: 'string', default: '4222' }, + }, + }); + if (values.help) { + console.log(`Usage: npm run dhtnode -- [options] + + -p, --port PORT Local UDP port (default: 4222; 0 selects a free port) + -b, --bootstrap HOST Optional bootstrap hostname or IP address + --bootstrap-port PORT Bootstrap UDP port (default: 4222) + --ipv4 ADDRESS IPv4 bind address (default: 0.0.0.0) + --ipv6 ADDRESS IPv6 bind address (default: ::) + -h, --help Show help + +Commands: ${commands} +Keys are text labels hashed with SHA-1. Values are UTF-8 text.`); + return; + } + const port = portNumber(values.port, 'port'); + const bootstrapPort = portNumber(values['bootstrap-port'], 'bootstrap-port', 1); + const { DhtRunner, Identity, hash }: typeof import('../index.js') = require('..'); + const identity = await Identity.generate('dhtnode.ts'); + const runner = new DhtRunner(); + const listeners = new Map(); + const pending = new Set>(); + let lines: Interface | undefined; + let closing: Promise | undefined; + + function close(): Promise { + if (!closing) { + for (const listener of listeners.values()) listener.cancel(); + listeners.clear(); + lines?.close(); + closing = runner.close(); + } + return closing; + } + + function print(message: string): void { + if (process.stdout.isTTY && lines) { + clearLine(process.stdout, 0); + cursorTo(process.stdout, 0); + } + console.log(message); + if (!closing && process.stdin.isTTY) lines?.prompt(true); + } + + const stop = (): void => { close().catch(report); }; + const printValue = (label: string, value: Value): void => { + print(`${label} ${value.id}: ${JSON.stringify(value.data.toString('utf8'))}${value.isSigned ? ' [signed]' : ''}`); + }; + const printStatus = (status: ConnectivityStatus): void => { + print(`Connectivity: IPv4 ${status.ipv4}, IPv6 ${status.ipv6}`); + }; + + async function execute(line: string): Promise { + const [, command, key, data] = line.trim().match(/^(\S+)(?:\s+(\S+))?(?:\s+(.+))?$/) || []; + if (!command) return; + if (command === 'help') { + print(commands); + return; + } + if (!key) throw new Error('Expected a key. Type help for commands.'); + const location = hash(key); + switch (command) { + case 'get': { + const found = await runner.get(location); + if (closing) return; + for (const value of found) printValue(key, value); + if (!found.length) print(`No values for ${key}`); + break; + } + case 'put': + case 'sign': { + if (data === undefined) throw new Error(`Usage: ${command} KEY TEXT`); + const payload = Buffer.from(data, 'utf8'); + const ok = command === 'sign' ? await runner.putSigned(location, payload) + : await runner.put(location, payload); + if (closing) return; + if (!ok) throw new Error(`Put failed for ${key}`); + print(`Stored ${key}`); + break; + } + case 'listen': + if (!listeners.has(key)) { + listeners.set(key, runner.listen(location, (value, expired) => { + printValue(`${expired ? 'Expired' : 'Value'} ${key}`, value); + })); + } + print(`Listening ${key}`); + break; + case 'cancel': + listeners.get(key)?.cancel(); + listeners.delete(key); + print(`Cancelled ${key}`); + break; + default: + throw new Error('Unknown command. Type help for commands.'); + } + } + + runner.on('error', error => { report(error); stop(); }); + runner.on('statusChanged', printStatus); + process.on('SIGINT', stop); + process.on('SIGTERM', stop); + try { + runner.run({ port, ipv4: values.ipv4, ipv6: values.ipv6, identity }); + if (values.bootstrap) runner.bootstrap(values.bootstrap, bootstrapPort); + lines = createInterface({ input: process.stdin, output: process.stdout }); + lines.on('SIGINT', stop); + lines.setPrompt('dht> '); + print(`Node ${runner.getNodeId()} listening on UDP ${runner.getBoundPort()}`); + printStatus(runner.getStatus()); + if (process.stdin.isTTY) print(commands); + for await (const line of lines) { + if (closing) break; + if (/^(quit|exit)(?:\s|$)/.test(line.trim())) { + stop(); + break; + } + const operation = execute(line).catch(error => { + if (!closing) console.error(errorMessage(error)); + }).finally(() => { pending.delete(operation); }); + pending.add(operation); + if (process.stdin.isTTY) lines.prompt(); + } + if (!closing) await Promise.all(pending); + } finally { + try { + await close(); + await Promise.all(pending); + } finally { + process.removeListener('SIGINT', stop); + process.removeListener('SIGTERM', stop); + } + } +} + +main().catch(report); \ No newline at end of file diff --git a/nodejs/index.js b/nodejs/index.js index bb93b805..7a98978a 100644 --- a/nodejs/index.js +++ b/nodejs/index.js @@ -1,8 +1,8 @@ 'use strict'; const native = process.env.OPENDHT_NODE_BINDING - ? require(process.env.OPENDHT_NODE_BINDING) - : require('node-gyp-build')(__dirname); + ? require(process.env.OPENDHT_NODE_BINDING) + : require('node-gyp-build')(__dirname); const { EventEmitter } = require('node:events'); const { PublicKey, PrivateKey, Certificate, Identity, handle } = require('./crypto')(native); @@ -11,74 +11,74 @@ function eventReporter(owner, event) { } class DhtRunner extends EventEmitter { - #native; - #closing; - #listeners = 0; - #pending = 0; + #native; + #closing; + #listeners = 0; + #pending = 0; - constructor() { - super(); + constructor() { + super(); const owner = new WeakRef(this); this.#native = new native.NativeRunner(eventReporter(owner, 'error'), eventReporter(owner, 'statusChanged')); - } + } - #updateOwner() { + #updateOwner() { this.#native.setOwner(this.#listeners || this.#pending || this.listenerCount('error') || this.listenerCount('statusChanged') ? this : null); - } + } - #track(operation) { - this.#pending++; - this.#updateOwner(); - return operation.finally(() => { - this.#pending--; - this.#updateOwner(); - }); - } + #track(operation) { + this.#pending++; + this.#updateOwner(); + return operation.finally(() => { + this.#pending--; + this.#updateOwner(); + }); + } - on(event, listener) { super.on(event, listener); this.#updateOwner(); return this; } - addListener(event, listener) { return this.on(event, listener); } - prependListener(event, listener) { super.prependListener(event, listener); this.#updateOwner(); return this; } - removeListener(event, listener) { super.removeListener(event, listener); this.#updateOwner(); return this; } - off(event, listener) { return this.removeListener(event, listener); } - removeAllListeners(...args) { super.removeAllListeners(...args); this.#updateOwner(); return this; } + on(event, listener) { super.on(event, listener); this.#updateOwner(); return this; } + addListener(event, listener) { return this.on(event, listener); } + prependListener(event, listener) { super.prependListener(event, listener); this.#updateOwner(); return this; } + removeListener(event, listener) { super.removeListener(event, listener); this.#updateOwner(); return this; } + off(event, listener) { return this.removeListener(event, listener); } + removeAllListeners(...args) { super.removeAllListeners(...args); this.#updateOwner(); return this; } - run(options) { - if (options?.identity !== undefined) options = { ...options, identity: handle(options.identity, Identity) }; - this.#native.run(options); - } - isRunning() { return this.#native.isRunning(); } - getBoundPort(family = 4) { return this.#native.getBoundPort(family); } - getNodeId() { return this.#native.getNodeId(); } + run(options) { + if (options?.identity !== undefined) options = { ...options, identity: handle(options.identity, Identity) }; + this.#native.run(options); + } + 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)); } - putSigned(key, data, options) { return this.#track(this.#native.putSigned(key, data, options)); } - putEncrypted(key, data, recipient, options) { - return this.#track(this.#native.putEncrypted(key, data, options, handle(recipient, PublicKey))); - } + 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)); } + putSigned(key, data, options) { return this.#track(this.#native.putSigned(key, data, options)); } + putEncrypted(key, data, recipient, options) { + return this.#track(this.#native.putEncrypted(key, data, options, handle(recipient, PublicKey))); + } - listen(key, callback) { - const token = this.#native.listen(key, callback); - this.#listeners++; - this.#updateOwner(); - let cancelled = false; - return { - cancel: () => { - if (cancelled) return; - this.#native.cancelListen(token); - cancelled = true; - this.#listeners--; - this.#updateOwner(); - }, - }; - } + listen(key, callback) { + const token = this.#native.listen(key, callback); + this.#listeners++; + this.#updateOwner(); + let cancelled = false; + return { + cancel: () => { + if (cancelled) return; + this.#native.cancelListen(token); + cancelled = true; + this.#listeners--; + this.#updateOwner(); + }, + }; + } - close() { return this.#closing ??= this.#native.close(); } - ref() { this.#native.ref(); return this; } - unref() { this.#native.unref(); return this; } - [Symbol.asyncDispose]() { return this.close(); } + close() { return this.#closing ??= this.#native.close(); } + ref() { this.#native.ref(); return this; } + unref() { this.#native.unref(); return this; } + [Symbol.asyncDispose]() { return this.close(); } } module.exports = { DhtRunner, PublicKey, PrivateKey, Certificate, Identity, hash: native.hash, version: native.version }; \ No newline at end of file diff --git a/nodejs/package.json b/nodejs/package.json index 617220da..37dda39b 100644 --- a/nodejs/package.json +++ b/nodejs/package.json @@ -1,36 +1,42 @@ { - "name": "@opendht/node", - "version": "0.1.0", - "private": true, - "description": "Node-API bindings for OpenDHT", - "license": "MIT", - "main": "index.js", - "types": "index.d.ts", - "files": [ - "index.js", - "index.d.ts", - "crypto.js", - "README.md", - "prebuilds/" - ], - "engines": { - "node": ">=22.9.0" - }, - "scripts": { - "build": "cmake-js compile", - "bundle": "node scripts/bundle.js", - "package": "npm run bundle && npm pack --pack-destination dist", - "test:package": "node scripts/test-package.js", - "test": "node --test test/*.test.js", - "typecheck": "tsc --noEmit --strict --target ES2022 --module Node16 --moduleResolution Node16 test/types.ts" - }, - "dependencies": { - "node-gyp-build": "^4.8.4" - }, - "devDependencies": { - "@types/node": "^22.0.0", - "cmake-js": "^8.0.0", - "node-addon-api": "^8.5.0", - "typescript": "^5.9.3" - } + "name": "@opendht/node", + "version": "0.1.0", + "private": true, + "description": "Node-API bindings for OpenDHT", + "license": "MIT", + "main": "index.js", + "types": "index.d.ts", + "files": [ + "index.js", + "index.d.ts", + "crypto.js", + "README.md", + "examples/dhtnode.js", + "examples/dhtnode.ts", + "prebuilds/" + ], + "engines": { + "node": ">=22.9.0" + }, + "scripts": { + "build": "cmake-js compile", + "postbuild": "npm run build:examples", + "build:examples": "tsc --strict --target ES2022 --module Node16 --moduleResolution Node16 --rootDir examples --outDir examples examples/dhtnode.ts", + "dhtnode": "node examples/dhtnode.js", + "prepack": "npm run build:examples", + "bundle": "node scripts/bundle.js", + "package": "npm run bundle && npm pack --pack-destination dist", + "test:package": "node scripts/test-package.js", + "test": "node --test test/*.test.js", + "typecheck": "tsc --noEmit --strict --target ES2022 --module Node16 --moduleResolution Node16 test/types.ts examples/dhtnode.ts" + }, + "dependencies": { + "node-gyp-build": "^4.8.4" + }, + "devDependencies": { + "@types/node": "^22.0.0", + "cmake-js": "^8.0.0", + "node-addon-api": "^8.5.0", + "typescript": "^5.9.3" + } } \ No newline at end of file diff --git a/nodejs/test/basic.test.js b/nodejs/test/basic.test.js index be339435..a7f0ead5 100644 --- a/nodejs/test/basic.test.js +++ b/nodejs/test/basic.test.js @@ -171,7 +171,7 @@ test('queue overflow reports an error and rejects close', { timeout: 15000 }, as const runner = startNode(); const key = hash('bounded listener queue'); const reported = once(runner, 'error'); - const subscription = runner.listen(key, () => {}); + const subscription = runner.listen(key, () => { }); const puts = []; try { for (let index = 0; index < 3000; index++) { @@ -180,7 +180,7 @@ test('queue overflow reports an error and rejects close', { timeout: 15000 }, as })); } const deadline = performance.now() + 5000; - while (runner.isRunning() && performance.now() < deadline) {} + while (runner.isRunning() && performance.now() < deadline) { } assert.equal(runner.isRunning(), false); const [error] = await reported; assert.equal(error.code, 'ERR_DHT_OVERFLOW'); @@ -188,7 +188,7 @@ test('queue overflow reports an error and rejects close', { timeout: 15000 }, as await assert.rejects(runner.close(), { code: 'ERR_DHT_OVERFLOW' }); } finally { subscription.cancel(); - await runner.close().catch(() => {}); + await runner.close().catch(() => { }); } });