Compare commits

...
31 Commits
Author SHA1 Message Date
Adrien BeraudandCopilot dff27203a4 build: prepare 4.5.0 release candidate 1
Co-authored-by: Copilot <223556219+Copilot@users.nply.github.com>
2026-11-03 22:48:09 -05:00
Adrien BeraudandCopilot 506d9f9cbb ci: install MessagePack C++ headers on Ubuntu
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
2026-11-03 22:44:50 -05:00
dependabot[bot] 349f4e5fc6 build(deps): bump pypa/cibuildwheel from 4.3.4 to 4.3.5
Bumps [pypa/cibuildwheel](https://github.com/pypa/cibuildwheel) from 4.3.4 to 4.3.5.
- [Release notes](https://github.com/pypa/cibuildwheel/releases)
- [Changelog](https://github.com/pypa/cibuildwheel/blob/main/docs/changelog.md)
- [Commits](https://github.com/pypa/cibuildwheel/compare/v4.3.4...v4.3.5)

---
updated-dependencies:
- dependency-name: pypa/cibuildwheel
  dependency-version: 4.3.5
  dependency-type: direct:production
  update-type: version-update:semver-patch
...

Signed-off-by: dependabot[bot] <support@github.com>
2026-11-03 22:41:04 -05:00
dependabot[bot] 11d69d02e8 build(deps): bump actions/setup-node from 6 to 7
Bumps [actions/setup-node](https://github.com/actions/setup-node) from 6 to 7.
- [Release notes](https://github.com/actions/setup-node/releases)
- [Commits](https://github.com/actions/setup-node/compare/v6...v7)

---
updated-dependencies:
- dependency-name: actions/setup-node
  dependency-version: '7'
  dependency-type: direct:production
  update-type: version-update:semver-major
...

Signed-off-by: dependabot[bot] <support@github.com>
2026-11-03 22:40:02 -05:00
Adrien Beraud 3c87991e92 docs: make Rust README precise and factual 2026-10-29 22:46:57 -04:00
Adrien Beraud 52287bf608 ci: use npm environment for publishing 2026-10-29 22:27:34 -04:00
Adrien Beraud 061ce2deb4 docker: build with RelWithDebInfo 2026-10-29 22:27:04 -04:00
Adrien Beraud 64b536fb53 ci: publish Python wheels from the tag release workflow 2026-10-29 21:48:33 -04:00
Adrien Beraud 67c76f17de README: add Node.js bindings 2026-10-29 21:20:25 -04:00
Adrien Beraud 25fde756e8 docs: add linked package and Docker badges 2026-10-29 21:12:35 -04:00
Adrien Beraud f6b3eda280 rust: rewrite README 2026-10-29 20:55:07 -04:00
Adrien Beraud 9aee407a42 rust: avoid dual-stack conflict in startup error test 2026-10-29 10:35:51 -04:00
Adrien Beraud 722ef44c7d rust: move owned Value directly through callbacks and streams 2026-10-28 23:51:46 -04:00
Adrien Beraud 6ed2720bd7 rust: test single-copy async value ownership 2026-10-28 23:51:46 -04:00
Adrien Beraud 9005f73141 rust: add optional Serde typed operations 2026-10-28 23:51:46 -04:00
Adrien Beraud 9785c0a127 rust: add typed operation regression tests 2026-10-28 23:51:41 -04:00
Adrien Beraud 5662cc36d1 rust: publish crate on release with trusted publishing 2026-10-28 22:04:57 -04:00
Adrien Beraud e116cfe548 rust: add optional runtime-independent async API 2026-10-28 21:11:43 -04:00
Adrien Beraud 1c5e8378b8 rust: add async adapter regression tests 2026-10-28 21:11:42 -04:00
Adrien Beraud c630e80345 ci: use trusted publishing for Python wheels 2026-10-29 00:26:07 +00:00
cyrilleandCopilot 17f83edce4 proxy: add Huawei (HMS) push type
Route push notifications to Huawei devices, which have no Google Play
Services and therefore cannot receive FCM. gorush, already used as the
push relay, supports HMS natively as platform 3.

HMS takes its payload as a serialized JSON string in "huawei_data" and
carries its own ttl/priority fields, so it needs its own branch rather
than the generic "data" object used for FCM and APNs.

The enum value is appended last: MSGPACK_ADD_ENUM serializes it as an
integer, so reordering would break compatibility with deployed proxies.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
2026-10-29 00:02:02 +00:00
Adrien Beraud afab251376 nodejs: isolate native Linux dependency symbols
Use RTLD_DEEPBIND when loading the Linux addon so its llhttp calls do not resolve to Node's incompatible bundled parser. The CI image uses llhttp 9.2.1 while Node 22 exports 9.4.3, corrupting the heap when the proxy client starts. Preserve default loading on other platforms.
2026-10-27 23:04:32 -04:00
Adrien Beraud 86d452fbb6 nodejs: bundle source-built llhttp license notices
The Noble dependency image builds llhttp from source, so dpkg-query cannot identify its license package. Include the pinned upstream MIT notice when bundling Linux llhttp libraries while retaining package-owner checks for other dependencies.
2026-10-27 23:04:32 -04:00
Adrien Beraud 08ff775232 ci: rely on npm trusted publishing 2026-10-27 22:48:58 -04:00
Adrien Beraud 738e0e22bf nodejs: add streaming iterators and startup proxy configuration 2026-10-27 22:46:34 -04:00
Adrien Beraud 8ec5b47c86 nodejs: test streaming iterators and proxy configuration 2026-10-27 22:09:21 -04:00
Adrien Beraud e1c31568f9 readme: cleanup 2026-10-27 22:01:05 -04:00
Adrien Beraud 45bad769b4 nodejs: fix npm archive paths and validate release versions 2026-10-27 21:22:15 -04:00
Adrien Beraud 09780fd923 nodejs: publish verified npm archives from tag releases 2026-10-27 20:36:12 -04:00
Adrien Beraud 747e70d1ae nodejs: simplify design 2026-10-27 20:12:17 -04:00
Adrien Beraud 05abfa89bf nodejs: initialize GnuTLS before using the addon 2026-10-27 11:33:17 -04:00
40 changed files with 2452 additions and 456 deletions

No files matched your search

+15 -3
View File
@@ -4,9 +4,9 @@ env:
LLHTTP_VERSION: "v9.2.1"
RESTINIO_VERSION: "0.7.10"
EXPECTED_LITE_URL: "https://raw.githubusercontent.com/martinmoene/expected-lite/master/include/nonstd/expected.hpp"
UBUNTU_FULL_DEPS: "libncurses5-dev libreadline-dev nettle-dev libasio-dev libgnutls28-dev libuv1-dev python3-pip python3-dev python3-setuptools libcppunit-dev libjsoncpp-dev libfmt-dev libmsgpack-dev libargon2-dev"
UBUNTU_MESON_DEPS: "ninja-build libncurses5-dev libreadline-dev nettle-dev libasio-dev libgnutls28-dev libuv1-dev python3-dev python3-setuptools python3-pip libcppunit-dev libjsoncpp-dev libfmt-dev libmsgpack-dev libargon2-dev"
UBUNTU_MINIMAL_DEPS: "libncurses5-dev libreadline-dev nettle-dev libfmt-dev libgnutls28-dev libcppunit-dev libmsgpack-dev libargon2-dev"
UBUNTU_FULL_DEPS: "libncurses5-dev libreadline-dev nettle-dev libasio-dev libgnutls28-dev libuv1-dev python3-pip python3-dev python3-setuptools libcppunit-dev libjsoncpp-dev libfmt-dev libmsgpack-cxx-dev libargon2-dev"
UBUNTU_MESON_DEPS: "ninja-build libncurses5-dev libreadline-dev nettle-dev libasio-dev libgnutls28-dev libuv1-dev python3-dev python3-setuptools python3-pip libcppunit-dev libjsoncpp-dev libfmt-dev libmsgpack-cxx-dev libargon2-dev"
UBUNTU_MINIMAL_DEPS: "libncurses5-dev libreadline-dev nettle-dev libfmt-dev libgnutls28-dev libcppunit-dev libmsgpack-cxx-dev libargon2-dev"
DEBIAN13_DEPS: "git libtool pkg-config wget cmake libncurses5-dev libreadline-dev nettle-dev libasio-dev libgnutls28-dev libuv1-dev python3-pip python3-dev python3-setuptools libcppunit-dev libjsoncpp-dev libfmt-dev libmsgpack-cxx-dev libargon2-dev"
on: [push, pull_request]
@@ -33,6 +33,18 @@ jobs:
- name: Build with Ninja
run: cd build && ninja
- uses: dtolnay/rust-toolchain@stable
- name: Test Rust bindings
env:
LIBRARY_PATH: ${{ github.workspace }}/build
LD_LIBRARY_PATH: ${{ github.workspace }}/build
RUSTFLAGS: -L native=${{ github.workspace }}/build
run: |
cargo test --manifest-path rust/Cargo.toml --all-targets
cargo test --manifest-path rust/Cargo.toml --all-features --all-targets
cargo test --manifest-path rust/Cargo.toml --all-features --doc
build-ubuntu-minimal:
name: Ubuntu/GCC minimal build
runs-on: ubuntu-latest
+1 -1
View File
@@ -12,7 +12,7 @@ jobs:
sudo apt update && \
sudo apt install libncurses5-dev libreadline-dev nettle-dev \
libgnutls28-dev libuv1-dev python3-dev python3-setuptools libcppunit-dev libjsoncpp-dev \
libfmt-dev libmsgpack-dev libargon2-dev libasio-dev \
libfmt-dev libmsgpack-cxx-dev libargon2-dev libasio-dev \
llvm llvm-dev clang clang-tools && \
sudo apt remove gcc g++
+33 -26
View File
@@ -2,6 +2,7 @@ name: Node.js Package
on:
push:
branches: ['**']
paths:
- 'nodejs/**'
- '.github/workflows/nodejs-package.yml'
@@ -23,8 +24,7 @@ on:
- 'vcpkg*.json'
- 'ports/**'
- 'LICENSE'
release:
types: [published]
workflow_call:
workflow_dispatch:
permissions:
@@ -64,14 +64,26 @@ jobs:
VCPKG_BINARY_SOURCES: clear;files,${{ github.workspace }}/vcpkg-binary-cache,readwrite
steps:
- uses: actions/checkout@v7
- uses: actions/setup-node@v6
- uses: actions/setup-node@v7
with:
node-version: '22'
cache: npm
cache-dependency-path: nodejs/package-lock.json
- name: Install macOS dependencies
if: runner.os == 'macOS'
run: brew install cmake ninja pkg-config dylibbundler gnutls nettle argon2 msgpack-cxx fmt jsoncpp simdutf
run: brew install cmake ninja pkg-config dylibbundler gnutls nettle argon2 msgpack-cxx fmt jsoncpp simdutf asio llhttp openssl
- name: Install restinio
if: runner.os == 'macOS'
env:
RESTINIO_VERSION: '0.7.10'
run: |
include="$RUNNER_TEMP/opendht-proxy/include"
mkdir -p "$include/nonstd"
curl -fsSL --retry 3 https://raw.githubusercontent.com/martinmoene/expected-lite/v0.10.0/include/nonstd/expected.hpp -o "$include/nonstd/expected.hpp"
curl -fsSL --retry 3 "https://github.com/Stiffstream/restinio/releases/download/v${RESTINIO_VERSION}/restinio-${RESTINIO_VERSION}.tar.bz2" -o "$RUNNER_TEMP/restinio.tar.bz2"
tar -xjf "$RUNNER_TEMP/restinio.tar.bz2" -C "$RUNNER_TEMP"
cp -R "$RUNNER_TEMP/restinio-${RESTINIO_VERSION}/dev/restinio" "$include/"
echo "CMAKE_PREFIX_PATH=$RUNNER_TEMP/opendht-proxy" >> "$GITHUB_ENV"
- name: Set up MSVC
if: runner.os == 'Windows'
uses: ilammy/msvc-dev-cmd@v1
@@ -98,6 +110,8 @@ jobs:
run: |
if [[ "$RUNNER_OS" == 'macOS' ]]; then
export MACOSX_DEPLOYMENT_TARGET=15.0
openssl_prefix=$(brew --prefix openssl)
export PKG_CONFIG_PATH="$openssl_prefix/lib/pkgconfig${PKG_CONFIG_PATH:+:$PKG_CONFIG_PATH}"
fi
npm run build -- --CDOPENDHT_DOWNLOAD_DEPS=OFF
- name: Build Windows addon
@@ -110,6 +124,7 @@ jobs:
"--CDVCPKG_INSTALLED_DIR=$env:VCPKG_INSTALLED_DIR" `
"--CDVCPKG_TARGET_TRIPLET=$env:VCPKG_DEFAULT_TRIPLET" `
--CDVCPKG_MANIFEST_NO_DEFAULT_FEATURES=ON `
--CDVCPKG_MANIFEST_FEATURES=proxy `
--CDCMAKE_MSVC_RUNTIME_LIBRARY=MultiThreadedDLL `
--CDOPENDHT_DOWNLOAD_DEPS=OFF
if ($LASTEXITCODE -ne 0) { exit $LASTEXITCODE }
@@ -136,7 +151,7 @@ jobs:
runs-on: ubuntu-26.04
steps:
- uses: actions/checkout@v7
- uses: actions/setup-node@v6
- uses: actions/setup-node@v7
with:
node-version: '22'
- uses: actions/download-artifact@v8
@@ -144,6 +159,17 @@ jobs:
pattern: opendht-node-prebuild-*
path: nodejs/prebuilds
merge-multiple: true
- name: Set release version
if: github.event_name == 'push' && github.ref_type == 'tag'
working-directory: nodejs
env:
RELEASE_TAG: ${{ github.ref_name }}
run: |
release_version=$(npm version "${RELEASE_TAG#v}" --no-git-tag-version --ignore-scripts --allow-same-version)
if [[ "$release_version" != "$RELEASE_TAG" ]]; then
printf '::error::Release tag %s is not an exact npm semantic version (normalized to %s).\n' "$RELEASE_TAG" "$release_version" >&2
exit 1
fi
- name: Pack all platforms
working-directory: nodejs
run: |
@@ -168,7 +194,7 @@ jobs:
node: ['22', '24', '26']
steps:
- uses: actions/checkout@v7
- uses: actions/setup-node@v6
- uses: actions/setup-node@v7
with:
node-version: ${{ matrix.node }}
- uses: actions/download-artifact@v8
@@ -177,23 +203,4 @@ jobs:
path: nodejs/dist
- name: Test installed archive without a native build toolchain
working-directory: nodejs
run: npm run test:package -- dist
release:
name: Attach npm archive to release
if: github.event_name == 'release'
needs: smoke
runs-on: ubuntu-26.04
permissions:
contents: write
steps:
- uses: actions/download-artifact@v8
with:
name: opendht-node-npm
path: dist
- name: Upload verified archive
env:
GH_TOKEN: ${{ github.token }}
GH_REPO: ${{ github.repository }}
RELEASE_TAG: ${{ github.event.release.tag_name }}
run: gh release upload "$RELEASE_TAG" dist/*.tgz --clobber
run: npm run test:package -- dist
+3 -19
View File
@@ -4,10 +4,10 @@ permissions:
contents: read
on:
release:
types: [published]
push:
branches: ['**']
pull_request:
workflow_call:
workflow_dispatch:
jobs:
@@ -23,7 +23,7 @@ jobs:
# Run locally with: pipx run cibuildwheel --platform linux python
- name: Build wheels
uses: pypa/cibuildwheel@v4.3.4
uses: pypa/cibuildwheel@v4.3.5
with:
package-dir: python
@@ -82,19 +82,3 @@ jobs:
with:
name: opendht-wheels-win_amd64
path: ./build/wheelhouse/*.whl
publish:
runs-on: ubuntu-latest
needs: [build_wheels, build_wheels_windows]
steps:
- uses: actions/download-artifact@v8
with:
pattern: opendht-wheels-*
path: dist
merge-multiple: true
- name: Publish package
if: github.event_name == 'push' && startsWith(github.ref, 'refs/tags')
uses: pypa/gh-action-pypi-publish@release/v1
with:
password: ${{ secrets.PYPI_API_TOKEN }}
+102 -1
View File
@@ -18,6 +18,19 @@ env:
IMAGE_NAME_ALPINE_DEPS: ${{ github.repository }}/opendht-deps-alpine
jobs:
nodejs-package:
name: Build and verify Node.js package
uses: ./.github/workflows/nodejs-package.yml
permissions:
contents: read
packages: read
python-wheels:
name: Build Python wheels
uses: ./.github/workflows/python-wheels.yml
permissions:
contents: read
build-and-push-deps-image:
name: Dependency Docker image
runs-on: ubuntu-latest
@@ -555,7 +568,7 @@ jobs:
create-github-release:
name: Create GitHub Release
runs-on: ubuntu-latest
needs: [build-windows-msi, build-windows-dhtnode]
needs: [build-windows-msi, build-windows-dhtnode, nodejs-package]
permissions:
contents: write
@@ -575,11 +588,99 @@ jobs:
name: opendht-windows-dhtnode
path: downloaded_artifacts/static_dhtnode
- name: Download npm archive
uses: actions/download-artifact@v8
with:
name: opendht-node-npm
path: downloaded_artifacts/nodejs
- name: Create Release
id: create_release
uses: softprops/action-gh-release@v3
with:
draft: true
prerelease: ${{ contains(github.ref_name, '-') }}
files: |
downloaded_artifacts/windows_msi/*.msi
downloaded_artifacts/static_dhtnode/**/*dhtnode.exe
downloaded_artifacts/nodejs/*.tgz
publish-npm:
name: Publish opendht to npm
needs: create-github-release
runs-on: ubuntu-26.04
environment:
name: npm
url: https://www.npmjs.com/package/opendht
permissions:
contents: read
id-token: write
steps:
- uses: actions/checkout@v7
- uses: actions/setup-node@v7
with:
node-version: '24'
registry-url: https://registry.npmjs.org/
package-manager-cache: false
- uses: actions/download-artifact@v8
with:
name: opendht-node-npm
path: nodejs/dist
- name: Publish verified archive
working-directory: nodejs
env:
RELEASE_TAG: ${{ github.ref_name }}
run: |
dist_tag=latest
if [[ "${RELEASE_TAG#v}" == *-* ]]; then
dist_tag=next
fi
npm publish ./dist/*.tgz --access public --provenance --tag "$dist_tag"
publish-pypi:
name: Publish opendht to PyPI
needs: [create-github-release, python-wheels]
runs-on: ubuntu-latest
environment:
name: pypi
url: https://pypi.org/p/opendht
permissions:
contents: read
id-token: write
steps:
- uses: actions/download-artifact@v8
with:
pattern: opendht-wheels-*
path: dist
merge-multiple: true
- name: Publish wheels
uses: pypa/gh-action-pypi-publish@release/v1
publish-crate:
name: Publish opendht to crates.io
needs: create-github-release
runs-on: ubuntu-latest
environment:
name: crates-io
url: https://crates.io/crates/opendht
permissions:
contents: read
id-token: write
steps:
- uses: actions/checkout@v7
- uses: taiki-e/install-action@v2
with:
tool: cargo-edit
- name: Set crate version from release tag
working-directory: rust
env:
RELEASE_TAG: ${{ github.ref_name }}
run: cargo set-version "${RELEASE_TAG#v}"
- name: Authenticate with crates.io
id: auth
uses: rust-lang/crates-io-auth-action@v1
- name: Publish crate
working-directory: rust
env:
CARGO_REGISTRY_TOKEN: ${{ steps.auth.outputs.token }}
run: cargo publish --allow-dirty
+1 -1
View File
@@ -133,7 +133,7 @@ the assembled npm archive on Node 22, 24, and 26 before release attachment.
Enable `OPENDHT_C=ON` to build the C ABI. With `BUILD_TESTING=ON`, CTest runs
`test_c`; Meson registers the same test as `CBindings` when `c` and `tests` are
enabled. Rust regression and callback-lifetime tests run with Cargo; see
[the Rust binding guide](rust/README.md) for build and library-path settings.
[the Rust build guide](rust/BUILD.md) for build and library-path settings.
`dht_infohash` and `dht_pkid` have `uint32_t` alignment, matching the native hash
types, while retaining their 20-byte and 32-byte sizes. Rebuild consumers against
+4 -4
View File
@@ -10,10 +10,10 @@ if(POLICY CMP0167)
endif()
set (opendht_VERSION_MAJOR 4)
set (opendht_VERSION_MINOR 4.99)
set (opendht_VERSION_MINOR 5.0)
set (opendht_VERSION ${opendht_VERSION_MAJOR}.${opendht_VERSION_MINOR})
set (PACKAGE_VERSION ${opendht_VERSION})
set (VERSION "${opendht_VERSION}")
set (PACKAGE_VERSION "${opendht_VERSION}-rc.1")
set (VERSION "${PACKAGE_VERSION}")
project (opendht
VERSION ${opendht_VERSION}
DESCRIPTION "OpenDHT: A lightweight C++17 Distributed Hash Table library."
@@ -421,7 +421,7 @@ else ()
endif ()
set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -DMSGPACK_NO_BOOST -DMSGPACK_DISABLE_LEGACY_NIL -DMSGPACK_DISABLE_LEGACY_CONVERT")
add_definitions(-DPACKAGE_VERSION="${opendht_VERSION}")
add_definitions(-DPACKAGE_VERSION="${PACKAGE_VERSION}")
if (ASIO_INCLUDE_DIR)
include_directories (SYSTEM "${ASIO_INCLUDE_DIR}")
+5 -2
View File
@@ -4,7 +4,10 @@
<a id="user-content-opendht-" class="anchor" href="/savoirfairelinux/opendht/blob/master/README.md#opendht-" aria-hidden="true"></a>OpenDHT
</h1>
![PyPI - Version](https://img.shields.io/pypi/v/opendht?style=flat)
[![PyPI - Version](https://img.shields.io/pypi/v/opendht?style=flat)](https://pypi.org/project/opendht/)
[![crates.io - Version](https://img.shields.io/crates/v/opendht?style=flat)](https://crates.io/crates/opendht)
[![npm - Version](https://img.shields.io/npm/v/opendht?style=flat)](https://www.npmjs.com/package/opendht)
[![Docker Image](https://img.shields.io/badge/Docker-ghcr.io-2496ED?style=flat&logo=docker&logoColor=white)](https://github.com/savoirfairelinux/opendht/pkgs/container/opendht%2Fopendht)
A lightweight C++17 Distributed Hash Table implementation.
@@ -17,7 +20,7 @@ Values are distributed over the network, with redundancy.
* Public key cryptography layer providing optional data signature and encryption (using GnuTLS)
* IPv4 and IPv6 support
* Clean and powerful **C++17** map API
* Bindings for **C, Rust & Python 3**
* Bindings for **C, Rust, Node.js & Python 3**
* REST API with optional HTTP client+server with push notification support
## Documentation
+1
View File
@@ -8,6 +8,7 @@ RUN cd opendht && mkdir build && cd build \
&& cmake -GNinja .. -DCMAKE_INSTALL_PREFIX=/usr \
-DBUILD_TESTING=Off \
-DCMAKE_INTERPROCEDURAL_OPTIMIZATION=On \
-DCMAKE_BUILD_TYPE=RelWithDebInfo \
-DOPENDHT_C=On \
-DOPENDHT_PEER_DISCOVERY=On \
-DOPENDHT_PYTHON=On \
+1
View File
@@ -10,6 +10,7 @@ ENV DESTDIR=/install
RUN cd opendht && mkdir build && cd build \
&& cmake .. -DCMAKE_INSTALL_PREFIX=/usr \
-DCMAKE_INTERPROCEDURAL_OPTIMIZATION=On \
-DCMAKE_BUILD_TYPE=RelWithDebInfo \
-DBUILD_TESTING=Off \
-DOPENDHT_C=On \
-DOPENDHT_PEER_DISCOVERY=On \
+1
View File
@@ -8,6 +8,7 @@ RUN cd opendht && mkdir build && cd build \
&& cmake .. -DCMAKE_INSTALL_PREFIX=/usr \
-DBUILD_TESTING=Off \
-DCMAKE_INTERPROCEDURAL_OPTIMIZATION=On \
-DCMAKE_BUILD_TYPE=RelWithDebInfo \
-DOPENDHT_C=On \
-DOPENDHT_PEER_DISCOVERY=On \
-DOPENDHT_PYTHON=On \
+1
View File
@@ -7,6 +7,7 @@ COPY . opendht
RUN cd opendht && mkdir build && cd build \
&& cmake .. -DCMAKE_INSTALL_PREFIX=/usr \
-DCMAKE_INTERPROCEDURAL_OPTIMIZATION=On \
-DCMAKE_BUILD_TYPE=RelWithDebInfo \
-DOPENDHT_C=On \
-DOPENDHT_PEER_DISCOVERY=On \
-DOPENDHT_PYTHON=On \
+1
View File
@@ -8,6 +8,7 @@ RUN cd opendht && mkdir build && cd build \
&& cmake .. -DCMAKE_INSTALL_PREFIX=/usr \
-DBUILD_TESTING=Off \
-DCMAKE_INTERPROCEDURAL_OPTIMIZATION=On \
-DCMAKE_BUILD_TYPE=RelWithDebInfo \
-DOPENDHT_C=On \
-DOPENDHT_PEER_DISCOVERY=On \
-DOPENDHT_PYTHON=On \
+5 -1
View File
@@ -19,7 +19,9 @@
#include <mutex>
namespace dht {
enum class PushType { None = 0, Android, iOS, UnifiedPush };
// Appended only: the values are serialized as integers by MSGPACK_ADD_ENUM,
// so reordering would break compatibility with deployed proxies.
enum class PushType { None = 0, Android, iOS, UnifiedPush, Huawei };
}
MSGPACK_ADD_ENUM(dht::PushType)
@@ -132,6 +134,7 @@ public:
PushStats androidPush;
PushStats iosPush;
PushStats unifiedPush;
PushStats huaweiPush;
/** Average requests per second */
double requestRate {0};
@@ -397,6 +400,7 @@ private:
PushStats androidPush_;
PushStats iosPush_;
PushStats unifiedPush_;
PushStats huaweiPush_;
// Thread-safe access to listeners map.
std::mutex lockListener_;
+7 -2
View File
@@ -5,7 +5,7 @@ project(
'cython',
license: 'MIT',
license_files: 'LICENSE',
version: '4.4.99',
version: '4.5.0-rc.1',
meson_version: '>=1.1.0',
default_options: ['cpp_std=c++20', 'warning_level=3'],
)
@@ -55,7 +55,12 @@ if host_machine.system() == 'linux' and io_uring.found()
endif
conf_data = configuration_data()
add_project_arguments('-DMSGPACK_NO_BOOST', '-DASIO_STANDALONE', language: 'cpp')
add_project_arguments(
'-DMSGPACK_NO_BOOST',
'-DASIO_STANDALONE',
'-DPACKAGE_VERSION="' + meson.project_version() + '"',
language: 'cpp',
)
if io_uring.found()
add_project_arguments('-DASIO_HAS_IO_URING', '-DASIO_DISABLE_EPOLL', language: 'cpp')
conf_data.set('iouring_lib', ', liburing')
+1
View File
@@ -6,6 +6,7 @@ set(CMAKE_CXX_STANDARD_REQUIRED ON)
set(CMAKE_POSITION_INDEPENDENT_CODE ON)
set(OPENDHT_TOOLS OFF CACHE BOOL "Build DHT tools")
set(OPENDHT_PYTHON OFF CACHE BOOL "Build Python bindings")
set(OPENDHT_PROXY_CLIENT ON CACHE BOOL "Enable DHT proxy client")
set(OPENDHT_PEER_DISCOVERY OFF CACHE BOOL "Enable multicast peer discovery")
set(OPENDHT_DOCUMENTATION OFF CACHE BOOL "Build documentation")
set(BUILD_TESTING OFF CACHE BOOL "Build native tests")
+70 -60
View File
@@ -1,13 +1,19 @@
# OpenDHT Node.js Bindings
Initial Node-API bindings using `node-addon-api` directly over the C++ library.
The package is private while its API and binary distribution are being developed.
Node.js 22.9 or later is required; the addon targets Node-API version 8.
## Install
```sh
npm install opendht
```
## Build
Install the native dependencies described in [BUILD.md](../BUILD.md), plus Node.js,
a C++17 compiler, and CMake 3.21 or later. Ninja is optional.
a C++17 compiler, and CMake 3.21 or later. Ninja is optional. Proxy-client support
defaults to enabled and requires ASIO, llhttp, RESTinio, and expected-lite headers.
```sh
cd nodejs
@@ -26,47 +32,20 @@ Native CMake options can be passed through, for example:
npm run build -- --CDOPENDHT_PEER_DISCOVERY=ON
```
For an existing build, change cached options directly with CMake, for example
`cmake -S . -B build -DOPENDHT_PROXY_CLIENT=ON`, then rebuild. A custom build can
disable proxy support with `-DOPENDHT_PROXY_CLIENT=OFF`; requesting a proxy then
throws an explicit unsupported-build error.
The loader checks local builds, then compatible Node-API binaries under `prebuilds/`.
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
import { DhtRunner, hash } from './nodejs/index.js';
import { DhtRunner, hash } from 'opendht';
async function main() {
const writer = new DhtRunner();
@@ -79,7 +58,7 @@ async function main() {
const key = hash('example');
const stored = await writer.put(key, Buffer.from('hello'), { id: 42n });
if (!stored) throw new Error('DHT put failed');
for (const value of await reader.get(key)) {
for await (const value of reader.get(key)) {
console.log(value.id, value.data.toString());
}
} finally {
@@ -98,6 +77,7 @@ main().catch(console.error);
- `run({port, ipv4, ipv6, network, persistPath, identity})` starts a threaded node. Port 0
selects a free port. The defaults bind both IP families on all interfaces.
`persistPath` is optional; use an empty string to disable persistence.
`proxyServer` and `proxyUserAgent` configure proxy mode (see below).
- `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'`.
@@ -108,12 +88,16 @@ main().catch(console.error);
peer is connected; subsequent DHT operations report their own completion.
- `put(key, Uint8Array, {id, type, userType, permanent})` copies the input and
resolves to the native success boolean. IDs are unsigned 64-bit `bigint` values.
- `get(key)` resolves to value snapshots, or rejects with `ERR_DHT_GET_FAILED`.
- `get(key)` returns an async iterator of values as they arrive. The iterator
ends when the lookup completes, or rejects with `ERR_DHT_GET_FAILED` on failure.
`getAll(key)` returns a single promise resolving to the complete value array.
Values contain `id`, `type`, `seq`, `userType`, copied `Buffer` data, `isSigned`,
`isEncrypted`, and an `owner` long fingerprint (or null for unsigned values).
- `listen(key, (value, expired) => {})` returns a listener with idempotent
`cancel()`. Queued deliveries are suppressed after cancellation. Callback
return values are ignored; callbacks may cancel or close the runner.
- `listen(key)` without a callback returns an async iterator of `{value, expired}`
events. It continues until cancelled or the runner closes.
- `close()` is idempotent and returns the same promise on repeated calls. It
waits off the JS thread for graceful shutdown, then joins the native threads.
After two seconds it forces the join if graceful shutdown has not completed.
@@ -124,16 +108,51 @@ 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/status handlers keep their
Pending operations, active listeners, and registered status handlers keep their
runner alive until completion, cancellation, handler removal, or explicit close.
Graceful persistence cannot be guaranteed if the process is killed.
The cross-thread queue is limited to 1,024 events and 16 MiB, and each accumulated
get result is limited to 16 MiB and 65,536 values. Exceeding a limit closes the
runner, emits an `error` event with code `ERR_DHT_OVERFLOW`, and rejects pending
operations and `close()`. Install an `error` handler on long-lived runners. As
with other Node EventEmitters, an unhandled `error` event is fatal. There is no
silent dropping of events while continuing an apparently healthy listener.
The binding does not cap the callback queue, accumulated `getAll()` results, or
iterator buffers. Slow iterator consumers can accumulate values without
backpressure. Queued callbacks retain their native values until delivery or cleanup.
## Async Iteration
`get()` and callback-free `listen()` start immediately and return iterators with
`next()`, `return()`, `throw()`, `cancel()`, and `Symbol.asyncDispose`.
Leaving a `for await` loop with `break` or an exception cancels the subscription.
Explicit cancellation discards buffered values and settles pending reads as done.
Closing the runner rejects active get iterators with `ERR_DHT_CLOSED`; listener
iterators end normally. Successful get completion preserves unread buffered values.
```js
for await (const value of runner.get(hash('example'))) {
console.log(value.id, value.data);
}
const snapshots = await runner.getAll(hash('example'));
for await (const { value, expired } of runner.listen(hash('updates'))) {
console.log(value.id, expired);
if (expired) break;
}
```
## Proxy Configuration
Pass the URL of an OpenDHT HTTP(S) proxy at startup:
```js
runner.run({
proxyServer: 'https://proxy.example.com',
proxyUserAgent: 'my-app/1.0',
});
```
This selects the native OpenDHT proxy client, not an HTTP CONNECT or SOCKS proxy.
The node does not bind UDP sockets in this mode; `port`, `ipv4`, `ipv6`, and
`bootstrap()` are for direct DHT nodes. Omit `proxyServer` or use an empty string
for direct networking. Proxy settings are startup-only; create a new runner to
change transports or servers. Packaged binaries include proxy-client support.
## Crypto
@@ -162,7 +181,7 @@ integer between 2048 and 8192; the default is 2048.
transparently decrypted for the recipient by OpenDHT's get/listen layer.
```js
import { Identity, DhtRunner } from './nodejs/index.js';
import { Identity, DhtRunner } from 'opendht';
async function createAuthenticatedNode() {
const identity = await Identity.generate('example', { algorithm: 'rsa', bits: 2048 });
@@ -173,7 +192,7 @@ async function createAuthenticatedNode() {
```
Crypto methods take Uint8Array/Buffer inputs, not implicit text encodings. Imports
must be nonempty and crypto inputs are limited to 16 MiB. Malformed handles/data,
must be nonempty. Malformed handles/data,
mismatched identities, and unsupported encryption keys raise JS exceptions.
Only generation is asynchronous; import/export/sign/encrypt/decrypt are
synchronous and should be used from worker threads for large workloads.
@@ -212,22 +231,13 @@ The [Node.js package workflow](../.github/workflows/nodejs-package.yml) builds:
| macOS ARM64 and x64 | macOS 15 |
| Windows x64 | Windows Server 2022 / MSVC |
The Linux build uses `ghcr.io/savoirfairelinux/opendht/opendht-deps-noble`,
built from [the Noble Dockerfile](../docker/DockerfileDepsNoble), restored from
the Ubuntu 24.04 recipe in git history. The release workflow publishes this
multi-architecture image alongside the other Docker images. Node builds use its
published `latest` tag. simdutf is linked statically because Ubuntu 24.04 does not
provide its development package.
CI combines platform prebuilds into one npm archive and verifies that archive on
Node 22, 24, and 26 on each platform, including Ubuntu 24.04 and 26.04. Published
GitHub releases receive the verified archive only after all install tests pass.
The package remains private;
the workflow does not publish to the npm registry or require an npm token.
Node 22, 24, and 26 on each platform, including Ubuntu 24.04 and 26.04. Branch
pushes, pull requests, and manual runs build and test without publishing.
musl/Alpine and other architectures are not currently built. Node-API stability
does not remove OS/libc or native dependency compatibility requirements. Review
the included dependency licenses and source-distribution obligations before
redistributing binaries beyond these development/release artifacts.
redistributing binaries.
Query filters, async iterators, AbortSignal support, certificate trust-store
validation, CA issuance controls, and proxy controls remain future work.
Query filters, AbortSignal support, certificate trust-store validation, CA
issuance controls, and runtime proxy switching remain future work.
+139 -182
View File
@@ -1,19 +1,29 @@
#include <napi.h>
#include <opendht.h>
#include <gnutls/gnutls.h>
#include "crypto.h"
#include <algorithm>
#include <atomic>
#include <chrono>
#include <cmath>
#include <condition_variable>
#include <functional>
#include <mutex>
#include <optional>
#include <unordered_map>
namespace {
constexpr size_t maxQueuedEvents = 1024;
constexpr size_t maxResultBytes = 16 * 1024 * 1024;
struct GnuTlsContext
{
const int result = gnutls_global_init();
~GnuTlsContext()
{
if (result == GNUTLS_E_SUCCESS)
gnutls_global_deinit();
}
};
using opendht_node::guarded;
@@ -70,64 +80,30 @@ optionsObject(Napi::Value value)
return value.As<Napi::Object>();
}
struct ValueData
Napi::Object
valueToJS(Napi::Env env, const dht::Value& value)
{
explicit ValueData(const dht::Value& value)
: id(value.id)
, type(value.type)
, seq(value.seq)
, data(value.data)
, userType(value.user_type)
, owner(value.owner ? value.owner->getLongId().toString() : "")
, isSigned(value.isSigned())
, isEncrypted(value.isEncrypted())
{}
Napi::Object toJS(Napi::Env env) const
{
auto result = Napi::Object::New(env);
result.Set("id", Napi::BigInt::New(env, id));
result.Set("type", Napi::Number::New(env, type));
result.Set("seq", Napi::Number::New(env, seq));
result.Set("data", Napi::Buffer<uint8_t>::Copy(env, data.data(), data.size()));
result.Set("userType", Napi::String::New(env, userType));
result.Set("owner", owner.empty() ? Napi::Value(env.Null()) : Napi::Value(Napi::String::New(env, owner)));
result.Set("isSigned", Napi::Boolean::New(env, isSigned));
result.Set("isEncrypted", Napi::Boolean::New(env, isEncrypted));
return result;
}
size_t size() const { return sizeof(ValueData) + data.size() + userType.size() + owner.size(); }
uint64_t id;
uint16_t type;
uint16_t seq;
dht::Blob data;
std::string userType;
std::string owner;
bool isSigned;
bool isEncrypted;
};
struct Event
{
enum class Kind { Value, Done, Overflow, Status };
Kind kind;
uint64_t operation;
bool flag;
std::optional<ValueData> value;
dht::NodeStatus ipv4 = dht::NodeStatus::Disconnected;
dht::NodeStatus ipv6 = dht::NodeStatus::Disconnected;
size_t size() const { return sizeof(Event) + (value ? value->size() : 0); }
};
auto result = Napi::Object::New(env);
result.Set("id", Napi::BigInt::New(env, value.id));
result.Set("type", Napi::Number::New(env, value.type));
result.Set("seq", Napi::Number::New(env, value.seq));
result.Set("data", Napi::Buffer<uint8_t>::Copy(env, value.data.data(), value.data.size()));
result.Set("userType", Napi::String::New(env, value.user_type));
result.Set("owner",
value.owner ? Napi::Value(Napi::String::New(env, value.owner->getLongId().to_view()))
: Napi::Value(env.Null()));
result.Set("isSigned", Napi::Boolean::New(env, value.isSigned()));
result.Set("isEncrypted", Napi::Boolean::New(env, value.isEncrypted()));
return result;
}
struct PendingOperation
{
Napi::Promise::Deferred deferred;
bool get;
std::vector<ValueData> values;
size_t bytes = 0;
std::vector<std::shared_ptr<const dht::Value>> values;
Napi::FunctionReference onValue;
std::shared_ptr<std::atomic_bool> cancelled;
};
struct Subscription
@@ -139,6 +115,8 @@ struct Subscription
struct RunnerState
{
using Callback = std::function<void(RunnerState&)>;
explicit RunnerState(Napi::Env environment)
: env(environment)
{}
@@ -203,99 +181,49 @@ struct RunnerState
}
}
void send(std::unique_ptr<Event> event)
{
if (closing || overflowed)
return;
const auto eventSize = event->size();
const auto previousCount = queuedEvents.fetch_add(1);
const auto previousBytes = queuedBytes.fetch_add(eventSize);
if (previousCount >= maxQueuedEvents || eventSize > maxResultBytes
|| previousBytes > maxResultBytes - eventSize) {
queuedEvents--;
queuedBytes -= eventSize;
signalOverflow();
return;
}
if (dispatcher.NonBlockingCall(static_cast<void*>(event.get())) == napi_ok) {
event.release();
} else {
queuedEvents--;
queuedBytes -= eventSize;
}
}
void sendValue(uint64_t operation, const dht::Value& value, bool expired = false)
{
if (closing || overflowed)
return;
if (value.data.size() > maxResultBytes || value.user_type.size() > maxResultBytes) {
signalOverflow();
return;
}
send(std::make_unique<Event>(Event {Event::Kind::Value, operation, expired, ValueData(value)}));
}
void signalOverflow()
{
if (overflowed.exchange(true))
return;
auto event = std::make_unique<Event>(Event {Event::Kind::Overflow, 0, false, {}});
if (dispatcher.NonBlockingCall(static_cast<void*>(event.get())) == napi_ok)
event.release();
}
void handle(Event& event)
void send(Callback callback)
{
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();
if (!onError.IsEmpty())
onError.Call({makeCloseError().Value()});
return;
}
if (event.kind == Event::Kind::Value) {
const auto listener = listeners.find(event.operation);
if (listener != listeners.end()) {
listener->second.callback.Call({event.value->toJS(env), Napi::Boolean::New(env, event.flag)});
} else if (const auto pending = operations.find(event.operation); pending != operations.end()) {
auto& operation = pending->second;
operation.bytes += event.value->size();
if (operation.bytes > maxResultBytes || operation.values.size() >= 65536) {
overflowed = true;
startClose();
if (!onError.IsEmpty())
onError.Call({makeCloseError().Value()});
return;
}
operation.values.emplace_back(std::move(*event.value));
auto task = std::make_unique<Callback>(std::move(callback));
if (dispatcher.NonBlockingCall(static_cast<void*>(task.get())) == napi_ok)
task.release();
}
void sendValue(uint64_t operation, std::shared_ptr<const dht::Value> value, bool expired = false)
{
send([operation, value = std::move(value), expired](RunnerState& state) mutable {
const auto listener = state.listeners.find(operation);
if (listener != state.listeners.end()) {
listener->second.callback.Call({valueToJS(state.env, *value), Napi::Boolean::New(state.env, expired)});
} else if (const auto pending = state.operations.find(operation); pending != state.operations.end()) {
if (!pending->second.onValue.IsEmpty())
pending->second.onValue.Call({valueToJS(state.env, *value)});
else
pending->second.values.emplace_back(std::move(value));
}
return;
}
const auto pending = operations.find(event.operation);
});
}
void completeOperation(uint64_t identifier, bool ok)
{
const auto pending = operations.find(identifier);
if (pending == operations.end())
return;
auto operation = std::move(pending->second);
operations.erase(pending);
if (!operation.get) {
operation.deferred.Resolve(Napi::Boolean::New(env, event.flag));
} else if (!event.flag) {
operation.deferred.Resolve(Napi::Boolean::New(env, ok));
} else if (!ok) {
auto error = Napi::Error::New(env, "DHT get failed");
error.Set("code", "ERR_DHT_GET_FAILED");
operation.deferred.Reject(error.Value());
} else if (!operation.onValue.IsEmpty()) {
operation.deferred.Resolve(env.Undefined());
} else {
auto values = Napi::Array::New(env, operation.values.size());
for (size_t index = 0; index < operation.values.size(); ++index)
values.Set(static_cast<uint32_t>(index), operation.values[index].toJS(env));
values.Set(static_cast<uint32_t>(index), valueToJS(env, *operation.values[index]));
operation.deferred.Resolve(values);
}
}
@@ -303,27 +231,19 @@ 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, dht::statusToStr(closing ? dht::NodeStatus::Disconnected : status4)));
result.Set("ipv6", Napi::String::New(env, 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<RunnerState*>(context);
std::unique_ptr<Event> event(static_cast<Event*>(data));
if (event->kind != Event::Kind::Overflow) {
state->queuedEvents--;
state->queuedBytes -= event->size();
}
if (!environment || state->envClosing)
std::unique_ptr<Callback> callback(static_cast<Callback*>(data));
if (!environment || state->envClosing || state->closing)
return;
try {
state->handle(*event);
(*callback)(*state);
} catch (const Napi::Error& error) {
napi_fatal_exception(environment, error.Value());
} catch (const std::exception& error) {
@@ -371,11 +291,8 @@ struct RunnerState
Napi::Error makeCloseError() const
{
auto error = Napi::Error::New(env,
overflowed ? "OpenDHT result queue exceeded its limit"
: closeError.empty() ? "DHT runner is closed"
: closeError);
error.Set("code", overflowed ? "ERR_DHT_OVERFLOW" : "ERR_DHT_CLOSED");
auto error = Napi::Error::New(env, closeError.empty() ? "DHT runner is closed" : closeError);
error.Set("code", "ERR_DHT_CLOSED");
return error;
}
@@ -390,7 +307,7 @@ struct RunnerState
for (auto& pending : state->operations)
pending.second.deferred.Reject(error.Value());
if (state->closeDeferred) {
if (!state->overflowed && state->closeError.empty())
if (state->closeError.empty())
state->closeDeferred->Resolve(Napi::Env(environment).Undefined());
else
state->closeDeferred->Reject(error.Value());
@@ -400,7 +317,6 @@ struct RunnerState
state->listeners.clear();
state->closeDeferred.reset();
state->closePromise.Reset();
state->onError.Reset();
state->onStatus.Reset();
state->owner.Reset();
napi_delete_async_work(environment, state->closeWork);
@@ -415,9 +331,6 @@ struct RunnerState
napi_async_cleanup_hook_handle cleanupHook {};
std::shared_ptr<RunnerState>* closeOwner {};
std::atomic_bool closing {false};
std::atomic_bool overflowed {false};
std::atomic_size_t queuedEvents {0};
std::atomic_size_t queuedBytes {0};
bool envClosing = false;
bool started = false;
bool hasIdentity = false;
@@ -428,7 +341,6 @@ struct RunnerState
std::unordered_map<uint64_t, Subscription> listeners;
std::optional<Napi::Promise::Deferred> closeDeferred;
Napi::Reference<Napi::Promise> closePromise;
Napi::FunctionReference onError;
Napi::FunctionReference onStatus;
Napi::ObjectReference owner;
dht::NodeStatus status4 = dht::NodeStatus::Disconnected;
@@ -443,9 +355,7 @@ public:
, state_(RunnerState::create(info.Env()))
{
if (info.Length() && info[0].IsFunction())
state_->onError = Napi::Persistent(info[0].As<Napi::Function>());
if (info.Length() > 1 && info[1].IsFunction())
state_->onStatus = Napi::Persistent(info[1].As<Napi::Function>());
state_->onStatus = Napi::Persistent(info[0].As<Napi::Function>());
}
~Runner() override { state_->startClose(); }
@@ -463,6 +373,8 @@ public:
InstanceMethod("getStatus", &Runner::getStatus),
InstanceMethod("bootstrap", &Runner::bootstrap),
InstanceMethod("get", &Runner::get),
InstanceMethod("getAll", &Runner::getAll),
InstanceMethod("cancelGet", &Runner::cancelGet),
InstanceMethod("put", &Runner::put),
InstanceMethod("putSigned", &Runner::putSigned),
InstanceMethod("putEncrypted", &Runner::putEncrypted),
@@ -494,11 +406,23 @@ 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"));
if (options.Has("proxyServer"))
config.proxy_server = text(options.Get("proxyServer"), "proxyServer");
if (options.Has("proxyUserAgent")) {
config.proxy_user_agent = text(options.Get("proxyUserAgent"), "proxyUserAgent");
if (config.proxy_user_agent.find_first_of("\r\n") != std::string::npos)
throw Napi::TypeError::New(info.Env(), "proxyUserAgent must not contain line breaks");
}
dht::DhtRunner::Context context;
const std::weak_ptr<RunnerState> state = state_;
context.statusChangedCallback = [state](dht::NodeStatus ipv4, dht::NodeStatus ipv6) {
if (auto owner = state.lock())
owner->send(std::make_unique<Event>(Event {Event::Kind::Status, 0, false, {}, ipv4, ipv6}));
owner->send([ipv4, ipv6](RunnerState& state) {
state.status4 = ipv4;
state.status6 = ipv6;
if (!state.onStatus.IsEmpty())
state.onStatus.Call({state.status()});
});
};
const auto service = std::to_string(port);
state_->runner.run(ipv4.c_str(), ipv6.c_str(), service.c_str(), config, std::move(context));
@@ -516,7 +440,7 @@ private:
return state_->closePromise.Value();
auto deferred = Napi::Promise::Deferred::New(info.Env());
if (state_->finished) {
if (state_->overflowed || !state_->closeError.empty())
if (!state_->closeError.empty())
deferred.Reject(state_->makeCloseError().Value());
else
deferred.Resolve(info.Env().Undefined());
@@ -531,7 +455,7 @@ private:
Napi::Value isRunning(const Napi::CallbackInfo& info)
{
return Napi::Boolean::New(info.Env(), !state_->closing && !state_->overflowed && state_->runner.isRunning());
return Napi::Boolean::New(info.Env(), !state_->closing && state_->runner.isRunning());
}
Napi::Value getStatus(const Napi::CallbackInfo& info)
@@ -554,7 +478,7 @@ private:
{
return guarded(info, [&]() -> Napi::Value {
state_->requireRunning();
return Napi::String::New(info.Env(), state_->runner.getNodeId().toString());
return Napi::String::New(info.Env(), state_->runner.getNodeId().to_view());
});
}
@@ -571,39 +495,69 @@ private:
});
}
Napi::Value get(const Napi::CallbackInfo& info)
Napi::Value get(const Napi::CallbackInfo& info) { return getValues(info, true); }
Napi::Value getAll(const Napi::CallbackInfo& info) { return getValues(info, false); }
Napi::Value getValues(const Napi::CallbackInfo& info, bool stream)
{
return guarded(info, [&]() -> Napi::Value {
state_->requireRunning();
const auto key = infoHash(info[0]);
if (stream && !info[1].IsFunction())
throw Napi::TypeError::New(info.Env(), "callback must be a function");
const auto operation = state_->nextOperation++;
auto deferred = Napi::Promise::Deferred::New(info.Env());
state_->operations.emplace(operation, PendingOperation {deferred, true, {}});
auto cancelled = std::make_shared<std::atomic_bool>(false);
PendingOperation pending {deferred, true, {}, {}, cancelled};
if (stream)
pending.onValue = Napi::Persistent(info[1].As<Napi::Function>());
state_->operations.emplace(operation, std::move(pending));
const std::weak_ptr<RunnerState> state = state_;
try {
state_->runner.get(key,
dht::GetCallback(
[state, operation](const std::vector<std::shared_ptr<dht::Value>>& values) {
const auto owner = state.lock();
if (!owner || owner->closing || owner->overflowed)
return false;
for (const auto& value : values)
owner->sendValue(operation, *value);
return !owner->closing && !owner->overflowed;
}),
dht::GetCallback([state, operation, cancelled](
const std::vector<std::shared_ptr<dht::Value>>& values) {
const auto owner = state.lock();
if (!owner || owner->closing || *cancelled)
return false;
for (const auto& value : values)
owner->sendValue(operation, value);
return !owner->closing && !*cancelled;
}),
dht::DoneCallbackSimple([state, operation](bool ok) {
if (const auto owner = state.lock())
owner->send(
std::make_unique<Event>(Event {Event::Kind::Done, operation, ok, {}}));
owner->send([operation, ok](RunnerState& state) {
state.completeOperation(operation, ok);
});
}));
} catch (...) {
state_->operations.erase(operation);
throw;
}
if (stream) {
auto result = Napi::Object::New(info.Env());
result.Set("token", Napi::BigInt::New(info.Env(), operation));
result.Set("done", deferred.Promise());
return result;
}
return deferred.Promise();
});
}
Napi::Value cancelGet(const Napi::CallbackInfo& info)
{
return guarded(info, [&]() -> Napi::Value {
const auto operation = state_->operations.find(unsignedBigInt(info[0]));
if (operation != state_->operations.end() && !operation->second.onValue.IsEmpty()) {
*operation->second.cancelled = true;
operation->second.deferred.Resolve(info.Env().Undefined());
state_->operations.erase(operation);
}
return info.Env().Undefined();
});
}
Napi::Value put(const Napi::CallbackInfo& info) { return putValue(info, PutMode::Plain); }
Napi::Value putSigned(const Napi::CallbackInfo& info) { return putValue(info, PutMode::Signed); }
@@ -649,12 +603,12 @@ private:
}
const auto operation = state_->nextOperation++;
auto deferred = Napi::Promise::Deferred::New(info.Env());
state_->operations.emplace(operation, PendingOperation {deferred, false, {}});
state_->operations.emplace(operation, PendingOperation {deferred, false, {}, {}, {}});
const std::weak_ptr<RunnerState> state = state_;
try {
dht::DoneCallbackSimple done = [state, operation](bool ok) {
if (const auto owner = state.lock())
owner->send(std::make_unique<Event>(Event {Event::Kind::Done, operation, ok, {}}));
owner->send([operation, ok](RunnerState& state) { state.completeOperation(operation, ok); });
};
if (mode == PutMode::Signed)
state_->runner.putSigned(key, value, std::move(done), permanent);
@@ -685,11 +639,11 @@ private:
[state, operation](const std::vector<std::shared_ptr<dht::Value>>& values,
bool expired) {
const auto owner = state.lock();
if (!owner || owner->closing || owner->overflowed)
if (!owner || owner->closing)
return false;
for (const auto& value : values)
owner->sendValue(operation, *value, expired);
return !owner->closing && !owner->overflowed;
owner->sendValue(operation, value, expired);
return !owner->closing;
}))
.share();
state_->listeners.emplace(operation,
@@ -759,12 +713,15 @@ hash(const Napi::CallbackInfo& info)
} else {
throw Napi::TypeError::New(env, "Expected a string or Uint8Array");
}
return Napi::String::New(env, result.toString());
return Napi::String::New(env, result.to_view());
}
Napi::Object
initialize(Napi::Env env, Napi::Object exports)
{
static const GnuTlsContext crypto;
if (crypto.result != GNUTLS_E_SUCCESS)
throw Napi::Error::New(env, std::string("Unable to initialize GnuTLS: ") + gnutls_strerror(crypto.result));
opendht_node::initializeCrypto(env, exports);
exports.Set("NativeRunner", Runner::define(env));
exports.Set("hash", Napi::Function::New(env, hash));
-2
View File
@@ -51,8 +51,6 @@ bytes(Napi::Value value)
if (!value.IsTypedArray() || value.As<Napi::TypedArray>().TypedArrayType() != napi_uint8_array)
throw Napi::TypeError::New(value.Env(), "Expected a Uint8Array");
auto array = value.As<Napi::Uint8Array>();
if (array.ElementLength() > 16 * 1024 * 1024)
throw Napi::RangeError::New(value.Env(), "Crypto input exceeds 16 MiB");
dht::Blob result(array.ElementLength());
if (!result.empty())
std::copy_n(array.Data(), result.size(), result.data());
+1 -1
View File
@@ -96,7 +96,7 @@ Keys are text labels hashed with SHA-1. Values are UTF-8 text.`);
const location = hash(key);
switch (command) {
case 'get': {
const found = await runner.get(location);
const found = await runner.getAll(location);
if (closing) return;
for (const value of found) printValue(key, value);
if (!found.length) print(`No values for ${key}`);
+17 -1
View File
@@ -49,6 +49,8 @@ export interface RunOptions {
network?: number;
persistPath?: string;
identity?: Identity;
proxyServer?: string;
proxyUserAgent?: string;
}
export interface PutOptions {
@@ -73,6 +75,18 @@ export interface Listener {
cancel(): void;
}
export interface ListenEvent {
value: Value;
expired: boolean;
}
export interface ValueIterator<T> extends AsyncIterableIterator<T, undefined>, Listener, AsyncDisposable {
return(): Promise<IteratorResult<T, undefined>>;
throw(error?: unknown): Promise<IteratorResult<T, undefined>>;
}
export interface ListenerIterator extends ValueIterator<ListenEvent> { }
export type NodeStatus = 'disconnected' | 'connecting' | 'connected';
export interface ConnectivityStatus {
@@ -92,10 +106,12 @@ export class DhtRunner extends EventEmitter<RunnerEvents> implements AsyncDispos
getNodeId(): string;
getStatus(): ConnectivityStatus;
bootstrap(host: string, port: number): void;
get(key: string): Promise<Value[]>;
get(key: string): ValueIterator<Value>;
getAll(key: string): Promise<Value[]>;
put(key: string, data: Uint8Array, options?: PutOptions): Promise<boolean>;
putSigned(key: string, data: Uint8Array, options?: PutOptions): Promise<boolean>;
putEncrypted(key: string, data: Uint8Array, recipient: PublicKey, options?: PutOptions): Promise<boolean>;
listen(key: string): ListenerIterator;
listen(key: string, callback: (value: Value, expired: boolean) => void): Listener;
close(): Promise<void>;
ref(): this;
+105 -6
View File
@@ -1,10 +1,15 @@
import { EventEmitter } from 'node:events';
import { constants } from 'node:os';
import { dlopen } from 'node:process';
import load from 'node-gyp-build';
import createCrypto from './crypto.js';
const binding = { exports: {} };
dlopen(binding, process.env.OPENDHT_NODE_BINDING || load.path(import.meta.dirname));
const flags = process.platform === 'linux'
? constants.dlopen.RTLD_LAZY | constants.dlopen.RTLD_DEEPBIND
: undefined;
dlopen(binding, process.env.OPENDHT_NODE_BINDING || load.path(import.meta.dirname),
flags);
const native = binding.exports;
const { PublicKey, PrivateKey, Certificate, Identity, handle } = createCrypto(native);
@@ -17,16 +22,16 @@ class DhtRunner extends EventEmitter {
#closing;
#listeners = 0;
#pending = 0;
#iterators = new Set();
constructor() {
super();
const owner = new WeakRef(this);
this.#native = new native.NativeRunner(eventReporter(owner, 'error'), eventReporter(owner, 'statusChanged'));
this.#native = new native.NativeRunner(eventReporter(owner, 'statusChanged'));
}
#updateOwner() {
this.#native.setOwner(this.#listeners || this.#pending || this.listenerCount('error')
|| this.listenerCount('statusChanged') ? this : null);
this.#native.setOwner(this.#listeners || this.#pending || this.listenerCount('statusChanged') ? this : null);
}
#track(operation) {
@@ -46,6 +51,11 @@ class DhtRunner extends EventEmitter {
removeAllListeners(...args) { super.removeAllListeners(...args); this.#updateOwner(); return this; }
run(options) {
if (typeof options?.proxyServer === 'string' && options.proxyServer) {
const proxy = new URL(options.proxyServer);
if (proxy.protocol !== 'http:' && proxy.protocol !== 'https:')
throw new TypeError('proxyServer must be an HTTP(S) URL');
}
if (options?.identity !== undefined) options = { ...options, identity: handle(options.identity, Identity) };
this.#native.run(options);
}
@@ -54,7 +64,14 @@ class DhtRunner extends EventEmitter {
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)); }
get(key) {
return this.#iterate((push, finish) => {
const operation = this.#native.get(key, push);
this.#track(operation.done).then(() => finish(), finish);
return () => this.#native.cancelGet(operation.token);
}, true);
}
getAll(key) { return this.#track(this.#native.getAll(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) {
@@ -62,6 +79,12 @@ class DhtRunner extends EventEmitter {
}
listen(key, callback) {
if (callback === undefined) {
return this.#iterate(push => {
const subscription = this.listen(key, (value, expired) => push({ value, expired }));
return () => subscription.cancel();
});
}
const token = this.#native.listen(key, callback);
this.#listeners++;
this.#updateOwner();
@@ -77,7 +100,83 @@ class DhtRunner extends EventEmitter {
};
}
close() { return this.#closing ??= this.#native.close(); }
#iterate(subscribe, rejectOnClose = false) {
let values = [];
let offset = 0;
const pending = [];
let done = false;
let failure;
const finish = error => {
if (done) return;
done = true;
failure = error;
unsubscribe();
this.#iterators.delete(onClose);
if (failure) {
values = [];
offset = 0;
}
for (const { resolve, reject } of pending.splice(0)) {
if (failure) reject(failure);
else resolve({ value: undefined, done: true });
}
};
const cancel = () => {
values = [];
offset = 0;
failure = undefined;
finish();
};
const onClose = () => {
if (rejectOnClose) {
const error = new Error('DHT runner is closed');
error.code = 'ERR_DHT_CLOSED';
finish(error);
} else cancel();
};
const unsubscribe = subscribe(value => {
if (done) return;
const waiter = pending.shift();
if (waiter) waiter.resolve({ value, done: false });
else values.push(value);
}, finish);
this.#iterators.add(onClose);
return {
next() {
if (failure) return Promise.reject(failure);
if (offset < values.length) {
const value = values[offset];
values[offset++] = undefined;
if (offset === values.length) {
values = [];
offset = 0;
}
return Promise.resolve({ value, done: false });
}
if (done) return Promise.resolve({ value: undefined, done: true });
return new Promise((resolve, reject) => pending.push({ resolve, reject }));
},
return() {
cancel();
return Promise.resolve({ value: undefined, done: true });
},
throw(error) {
cancel();
return Promise.reject(error);
},
cancel,
[Symbol.asyncIterator]() { return this; },
[Symbol.asyncDispose]() { cancel(); return Promise.resolve(); },
};
}
close() {
if (!this.#closing) {
this.#closing = this.#native.close();
for (const cancel of this.#iterators) cancel();
}
return this.#closing;
}
ref() { this.#native.ref(); return this; }
unref() { this.#native.unref(); return this; }
[Symbol.asyncDispose]() { return this.close(); }
+10 -2
View File
@@ -1,9 +1,17 @@
{
"name": "opendht",
"version": "0.1.0",
"private": true,
"description": "Distributed hash table for Node.js",
"license": "MIT",
"repository": {
"type": "git",
"url": "git+https://github.com/savoirfairelinux/opendht.git",
"directory": "nodejs"
},
"publishConfig": {
"access": "public",
"registry": "https://registry.npmjs.org/"
},
"type": "module",
"main": "index.js",
"types": "index.d.ts",
@@ -44,7 +52,7 @@
"devDependencies": {
"@types/node": "^22.20.5",
"cmake-js": "^8.0.0",
"node-addon-api": "^8.5.0",
"node-addon-api": "^8.8.0",
"typescript": "^7.0.2"
}
}
+3
View File
@@ -27,6 +27,9 @@ function copyNotices(dependencies) {
name = match[2];
notices = fs.readdirSync(match[1]).filter(file => /^(COPYING|LICENSE|COPYRIGHT)([._-]|$)/i.test(file))
.map(file => path.join(match[1], file));
} else if (process.platform === 'linux' && /^libllhttp\.so(?:\.|$)/.test(path.basename(real))) {
name = 'llhttp';
notices = [path.join(repository, 'resources', 'llhttp-LICENSE-MIT')];
} else if (process.platform === 'linux') {
let owner;
try { owner = run('dpkg-query', ['-S', real]); }
+266 -26
View File
@@ -1,6 +1,7 @@
import assert from 'node:assert/strict';
import { execFile } from 'node:child_process';
import { EventEmitter, once } from 'node:events';
import { createServer } from 'node:http';
import { test } from 'node:test';
import { fileURLToPath } from 'node:url';
import { promisify } from 'node:util';
@@ -29,6 +30,7 @@ test('runner lifecycle and argument validation', async () => {
const runner = new DhtRunner();
assert.equal(runner.isRunning(), false);
assert.throws(() => runner.get(hash('idle')), /running/i);
assert.throws(() => runner.getAll(hash('idle')), /running/i);
assert.throws(() => runner.run({ port: -1 }), RangeError);
assert.throws(() => runner.run({ port: 65536 }), RangeError);
assert.throws(() => runner.run({ port: 1.5 }), RangeError);
@@ -111,6 +113,7 @@ test('two loopback nodes exchange binary values and cancel listeners', { timeout
const key = hash('node-addon-api integration');
const payload = Buffer.from([0, 1, 255, 0, 42]);
let subscription;
let retainedValue;
try {
writer.bootstrap('127.0.0.1', reader.getBoundPort());
reader.bootstrap('127.0.0.1', writer.getBoundPort());
@@ -126,24 +129,33 @@ test('two loopback nodes exchange binary values and cancel listeners', { timeout
}), true);
const [value] = await received;
assert.equal(value.id, 18446744073709551614n);
assert.equal(value.type, 0);
assert.ok(Buffer.isBuffer(value.data));
assert.deepEqual(value.data, payload);
assert.equal(value.userType, 'nodejs-test');
assert.equal(typeof value.seq, 'number');
const values = await reader.get(key);
assert.ok(values.some(candidate => candidate.id === value.id && candidate.data.equals(payload)));
assert.equal(value.owner, null);
assert.equal(value.isSigned, false);
assert.equal(value.isEncrypted, false);
value.data.fill(7);
const values = await reader.getAll(key);
retainedValue = values.find(candidate => candidate.id === value.id);
assert.ok(retainedValue);
assert.deepEqual(retainedValue.data, payload);
subscription.cancel();
subscription.cancel();
} finally {
subscription?.cancel();
await Promise.all([writer.close(), reader.close()]);
}
assert.deepEqual(retainedValue.data, payload);
});
test('close settles pending operations and suppresses queued listener deliveries', { timeout: 10000 }, async () => {
const runner = startNode();
let deliveries = 0;
const subscription = runner.listen(hash('closing'), () => deliveries++);
const pending = runner.get(hash('unreachable'));
const pending = runner.getAll(hash('unreachable'));
const settled = assert.rejects(pending, { code: 'ERR_DHT_CLOSED' });
await runner.close();
await settled;
@@ -151,6 +163,236 @@ test('close settles pending operations and suppresses queued listener deliveries
assert.equal(deliveries, 0);
});
test('get iterators settle pending reads on cancellation and reject on runner close', { timeout: 10000 }, async () => {
const runner = startNode();
try {
const cancelled = runner.get(hash('cancel get'));
assert.equal(cancelled[Symbol.asyncIterator](), cancelled);
const pending = [cancelled.next(), cancelled.next()];
await cancelled.return();
cancelled.cancel();
assert.deepEqual(await Promise.all(pending), [
{ value: undefined, done: true },
{ value: undefined, done: true },
]);
assert.deepEqual(await cancelled.next(), { value: undefined, done: true });
const closing = runner.get(hash('close get'));
const rejected = assert.rejects(closing.next(), { code: 'ERR_DHT_CLOSED' });
await runner.close();
await rejected;
await assert.rejects(closing.next(), { code: 'ERR_DHT_CLOSED' });
} finally {
await runner.close();
}
});
test('async listeners deliver queued values and cancel on loop exit', { timeout: 15000 }, async () => {
const writer = startNode();
const reader = startNode();
const key = hash('async listener');
let iterator;
try {
writer.bootstrap('127.0.0.1', reader.getBoundPort());
reader.bootstrap('127.0.0.1', writer.getBoundPort());
iterator = reader.listen(key);
assert.equal(iterator[Symbol.asyncIterator](), iterator);
const pending = iterator.next();
assert.equal(await writer.put(key, Buffer.from('first'), { id: 1n }), true);
const first = await pending;
assert.equal(first.done, false);
assert.equal(first.value.expired, false);
assert.equal(first.value.value.id, 1n);
assert.equal(first.value.value.data.toString(), 'first');
assert.equal(await writer.put(key, Buffer.from('second'), { id: 2n }), true);
for await (const { value, expired } of iterator) {
assert.equal(expired, false);
assert.equal(value.id, 2n);
assert.equal(value.data.toString(), 'second');
break;
}
assert.deepEqual(await iterator.next(), { value: undefined, done: true });
iterator.cancel();
} finally {
iterator?.cancel();
await Promise.all([writer.close(), reader.close()]);
}
});
test('async listener cancellation settles concurrent reads and drops buffered values', { timeout: 10000 }, async () => {
const runner = startNode();
const peer = startNode();
const key = hash('async cancellation');
let iterator;
try {
runner.bootstrap('127.0.0.1', peer.getBoundPort());
peer.bootstrap('127.0.0.1', runner.getBoundPort());
iterator = runner.listen(key);
const pending = [iterator.next(), iterator.next()];
iterator.cancel();
iterator.cancel();
assert.deepEqual(await Promise.all(pending), [
{ value: undefined, done: true },
{ value: undefined, done: true },
]);
assert.equal(await runner.put(key, Buffer.from('cancelled'), { id: 3n }), true);
assert.deepEqual(await iterator.next(), { value: undefined, done: true });
const buffered = runner.listen(key);
await runner.getAll(key);
await buffered.return();
assert.deepEqual(await buffered.next(), { value: undefined, done: true });
} finally {
iterator?.cancel();
await Promise.all([runner.close(), peer.close()]);
}
});
test('async listener throw, disposal, and runner close settle pending reads', { timeout: 10000 }, async () => {
const runner = startNode();
try {
const thrown = runner.listen(hash('async throw'));
const pendingThrow = thrown.next();
const failure = new Error('consumer failed');
await assert.rejects(thrown.throw(failure), error => error === failure);
assert.deepEqual(await pendingThrow, { value: undefined, done: true });
assert.deepEqual(await thrown.next(), { value: undefined, done: true });
const disposed = runner.listen(hash('async dispose'));
const pendingDispose = disposed.next();
await disposed[Symbol.asyncDispose]();
assert.deepEqual(await pendingDispose, { value: undefined, done: true });
const closing = runner.listen(hash('async close'));
const pendingClose = [closing.next(), closing.next()];
const close = runner.close();
assert.equal(runner.close(), close);
assert.deepEqual(await Promise.all(pendingClose), [
{ value: undefined, done: true },
{ value: undefined, done: true },
]);
await close;
assert.deepEqual(await closing.next(), { value: undefined, done: true });
assert.throws(() => runner.listen(hash('closed iterator')), /closed/i);
} finally {
await runner.close();
}
});
test('proxy configuration validates option types before starting', async () => {
const runner = new DhtRunner();
try {
assert.throws(() => runner.run({ proxyServer: 42 }), TypeError);
assert.throws(() => runner.run({ proxyUserAgent: false }), TypeError);
assert.throws(() => runner.run({ proxyUserAgent: 'client\r\nInjected: header' }), TypeError);
assert.throws(() => runner.run({ proxyServer: 'http://127.0.0.1\0' }), TypeError);
assert.throws(() => runner.run({ proxyServer: 'ftp://127.0.0.1' }), /http/i);
assert.throws(() => runner.run({ proxyServer: 'not a URL' }), /url/i);
assert.equal(runner.isRunning(), false);
runner.run({ port: 0, ipv4: '127.0.0.1', ipv6: '::1', proxyServer: '' });
assert.ok(runner.getBoundPort() > 0);
} finally {
await runner.close();
}
});
test('proxy configuration routes operations through HTTP with a custom user agent', { timeout: 15000 }, async () => {
const runner = new DhtRunner();
const requests = [];
const values = new Map();
let streamNextGet = false;
let streamResponse;
const server = createServer(async (request, response) => {
response.setHeader('content-type', 'application/json');
if (request.url === '/') {
response.end(JSON.stringify({
node_id: hash('proxy fixture'),
ipv4: { good: 1 },
ipv6: {},
public_ip: '127.0.0.1',
}));
return;
}
requests.push({ method: request.method, url: request.url, userAgent: request.headers['user-agent'] });
if (request.url === `/key/${hash('proxy failure')}`) {
response.statusCode = 500;
response.end();
return;
}
if (request.method === 'POST') {
const chunks = [];
for await (const chunk of request) chunks.push(chunk);
const value = JSON.parse(Buffer.concat(chunks).toString());
values.set(request.url, value);
response.end(JSON.stringify(value));
} else {
const value = values.get(request.url);
const body = value ? `${JSON.stringify(value)}\n` : '';
if (streamNextGet) {
streamNextGet = false;
streamResponse = response;
response.write(body);
} else response.end(body);
}
});
server.listen(0, '127.0.0.1');
try {
await once(server, 'listening');
runner.run({
proxyServer: `http://127.0.0.1:${server.address().port}`,
proxyUserAgent: 'opendht-node-test/1.0',
persistPath: '',
});
const key = hash('proxy configuration');
const payload = Buffer.from([0, 255, 42]);
assert.equal(await runner.put(key, payload, { id: 42n }), true);
const received = await runner.getAll(key);
assert.equal(received.length, 1);
assert.equal(received[0].id, 42n);
assert.deepEqual(received[0].data, payload);
streamNextGet = true;
const iterator = runner.get(key);
const first = await iterator.next();
assert.equal(first.done, false);
assert.equal(first.value.id, 42n);
assert.deepEqual(first.value.data, payload);
assert.equal(streamResponse.writableEnded, false);
streamResponse.end();
assert.deepEqual(await iterator.next(), { value: undefined, done: true });
assert.deepEqual(await iterator.next(), { value: undefined, done: true });
assert.deepEqual(requests.map(request => request.method), ['POST', 'GET', 'GET']);
assert.ok(requests.every(request => request.url === `/key/${key}`));
assert.ok(requests.every(request => request.userAgent === 'opendht-node-test/1.0'));
const empty = runner.get(hash('proxy empty'));
assert.deepEqual(await empty.next(), { value: undefined, done: true });
const failed = runner.get(hash('proxy failure'));
await assert.rejects(failed.next(), { code: 'ERR_DHT_GET_FAILED' });
await assert.rejects(runner.getAll(hash('proxy failure')), { code: 'ERR_DHT_GET_FAILED' });
const all = [];
for await (const value of runner.get(key)) all.push(value);
assert.equal(all.length, 1);
assert.deepEqual(all[0].data, payload);
streamNextGet = true;
const cancelled = runner.get(key);
for await (const value of cancelled) {
assert.equal(value.id, 42n);
break;
}
assert.equal(streamResponse.writableEnded, false);
assert.deepEqual(await cancelled.next(), { value: undefined, done: true });
streamResponse.end();
} finally {
streamResponse?.end();
await runner.close();
await new Promise((resolve, reject) => server.close(error => error ? reject(error) : resolve()));
}
});
test('worker termination cleans up native runners', { timeout: 10000 }, async () => {
const source = `
import { parentPort } from 'node:worker_threads';
@@ -179,12 +421,15 @@ test('listener callbacks may cancel themselves and close the runner', { timeout:
subscription.cancel();
finish(runner.close());
});
const put = runner.put(key, Buffer.from('value')).catch(error => {
assert.equal(error.code, 'ERR_DHT_CLOSED');
});
const puts = [];
for (let index = 0; index < 3000; index++) {
puts.push(runner.put(key, Buffer.from('value'), { id: BigInt(index + 1) }).catch(error => {
assert.equal(error.code, 'ERR_DHT_CLOSED');
}));
}
try {
await received;
await put;
await Promise.all(puts);
assert.equal(deliveries, 1);
} finally {
await runner.close();
@@ -202,28 +447,23 @@ test('unref and garbage collection do not strand native threads', { timeout: 200
}
});
test('queue overflow reports an error and rejects close', { timeout: 15000 }, async () => {
test('callback queue completes bursts beyond the former event limit', { timeout: 15000 }, async () => {
const runner = startNode();
const key = hash('bounded listener queue');
const reported = once(runner, 'error');
const subscription = runner.listen(key, () => { });
const key = hash('unbounded callback queue');
const errors = [];
const count = 3000;
runner.on('error', error => errors.push(error));
const puts = [];
try {
for (let index = 0; index < 3000; index++) {
puts.push(runner.put(key, Buffer.from('value'), { id: BigInt(index + 1) }).catch(error => {
assert.equal(error.code, 'ERR_DHT_OVERFLOW');
}));
}
const deadline = performance.now() + 5000;
while (runner.isRunning() && performance.now() < deadline) { }
assert.equal(runner.isRunning(), false);
const [error] = await reported;
assert.equal(error.code, 'ERR_DHT_OVERFLOW');
await Promise.all(puts);
await assert.rejects(runner.close(), { code: 'ERR_DHT_OVERFLOW' });
for (let index = 0; index < count; index++)
puts.push(runner.put(key, Buffer.from('value'), { id: BigInt(index + 1) }));
const results = await Promise.all(puts);
assert.equal(results.length, count);
assert.ok(results.every(result => typeof result === 'boolean'));
assert.equal(runner.isRunning(), true);
assert.deepEqual(errors, []);
} finally {
subscription.cancel();
await runner.close().catch(() => { });
await runner.close();
}
});
@@ -272,7 +512,7 @@ test('dhtnode example handles commands and quit while a get is pending', { timeo
import { DhtRunner } from ${JSON.stringify(moduleURL)};
const close = DhtRunner.prototype.close;
let rejectGet;
DhtRunner.prototype.get = function () {
DhtRunner.prototype.getAll = function () {
console.log('Get pending');
return new Promise((resolve, reject) => { rejectGet = reject; });
};
+4 -2
View File
@@ -27,6 +27,8 @@ test('EC keys sign and invalid crypto inputs raise JS errors', async () => {
const key = await PrivateKey.generate();
const restored = new PrivateKey(key.serialize());
assert.equal(restored.getPublicKey().checkSignature(payload, key.sign(payload)), true);
const largePayload = Buffer.alloc(16 * 1024 * 1024 + 1, 42);
assert.equal(restored.getPublicKey().checkSignature(largePayload, key.sign(largePayload)), true);
assert.throws(() => key.getPublicKey().encrypt(payload));
for (const Type of [PrivateKey, PublicKey, Certificate]) {
assert.throws(() => new Type(Buffer.alloc(0)));
@@ -64,12 +66,12 @@ test('signed and encrypted values round-trip between authenticated nodes', { tim
reader.bootstrap('127.0.0.1', writer.getBoundPort());
const signedKey = hash('nodejs signed value');
assert.equal(await writer.putSigned(signedKey, payload, { id: 41n }), true);
const signed = await reader.get(signedKey);
const signed = await reader.getAll(signedKey);
assert.ok(signed.some(value => value.data.equals(payload) && value.isSigned));
assert.equal(signed[0].owner, writerIdentity.certificate.getLongId());
const encryptedKey = hash('nodejs encrypted value');
assert.equal(await writer.putEncrypted(encryptedKey, payload, readerIdentity.certificate.getPublicKey(), { id: 42n }), true);
const decrypted = await reader.get(encryptedKey);
const decrypted = await reader.getAll(encryptedKey);
assert.ok(decrypted.some(value => value.data.equals(payload)));
assert.throws(() => writer.putEncrypted(encryptedKey, payload, {}), TypeError);
assert.throws(() => writer.putEncrypted(encryptedKey, payload, writerIdentity.certificate.getPublicKey()), /RSA/i);
+21 -2
View File
@@ -1,5 +1,5 @@
import { Certificate, DhtRunner, Identity, PrivateKey, PublicKey, hash, version } from '../index.js';
import type { ConnectivityStatus, NodeStatus, Value } from '../index.js';
import type { ConnectivityStatus, ListenEvent, ListenerIterator, NodeStatus, Value, ValueIterator } from '../index.js';
async function exercise(): Promise<void> {
const privateKey: PrivateKey = await PrivateKey.generate({ algorithm: 'rsa', bits: 2048 });
@@ -22,7 +22,12 @@ async function exercise(): Promise<void> {
const accepted: boolean = await runner.put(key, Buffer.from(version), { id: 42n, permanent: true });
const signed: boolean = await runner.putSigned(key, plaintext);
const encrypted: boolean = await runner.putEncrypted(key, plaintext, publicKey);
const values: Value[] = await runner.get(key);
const values: Value[] = await runner.getAll(key);
const get: ValueIterator<Value> = runner.get(key);
for await (const value of get) {
const payload: Buffer = value.data;
void payload;
}
const listener = runner.listen(key, (value, expired) => {
const payload: Buffer = value.data;
const id: bigint = value.id;
@@ -30,7 +35,21 @@ async function exercise(): Promise<void> {
});
runner.ref().unref();
listener.cancel();
const iterator: ListenerIterator = runner.listen(key);
for await (const event of iterator) {
const typed: ListenEvent = event;
const value: Value = typed.value;
const expired: boolean = typed.expired;
void [value, expired];
break;
}
iterator.cancel();
await iterator.return();
await iterator[Symbol.asyncDispose]();
await runner[Symbol.asyncDispose]();
const proxy = new DhtRunner();
proxy.run({ proxyServer: 'https://proxy.example.com', proxyUserAgent: 'typed-client/1.0' });
await proxy.close();
void [accepted, values, valid, signed, encrypted, status];
}
+22
View File
@@ -0,0 +1,22 @@
This software is licensed under the MIT License.
Copyright Fedor Indutny, 2018.
Permission is hereby granted, free of charge, to any person obtaining a
copy of this software and associated documentation files (the
"Software"), to deal in the Software without restriction, including
without limitation the rights to use, copy, modify, merge, publish,
distribute, sublicense, and/or sell copies of the Software, and to permit
persons to whom the Software is furnished to do so, subject to the
following conditions:
The above copyright notice and this permission notice shall be included
in all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN
NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE
USE OR OTHER DEALINGS IN THE SOFTWARE.
+70
View File
@@ -0,0 +1,70 @@
# Building and Testing the Rust Crate
The `opendht` crate links to the native OpenDHT C library. Cargo does not build
or bundle that library. Install Rust and the native dependencies described in
the [OpenDHT build guide](../BUILD.md).
## Install the Native Library
From the root of an OpenDHT source checkout, enable the C bindings:
```sh
cmake -S . -B build-bindings -DOPENDHT_C=ON -DBUILD_TESTING=ON
cmake --build build-bindings
ctest --test-dir build-bindings -R test_c --output-on-failure
cmake --install build-bindings
```
Installation may require administrator privileges, depending on the install
prefix. Set `CMAKE_INSTALL_PREFIX` when configuring to choose another location.
The linker and runtime loader must be able to find the installed libraries.
Applications can then depend on `opendht` from crates.io as shown in the
[README](README.md).
## Use an Uninstalled Native Build
You can use the native build directly by setting its library search paths.
For example, from the repository root on macOS:
```sh
export LIBRARY_PATH="$PWD/build-bindings${LIBRARY_PATH:+:$LIBRARY_PATH}"
export DYLD_LIBRARY_PATH="$PWD/build-bindings${DYLD_LIBRARY_PATH:+:$DYLD_LIBRARY_PATH}"
export RUSTFLAGS="-L native=$PWD/build-bindings"
cargo test --manifest-path rust/Cargo.toml --all-targets
```
On Linux use `LD_LIBRARY_PATH` instead of `DYLD_LIBRARY_PATH`; on Windows add
the native DLL directory to `PATH`. For an application outside this checkout,
use absolute paths to the native build and run Cargo in the application directory.
When linking an AddressSanitizer native build, use the same compiler runtime.
For example, with Clang:
```sh
export RUSTFLAGS="-L native=$PWD/build-bindings -C linker=clang -C link-arg=-fsanitize=address"
```
## Build and Test the Crate
With the native library available, run from the `rust` directory:
```sh
cargo build --all-targets
cargo test --all-targets
cargo test --doc
cargo test --all-features --all-targets
cargo test --all-features --doc
```
## Run the Examples
From the `rust` directory:
```sh
cargo run --example dhtnode
cargo run --features async --example dhtnode_async
cargo run --features serde --example dhtnode_async
```
The callback example contacts a public bootstrap server. The async example uses
two loopback nodes; enabling `serde` also demonstrates typed MessagePack records.
+21 -2
View File
@@ -1,11 +1,30 @@
[package]
name = "opendht"
version = "0.2.0"
version = "0.6.0"
authors = ["Sébastien Blin <sebastien.blin@savoirfairelinux.com>", "Adrien Beraud <adrien.beraud@savoirfairelinux.com>"]
edition = "2024"
description = "Rust bindings for OpenDHT"
license = "MIT"
license-file = "../LICENSE"
repository = "https://github.com/savoirfairelinux/opendht"
[features]
async = ["dep:futures-channel", "dep:futures-core"]
serde = ["async", "dep:serde", "dep:rmp-serde"]
[[example]]
name = "dhtnode_async"
required-features = ["async"]
[dependencies]
libc="0.2"
os_socketaddr="0.2.3"
futures-channel = { version = "0.3", optional = true }
futures-core = { version = "0.3", optional = true }
serde = { version = "1", optional = true }
rmp-serde = { version = "1", optional = true }
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[dev-dependencies]
futures = "0.3"
serde = { version = "1", features = ["derive"] }
serde_bytes = "0.11"
+136 -81
View File
@@ -1,97 +1,152 @@
# Rust Bindings for OpenDHT
# OpenDHT: a Distributed Hash Table
## Build and Test
OpenDHT is a Kademlia-based distributed hash table for storing and retrieving
small values over a peer-to-peer network without a central server. Common uses
include peer discovery, service announcements, and decentralized applications.
Install Rust and the native OpenDHT C library. From this directory:
The `opendht` crate provides Rust bindings to [OpenDHT](https://opendht.net), with
callback-based and runtime-independent async APIs. Capabilities include signed
and encrypted values, filtered reads, metadata queries, and typed MessagePack
records through Serde.
```sh
cargo build --all-targets
cargo test --all-targets
cargo test --doc
cargo run --example dhtnode
## Get Started
Add the crate to your dependencies:
```toml
[dependencies]
opendht = "0.6"
```
To use an uninstalled native build, set its library search paths. For example,
from the repository root on macOS:
The native OpenDHT library with C bindings must be installed separately; it is
not bundled with the crate. See the
[build guide](https://github.com/savoirfairelinux/opendht/blob/master/rust/BUILD.md)
for native installation, library paths, and source builds.
```sh
cmake -S . -B build-bindings -DOPENDHT_C=ON -DBUILD_TESTING=ON
cmake --build build-bindings
ctest --test-dir build-bindings -R test_c --output-on-failure
LIBRARY_PATH="$PWD/build-bindings" \
DYLD_LIBRARY_PATH="$PWD/build-bindings" \
RUSTFLAGS="-L native=$PWD/build-bindings" \
cargo test --manifest-path rust/Cargo.toml --all-targets
```
`DhtRunner` manages a node, `InfoHash` identifies a DHT key, and `Value` contains
the payload and metadata stored under that key. `put` publishes values, `get`
retrieves them, and `listen` subscribes to value updates and expiration events.
On Linux use `LD_LIBRARY_PATH` instead of `DYLD_LIBRARY_PATH`; on Windows add
the native DLL directory to `PATH`. When linking an AddressSanitizer native
build, use the same compiler runtime, for example
`RUSTFLAGS="-L native=$PWD/build-bindings -C linker=clang -C link-arg=-fsanitize=address"`.
## Ownership and Errors
Startup, crypto import/generation, serialization, and fallible settings return
`std::io::Result`. Public-key `pack()` exports bytes; `unpack(bytes)` replaces a
key only after successful import. Crypto methods accept binary buffers, including
embedded NULs. Passwords and other C strings reject embedded NULs.
Callbacks are owned `Send + 'static` closures, not borrowed `&mut` closures.
Use `move` captures and channels or `Arc` for shared application state. Native
handles are thread-bound; send copied data through channels rather than moving
a runner or value handle between threads. Handles own stable native pointers
and can be moved out of or back into their Rust boxes on the same thread.
Callback panics are contained at the
FFI boundary; a panicking value callback stops its operation.
`Value::boxed()` creates an independent copy, including payload and metadata,
so a retained value does not alias mutable native storage or in-flight puts.
This example starts a node on an OS-assigned port, bootstraps into the public
network, and listens for a value it publishes.
```rust
use opendht::{DhtRunner, InfoHash, Value};
use std::sync::mpsc;
use std::{sync::mpsc, time::Duration};
let mut runner = DhtRunner::new();
runner.run(0)?;
let hash = InfoHash::get("example");
let (sent, received) = mpsc::channel();
let token = runner.listen(&hash, move |value, expired| {
if !expired { let _ = sent.send(value.as_bytes()); }
true
})?;
runner.put(&hash, Value::new("hello"), |_| {}, false);
let bytes = received.recv_timeout(std::time::Duration::from_secs(5))?;
runner.cancel_listen(&hash, token);
runner.join()?;
# Ok::<(), Box<dyn std::error::Error>>(())
fn main() -> Result<(), Box<dyn std::error::Error>> {
let mut runner = DhtRunner::new();
runner.run(0)?;
runner.bootstrap("bootstrap.jami.net", 4222)?;
let key = InfoHash::get("my-app/hello");
let (sender, receiver) = mpsc::channel();
let listener = runner.listen(&key, move |value, expired| {
if !expired {
let _ = sender.send(value);
}
true
})?;
runner.put(&key, Value::new("Hello from Rust!"), |_| {}, false);
let value = receiver.recv_timeout(Duration::from_secs(5))?;
println!("{}", String::from_utf8_lossy(value.as_slice()));
runner.cancel_listen(&key, listener);
runner.join()?;
Ok(())
}
```
`join()` waits for native callbacks, releases registrations, and permits restart.
Do not invoke it from a DHT callback. `shutdown(callback)` requests graceful
shutdown; wait for its callback before `join()` when persistence is required.
Dropping a listener token frees only the token; explicitly call `cancel_listen`
to remove the registration, or call `join()` to stop all operations. Pending
get/put/query completions are reported as failure when the runner stops.
A bootstrap server supplies initial peers. Applications can use public
bootstrap servers or known peers on a private network. Values expire, and their
availability depends on participating nodes; the DHT does not provide durable
storage guarantees.
For non-threaded operation set `config.threaded = false` and repeatedly call
`loop_once()`. Priority operations can take an iteration before normal work.
`run_bind` selects IPv4/IPv6 bind addresses; `bound_port` reports OS-assigned ports.
## Optional Features
## Extended APIs
| Feature | Capabilities |
|---------|--------------|
| Default | Callback-based `DhtRunner`, binary values, keys, certificates, and signed/encrypted operations. |
| `async` | `aio::AsyncDhtRunner` with publication futures, bounded read/update streams, and automatic listener cancellation on drop. |
| `serde` | Serde-backed MessagePack records with `put_typed`, `get_typed`, and `listen_typed`. Enables `async`. |
- Values expose explicit IDs, user types, value types, sequence numbers,
push priorities, and push types.
- Keys and certificates support persistence; identities support RSA and EC
generation. Private keys support binary signing and RSA decryption.
- `get_filtered`, `listen_filtered`, and `query` accept native SQL-like filters,
such as `WHERE value_type=17`, `WHERE seq=7`, and
`SELECT id,user_type WHERE value_type=17`.
Projected fields are optional owned values in `QueryResult`.
- `register_type` installs native type policy and expiration settings.
- Configuration accepts owned status and certificate-store callbacks.
Certificate lookup/registration and connectivity changes are exposed.
- Push token/topic/platform setters forward native settings.
`push_notification_received` returns a `PushNotification`; call `poll()` for
the native result. In non-threaded mode continue driving `loop_once()`.
Disabled builds return `IgnoredDisabled`; stopped runners return the native
ignored outcome. Push tokens remain valid after the runner stops.
Enable `async` in the dependency configuration:
```toml
[dependencies]
opendht = { version = "0.6", features = ["async"] }
futures = "0.3"
```
Create the adapter with `AsyncDhtRunner::new(runner)` after starting a
`DhtRunner`. The adapter is runtime-independent and thread-bound; use a local
executor or Tokio's `LocalSet`. Dropping a listener stream cancels its
subscription. Streams use bounded buffers and report `WouldBlock` on overflow.
This example waits for a bootstrap peer, publishes a value with `.await`, and
reads matching values as a stream:
```rust,no_run
use futures::{executor::block_on, TryStreamExt};
use opendht::{aio::AsyncDhtRunner, DhtRunner, InfoHash, Value};
use std::{io, net::ToSocketAddrs};
fn main() -> io::Result<()> {
let mut runner = DhtRunner::new();
runner.run(0)?;
let runner = AsyncDhtRunner::new(runner);
let peers: Vec<_> = ("bootstrap.jami.net", 4222).to_socket_addrs()?.collect();
block_on(async {
runner.bootstrap(&peers).await?;
let key = InfoHash::get("my-app/hello-async");
runner.put(&key, Value::new("Hello from async Rust!"), false).await?;
let mut values = runner.get(&key);
while let Some(value) = values.try_next().await? {
println!("{}", String::from_utf8_lossy(value.as_slice()));
}
runner.shutdown().await?;
runner.join()
})
}
```
### Typed Records
The `serde` feature adds typed async operations and also enables `async`.
`put_typed` serializes records implementing `Serialize`; `get_typed` and
`listen_typed` decode records implementing `DeserializeOwned`. Typed reads retain
the decoded record alongside its original payload and native metadata.
Structs are serialized as compact MessagePack arrays. Field order is part of
the wire format and must remain compatible across schema changes. `serde_bytes`
encodes binary fields using MessagePack's binary representation.
## Working With Values
GET and LISTEN callbacks and async streams deliver owned `Box<Value>` instances.
Received values implement `Send` but not `Sync`; they can be retained after the
runner stops, moved to another thread, modified, and republished. `as_slice()`
borrows the payload without copying; `as_bytes()` allocates a separate `Vec<u8>`.
Values expose metadata including IDs, types, sequence numbers, and owner keys.
Filters select values by metadata, and queries return selected metadata fields.
`DhtRunner::listen` returns a token that must be passed to `cancel_listen` to
unsubscribe. Dropping the token does not cancel the registration; dropping an
async listener stream does.
`join()` stops the runner and waits for its threads to exit. It must not be
called from a callback. For graceful shutdown and configured state persistence,
wait for `shutdown` to complete before calling `join()`.
## Learn More
- [API reference](https://docs.rs/opendht)
- [Callback example](https://github.com/savoirfairelinux/opendht/blob/master/rust/examples/dhtnode.rs)
- [Async and typed two-node example](https://github.com/savoirfairelinux/opendht/blob/master/rust/examples/dhtnode_async.rs)
- [Build and test guide](https://github.com/savoirfairelinux/opendht/blob/master/rust/BUILD.md)
- [OpenDHT project and issue tracker](https://github.com/savoirfairelinux/opendht)
+80
View File
@@ -0,0 +1,80 @@
use futures::{executor::block_on, StreamExt, TryStreamExt};
use opendht::{aio::AsyncDhtRunner, DhtRunner, DhtRunnerConfig, InfoHash, Value};
use std::error::Error;
use std::io;
use std::net::SocketAddr;
fn main() -> Result<(), Box<dyn Error>> {
let mut source = DhtRunner::new();
let mut destination = DhtRunner::new();
source.run_bind("127.0.0.1", "::1", 0, DhtRunnerConfig::new())?;
destination.run_bind("127.0.0.1", "::1", 0, DhtRunnerConfig::new())?;
let address = SocketAddr::from(([127, 0, 0, 1], destination.bound_port(libc::AF_INET as _)));
let source = AsyncDhtRunner::new(source);
let destination = AsyncDhtRunner::new(destination);
block_on(async {
source.bootstrap(&[address]).await?;
let hash = InfoHash::get("async example");
let mut updates = destination.listen(&hash)?;
source
.put(&hash, Value::with_id(b"hello from Rust async", 42), false)
.await?;
let event = updates
.next()
.await
.ok_or_else(|| io::Error::other("listener ended before receiving a value"))??;
println!(
"LISTEN {}: {} (expired: {})",
event.value.id(),
String::from_utf8_lossy(event.value.as_slice()),
event.expired,
);
drop(updates);
let values: Vec<_> = destination.get(&hash).try_collect().await?;
println!("GET: {} values", values.len());
#[cfg(feature = "serde")]
{
#[derive(Debug, serde::Serialize, serde::Deserialize)]
struct Service {
version: u32,
name: String,
port: u16,
#[serde(with = "serde_bytes")]
payload: Vec<u8>,
}
let hash = InfoHash::get("typed async example");
let mut updates = destination.listen_typed::<Service>(&hash)?;
let record = Service {
version: 1,
name: "chat".into(),
port: 4222,
payload: vec![0, 0x80, 0xff],
};
source.put_typed(&hash, &record, false).await?;
let event = updates.next().await.ok_or_else(|| {
io::Error::other("typed listener ended before receiving a value")
})??;
println!(
"TYPED LISTEN {}: {:?}",
event.value.raw.id(),
event.value.data
);
drop(updates);
let values: Vec<_> = destination
.get_typed::<Service>(&hash)
.try_collect()
.await?;
println!("TYPED GET: {} records", values.len());
}
source.shutdown().await?;
destination.shutdown().await?;
source.join()?;
destination.join()?;
Ok(())
})
}
+429
View File
@@ -0,0 +1,429 @@
//! Runtime-independent futures and bounded streams, enabled by the `async` feature.
//!
//! Runners and listeners stay on their creating thread. Use a local executor or
//! Tokio `LocalSet`; received values own their native storage and can be
//! transferred between threads without another payload copy.
use crate::{DhtRunner, InfoHash, OpToken, PkId, QueryResult, Value};
use futures_channel::{mpsc, oneshot};
use futures_core::Stream;
use std::cell::RefCell;
use std::future::Future;
use std::io;
use std::net::SocketAddr;
use std::pin::Pin;
use std::rc::Rc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::task::{Context, Poll};
/// A listener update; expiration events retain the expired value's metadata.
pub struct ValueEvent {
pub value: Box<Value>,
pub expired: bool,
}
/// A native completion mapped to an I/O result.
///
/// Operations start immediately, before this future is polled. Dropping it
/// abandons the result but does not undo a put, ping, bootstrap, or shutdown.
#[must_use = "operations run immediately; await their result or explicitly drop it"]
pub struct Operation {
receiver: oneshot::Receiver<io::Result<()>>,
}
impl Future for Operation {
type Output = io::Result<()>;
fn poll(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Self::Output> {
Pin::new(&mut self.receiver).poll(context).map(|outcome| {
outcome.unwrap_or_else(|_| Err(stopped("DHT operation ended without completion")))
})
}
}
fn operation() -> (oneshot::Sender<io::Result<()>>, Operation) {
let (sender, receiver) = oneshot::channel();
(sender, Operation { receiver })
}
fn outcome(success: bool) -> io::Result<()> {
if success {
Ok(())
} else {
Err(io::Error::other(
"DHT operation failed or the runner stopped",
))
}
}
fn stopped(message: &'static str) -> io::Error {
io::Error::new(io::ErrorKind::Interrupted, message)
}
#[derive(Clone)]
struct StreamCompletion(Arc<Mutex<Option<oneshot::Sender<io::Result<()>>>>>);
impl StreamCompletion {
fn finish(&self, result: io::Result<()>) {
let sender = self
.0
.lock()
.unwrap_or_else(|error| error.into_inner())
.take();
if let Some(sender) = sender {
let _ = sender.send(result);
}
}
}
struct StreamSender<T> {
values: mpsc::Sender<T>,
completion: StreamCompletion,
}
impl<T> StreamSender<T> {
fn send(&mut self, value: io::Result<T>) -> bool {
let value = match value {
Ok(value) => value,
Err(error) => {
self.completion.finish(Err(error));
return false;
}
};
match self.values.try_send(value) {
Ok(()) => true,
Err(error) => {
if error.is_full() {
self.completion.finish(Err(io::Error::new(
io::ErrorKind::WouldBlock,
"DHT stream capacity exceeded; restart the operation to resynchronize",
)));
}
false
}
}
}
}
/// A bounded get or query stream.
///
/// Buffered items precede a terminal error. Native failure, runner shutdown,
/// and buffer overflow are never reported as successful EOF.
/// Dropping the stream stops delivery; native work stops at its next callback
/// or when the runner is joined.
#[must_use = "streams must be polled to receive values and completion errors"]
pub struct OperationStream<T> {
values: mpsc::Receiver<T>,
completion: oneshot::Receiver<io::Result<()>>,
completed: bool,
outcome: Option<io::Result<()>>,
}
impl<T: Unpin> Stream for OperationStream<T> {
type Item = io::Result<T>;
fn poll_next(self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Option<Self::Item>> {
let stream = self.get_mut();
if !stream.completed
&& let Poll::Ready(result) = Pin::new(&mut stream.completion).poll(context)
{
stream.completed = true;
stream.outcome = Some(
result.unwrap_or_else(|_| Err(stopped("DHT stream ended without completion"))),
);
stream.values.close();
}
match Pin::new(&mut stream.values).poll_next(context) {
Poll::Ready(Some(value)) => Poll::Ready(Some(Ok(value))),
Poll::Ready(None) if stream.completed => {
Poll::Ready(stream.outcome.take().and_then(Result::err).map(Err))
}
_ => Poll::Pending,
}
}
}
fn stream_channel<T>(capacity: usize) -> (StreamSender<T>, OperationStream<T>) {
let (values, receiver) = mpsc::channel(capacity - 1);
let (completed, completion) = oneshot::channel();
(
StreamSender {
values,
completion: StreamCompletion(Arc::new(Mutex::new(Some(completed)))),
},
OperationStream {
values: receiver,
completion,
completed: false,
outcome: None,
},
)
}
struct RunnerState {
runner: RefCell<Box<DhtRunner>>,
stopping: Arc<AtomicBool>,
}
impl Drop for RunnerState {
fn drop(&mut self) {
self.stopping.store(true, Ordering::SeqCst);
}
}
/// A listener registration that cancels itself when dropped or terminated.
///
/// The runner is retained until this registration is dropped. Calling `join()`
/// stops all registrations and makes pending listeners report an error.
#[must_use = "dropping a listener cancels its native registration"]
pub struct Listener {
state: Rc<RunnerState>,
hash: InfoHash,
token: Option<Box<OpToken>>,
stream: OperationStream<ValueEvent>,
}
impl Listener {
fn cancel(&mut self) {
if let Some(token) = self.token.take() {
self.state
.runner
.borrow_mut()
.cancel_listen(&self.hash, token);
}
}
}
impl Stream for Listener {
type Item = io::Result<ValueEvent>;
fn poll_next(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Option<Self::Item>> {
let result = Pin::new(&mut self.stream).poll_next(context);
if matches!(result, Poll::Ready(None) | Poll::Ready(Some(Err(_)))) {
self.cancel();
}
result
}
}
impl Drop for Listener {
fn drop(&mut self) {
self.cancel();
}
}
/// A shared, thread-bound async adapter for an already configured runner.
///
/// Clones share a runner without making native handles `Send` or `Sync`.
/// Native calls are short; futures and streams never hold a mutable runner
/// borrow while waiting. The default stream capacity is 64 items.
///
/// ```compile_fail
/// use opendht::{aio::AsyncDhtRunner, DhtRunner};
/// let runner = AsyncDhtRunner::new(DhtRunner::new());
/// std::thread::spawn(move || runner.join().unwrap());
/// ```
#[derive(Clone)]
pub struct AsyncDhtRunner {
state: Rc<RunnerState>,
capacity: usize,
}
impl AsyncDhtRunner {
pub fn new(runner: Box<DhtRunner>) -> Self {
Self::with_capacity(runner, 64).expect("default stream capacity is valid")
}
/// Set the maximum number of queued values per get, query, or listener.
/// A slow consumer receives `WouldBlock` and the stream terminates.
pub fn with_capacity(runner: Box<DhtRunner>, capacity: usize) -> io::Result<Self> {
if capacity == 0 {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
"stream capacity must be positive",
));
}
Ok(Self {
state: Rc::new(RunnerState {
runner: RefCell::new(runner),
stopping: Arc::new(AtomicBool::new(false)),
}),
capacity,
})
}
pub fn put(&self, hash: &InfoHash, value: Box<Value>, permanent: bool) -> Operation {
let (sender, pending) = operation();
self.state.runner.borrow_mut().put(
hash,
value,
move |success| {
let _ = sender.send(outcome(success));
},
permanent,
);
pending
}
pub fn put_signed(&self, hash: &InfoHash, value: Box<Value>, permanent: bool) -> Operation {
let (sender, pending) = operation();
self.state.runner.borrow_mut().put_signed(
hash,
value,
move |success| {
let _ = sender.send(outcome(success));
},
permanent,
);
pending
}
pub fn put_encrypted(
&self,
hash: &InfoHash,
recipient: &PkId,
value: Box<Value>,
permanent: bool,
) -> Operation {
let (sender, pending) = operation();
self.state.runner.borrow_mut().put_encrypted(
hash,
recipient,
value,
move |success| {
let _ = sender.send(outcome(success));
},
permanent,
);
pending
}
pub fn ping(&self, address: SocketAddr) -> Operation {
let (sender, pending) = operation();
self.state
.runner
.borrow_mut()
.ping(address, move |success| {
let _ = sender.send(outcome(success));
});
pending
}
/// Bootstrap pre-resolved addresses concurrently; one successful peer suffices.
/// DNS resolution is left to the application's executor.
pub fn bootstrap(&self, addresses: &[SocketAddr]) -> Operation {
let (sender, pending) = operation();
if addresses.is_empty() {
let _ = sender.send(Err(io::Error::new(
io::ErrorKind::InvalidInput,
"at least one bootstrap address is required",
)));
return pending;
}
let completion = StreamCompletion(Arc::new(Mutex::new(Some(sender))));
let remaining = Arc::new(std::sync::atomic::AtomicUsize::new(addresses.len()));
for &address in addresses {
let completion = completion.clone();
let remaining = remaining.clone();
self.state
.runner
.borrow_mut()
.ping(address, move |success| {
if success {
completion.finish(Ok(()));
}
if remaining.fetch_sub(1, Ordering::SeqCst) == 1 {
completion.finish(outcome(false));
}
});
}
pending
}
pub fn get(&self, hash: &InfoHash) -> OperationStream<Box<Value>> {
let (mut sender, stream) = stream_channel(self.capacity);
let completion = sender.completion.clone();
self.state.runner.borrow_mut().get(
hash,
move |value| sender.send(Ok(value)),
move |success| completion.finish(outcome(success)),
);
stream
}
pub fn get_filtered(
&self,
hash: &InfoHash,
filter: &str,
) -> io::Result<OperationStream<Box<Value>>> {
let (mut sender, stream) = stream_channel(self.capacity);
let completion = sender.completion.clone();
self.state.runner.borrow_mut().get_filtered(
hash,
filter,
move |value| sender.send(Ok(value)),
move |success| completion.finish(outcome(success)),
)?;
Ok(stream)
}
pub fn query(&self, hash: &InfoHash, query: &str) -> io::Result<OperationStream<QueryResult>> {
let (mut sender, stream) = stream_channel(self.capacity);
let completion = sender.completion.clone();
self.state.runner.borrow_mut().query(
hash,
query,
move |row| sender.send(Ok(row)),
move |success| completion.finish(outcome(success)),
)?;
Ok(stream)
}
pub fn listen(&self, hash: &InfoHash) -> io::Result<Listener> {
self.listen_filtered(hash, "")
}
pub fn listen_filtered(&self, hash: &InfoHash, filter: &str) -> io::Result<Listener> {
let (mut sender, stream) = stream_channel(self.capacity);
let token = self.state.runner.borrow_mut().listen_filtered(
hash,
filter,
move |value, expired| sender.send(Ok(ValueEvent { value, expired })),
)?;
Ok(Listener {
state: self.state.clone(),
hash: InfoHash { d: hash.d },
token: Some(token),
stream,
})
}
/// Request graceful shutdown. Call `join()` after this future succeeds when
/// persistence is required. Joining first reports an interrupted shutdown.
pub fn shutdown(&self) -> Operation {
let (sender, pending) = operation();
let stopping = self.state.stopping.clone();
self.state.runner.borrow_mut().shutdown(move || {
let result = if stopping.load(Ordering::SeqCst) {
Err(stopped("runner joined before graceful shutdown completed"))
} else {
Ok(())
};
let _ = sender.send(result);
});
pending
}
/// Drive a non-threaded runner without waiting for any async operation.
pub fn loop_once(&self) -> io::Result<()> {
self.state.runner.borrow_mut().loop_once()
}
/// Stop the runner and wait for native callbacks. This is a blocking call,
/// not an async wait, and must not be invoked from a native DHT callback.
pub fn join(&self) -> io::Result<()> {
self.state.stopping.store(true, Ordering::SeqCst);
self.state.runner.borrow_mut().join()
}
}
+5 -3
View File
@@ -272,7 +272,7 @@ extern "C" fn get_handler_cb(value: *const ffi::Value, ptr: *mut c_void) -> bool
}
callback(false, || unsafe {
let handler = ptr as *mut GetHandler;
(*handler).get_cb(Value::from_raw(ffi::dht_value_copy(value)).expect("value copy failed"))
(*handler).get_cb(Value::copy_from_received(value).expect("value copy failed"))
})
}
@@ -308,7 +308,7 @@ extern "C" fn listen_handler(value: *const ffi::Value, expired: bool, ptr: *mut
callback(false, || unsafe {
let handler = ptr as *mut ListenHandler;
(*handler).cb(
Value::from_raw(ffi::dht_value_copy(value)).expect("value copy failed"),
Value::copy_from_received(value).expect("value copy failed"),
expired,
)
})
@@ -537,6 +537,7 @@ impl DhtRunner {
result(unsafe { ffi::dht_runner_connectivity_changed(self.as_ptr()) })
}
/// Deliver owned values that callbacks may mutate or move to another thread.
pub fn get(
&mut self,
h: &InfoHash,
@@ -676,7 +677,8 @@ impl DhtRunner {
unsafe { ffi::dht_runner_cancel_put(self.as_ptr(), h, vid) }
}
/// Registers an owned callback. It is released after cancellation or `join()`.
/// Registers an owned callback receiving ordinary owned `Box<Value>` objects.
/// It is released after cancellation or `join()`.
/// Borrowed callbacks cannot outlive their stack frame:
///
/// ```compile_fail
+6
View File
@@ -40,6 +40,12 @@ mod infohash;
mod pkid;
mod value;
#[cfg(feature = "async")]
pub mod aio;
#[cfg(feature = "serde")]
pub mod typed;
pub use blob::Blob;
pub use dhtrunner::{
DhtNodeConfig, DhtRunner, DhtRunnerConfig, DhtSecureConfig, NodeStatus, OpToken,
+201
View File
@@ -0,0 +1,201 @@
//! MessagePack-encoded typed operations, enabled by the `serde` feature.
//!
//! This feature enables `async` and adds methods to [`AsyncDhtRunner`]. Values
//! use compact MessagePack payloads, with no binding-specific envelope or
//! implicit schema version. Structs are encoded as arrays in field order.
//! Applications can include a version field in their records and use
//! `serde_bytes` for binary fields rather than arrays of integers.
//!
//! ```no_run
//! use futures::TryStreamExt;
//! use opendht::{aio::AsyncDhtRunner, InfoHash};
//! use serde::{Deserialize, Serialize};
//!
//! #[derive(Debug, Serialize, Deserialize)]
//! struct Service {
//! version: u32,
//! name: String,
//! port: u16,
//! #[serde(with = "serde_bytes")]
//! payload: Vec<u8>,
//! }
//!
//! async fn exchange(
//! runner: &AsyncDhtRunner,
//! key: &InfoHash,
//! ) -> Result<(), Box<dyn std::error::Error>> {
//! let mut updates = runner.listen_typed::<Service>(key)?;
//! let record = Service {
//! version: 1, name: "chat".into(), port: 4222, payload: vec![0, 0x80, 0xff],
//! };
//! runner.put_typed(key, &record, false).await?;
//! if let Some(event) = updates.try_next().await? {
//! println!("{}: {:?}", event.value.raw.id(), event.value.data);
//! }
//! drop(updates);
//! let records: Vec<_> = runner.get_typed::<Service>(key).try_collect().await?;
//! Ok(())
//! }
//! ```
use crate::aio::{AsyncDhtRunner, ValueEvent};
use crate::{InfoHash, Value};
use futures_core::Stream;
use serde::{de::DeserializeOwned, Serialize};
use std::error::Error;
use std::fmt;
use std::io;
use std::pin::Pin;
use std::task::{Context, Poll};
/// A decoded record together with its original payload and native metadata.
pub struct TypedValue<T> {
pub data: T,
pub raw: Box<Value>,
}
impl<T: fmt::Debug> fmt::Debug for TypedValue<T> {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("TypedValue")
.field("data", &self.data)
.field("id", &self.raw.id())
.field("seq", &self.raw.seq())
.finish_non_exhaustive()
}
}
/// A decoded listener update, including whether the record expired.
#[derive(Debug)]
pub struct TypedEvent<T> {
pub value: TypedValue<T>,
pub expired: bool,
}
/// Typed operations distinguish native failures from payload format errors.
#[derive(Debug)]
#[non_exhaustive]
pub enum TypedError {
Operation(io::Error),
Serialize(rmp_serde::encode::Error),
Deserialize {
id: u64,
source: rmp_serde::decode::Error,
},
}
impl fmt::Display for TypedError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Operation(error) => write!(formatter, "DHT operation failed: {error}"),
Self::Serialize(error) => write!(formatter, "could not serialize typed value: {error}"),
Self::Deserialize { id, source } => {
write!(formatter, "could not deserialize DHT value {id}: {source}")
}
}
}
}
impl Error for TypedError {
fn source(&self) -> Option<&(dyn Error + 'static)> {
match self {
Self::Operation(error) => Some(error),
Self::Serialize(error) => Some(error),
Self::Deserialize { source, .. } => Some(source),
}
}
}
impl<T: DeserializeOwned> TryFrom<Box<Value>> for TypedValue<T> {
type Error = TypedError;
fn try_from(raw: Box<Value>) -> Result<Self, Self::Error> {
let data =
rmp_serde::from_slice(raw.as_slice()).map_err(|source| TypedError::Deserialize {
id: raw.id(),
source,
})?;
Ok(Self { data, raw })
}
}
impl<T: DeserializeOwned> TryFrom<ValueEvent> for TypedEvent<T> {
type Error = TypedError;
fn try_from(event: ValueEvent) -> Result<Self, Self::Error> {
Ok(Self {
value: event.value.try_into()?,
expired: event.expired,
})
}
}
struct DecodeStream<Source: Stream, Output> {
source: Source,
decode: fn(Source::Item) -> Result<Output, TypedError>,
}
impl<Source: Stream + Unpin, Output> Stream for DecodeStream<Source, Output> {
type Item = Result<Output, TypedError>;
fn poll_next(self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Option<Self::Item>> {
let stream = self.get_mut();
Pin::new(&mut stream.source)
.poll_next(context)
.map(|item| item.map(stream.decode))
}
}
fn decode<Input, Output: TryFrom<Input, Error = TypedError>>(
input: io::Result<Input>,
) -> Result<Output, TypedError> {
input.map_err(TypedError::Operation)?.try_into()
}
impl AsyncDhtRunner {
/// Serialize a record as MessagePack and publish it when this future is polled.
///
/// Unlike untyped `put`, this operation is lazy. Serialization failure does
/// not start a native put. Dropping it after publication starts does not
/// undo the put. Use the untyped API to customize value IDs and metadata.
pub async fn put_typed<T: Serialize + ?Sized>(
&self,
hash: &InfoHash,
value: &T,
permanent: bool,
) -> Result<(), TypedError> {
let bytes = rmp_serde::to_vec(value).map_err(TypedError::Serialize)?;
self.put(hash, Value::from_bytes(&bytes), permanent)
.await
.map_err(TypedError::Operation)
}
/// Start a get and decode each MessagePack payload on the consumer's thread.
///
/// A malformed or incompatible record yields a per-value error; polling can
/// continue to subsequent records. Native errors remain terminal. The stream
/// does not borrow the runner or hash, and records need not be `Unpin`.
pub fn get_typed<T: DeserializeOwned>(
&self,
hash: &InfoHash,
) -> impl Stream<Item = Result<TypedValue<T>, TypedError>> + Unpin + use<T> {
DecodeStream {
source: self.get(hash),
decode: decode::<_, TypedValue<T>>,
}
}
/// Listen for MessagePack records, preserving metadata and expiration events.
///
/// Decode errors do not cancel the registration. Dropping this stream does
/// cancel it, including when a short-circuiting consumer stops on an error.
pub fn listen_typed<T: DeserializeOwned>(
&self,
hash: &InfoHash,
) -> io::Result<impl Stream<Item = Result<TypedEvent<T>, TypedError>> + Unpin + use<T>> {
Ok(DecodeStream {
source: self.listen(hash)?,
decode: decode::<_, TypedEvent<T>>,
})
}
}
+77 -19
View File
@@ -14,7 +14,40 @@ pub use crate::ffi::DataView;
native_handle!(Value, dht_value_unref);
/// Rust owns an unaliased native value: constructors allocate it, received
/// values are deep-copied once, and boxed() deep-copies explicitly. Puts consume
/// the handle before native code retains it. Shared owner keys are const and
/// their lazy fingerprints are initialized before received values cross threads.
/// Mutation requires exclusive access; native storage may be freed on any thread.
/// Value is Send, not Sync.
unsafe impl Send for Value {}
impl AsRef<[u8]> for Value {
fn as_ref(&self) -> &[u8] {
self.as_slice()
}
}
impl fmt::Debug for Value {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("Value")
.field("id", &self.id())
.field("seq", &self.seq())
.finish_non_exhaustive()
}
}
impl Value {
pub(crate) unsafe fn copy_from_received(value: *const ffi::Value) -> io::Result<Box<Self>> {
let value = unsafe { Self::from_raw(ffi::dht_value_copy(value))? };
if let Some(owner) = value.owner() {
owner.id();
owner.long_id();
}
Ok(value)
}
pub fn with_id(data: &[u8], id: u64) -> Box<Value> {
unsafe {
Self::from_raw(ffi::dht_value_with_id_new(data.as_ptr(), data.len(), id))
@@ -41,13 +74,16 @@ impl Value {
}
pub fn as_bytes(&self) -> Vec<u8> {
unsafe {
let dv = ffi::dht_value_get_data(self.as_ptr());
if dv.size == 0 {
return Vec::new();
}
let slice = slice::from_raw_parts(dv.data, dv.size);
slice.to_vec()
self.as_slice().to_vec()
}
/// Borrow the payload without allocating or copying it.
pub fn as_slice(&self) -> &[u8] {
let view = self.dataview();
if view.size == 0 {
&[]
} else {
unsafe { slice::from_raw_parts(view.data, view.size) }
}
}
@@ -63,12 +99,8 @@ impl Value {
unsafe { PublicKey::from_raw(ffi::dht_value_get_owner(self.as_ptr())).ok() }
}
pub fn user_type(&self) -> Result<String, str::Utf8Error> {
unsafe {
Ok(CStr::from_ptr(ffi::dht_value_get_user_type(self.as_ptr()))
.to_str()?
.to_owned())
}
pub fn user_type(&self) -> Result<&str, str::Utf8Error> {
unsafe { CStr::from_ptr(ffi::dht_value_get_user_type(self.as_ptr())).to_str() }
}
pub fn set_user_type(&mut self, value: &str) -> io::Result<()> {
@@ -103,12 +135,8 @@ impl Value {
ffi::dht_value_set_priority(self.as_ptr(), value);
}
}
pub fn push_type(&self) -> Result<String, str::Utf8Error> {
unsafe {
Ok(CStr::from_ptr(ffi::dht_value_get_push_type(self.as_ptr()))
.to_str()?
.to_owned())
}
pub fn push_type(&self) -> Result<&str, str::Utf8Error> {
unsafe { CStr::from_ptr(ffi::dht_value_get_push_type(self.as_ptr())).to_str() }
}
pub fn set_push_type(&mut self, value: &str) -> io::Result<()> {
let value = CString::new(value)?;
@@ -131,3 +159,33 @@ impl fmt::Display for Value {
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn value_moves_between_threads_without_copying_payload() {
let mut value = Value::from_bytes(b"native\0payload\xff");
value.set_user_type("borrowed metadata").unwrap();
let view = unsafe { ffi::dht_value_get_data(value.as_ptr()) };
let value = std::thread::spawn(move || {
value.set_seq(7);
value
})
.join()
.unwrap();
assert_eq!(value.as_slice().as_ptr(), view.data);
assert_eq!(value.as_slice(), b"native\0payload\xff");
assert_eq!(value.seq(), 7);
let user_type: &str = value.user_type().unwrap();
assert_eq!(user_type, "borrowed metadata");
}
#[test]
fn empty_value_payload_is_borrowed_safely() {
let value = Value::from_bytes(&[]);
assert!(value.as_slice().is_empty());
assert_eq!(value.as_slice(), &[]);
}
}
+564 -2
View File
@@ -55,7 +55,6 @@ fn native_handles_can_move_out_of_their_boxes() {
let token = *runner
.listen(&hash, move |value, expired| {
if !expired {
let value = *value;
sent.send(value.as_bytes()).unwrap();
}
true
@@ -246,7 +245,7 @@ fn shutdown_owns_its_callback_and_contains_panics() {
fn startup_errors_reach_rust() {
let socket4 = std::net::UdpSocket::bind("0.0.0.0:0").unwrap();
let port = socket4.local_addr().unwrap().port();
let _socket6 = std::net::UdpSocket::bind((std::net::Ipv6Addr::UNSPECIFIED, port)).unwrap();
let _socket6 = std::net::UdpSocket::bind((std::net::Ipv6Addr::LOCALHOST, port)).unwrap();
assert!(DhtRunner::new().run(port).is_err());
}
@@ -481,3 +480,566 @@ fn threaded_signed_and_encrypted_roundtrips() {
runner.join().unwrap();
}
}
#[test]
fn callbacks_move_mutable_values_without_copying() {
let mut runner = DhtRunner::new();
let mut config = DhtRunnerConfig::new();
config.threaded = false;
runner.run_config(0, config).unwrap();
let hash = InfoHash::get("single Value callback ownership");
let (listen_sent, listen_received) = mpsc::channel();
let _token = runner
.listen(&hash, move |mut value: Box<Value>, expired| {
if !expired {
value.set_seq(8);
let pointer = value.as_slice().as_ptr() as usize;
let _ = listen_sent.send((pointer, value));
}
true
})
.unwrap();
let mut value = Value::with_id(b"immutable\0payload\xff", 42);
value.set_seq(7);
value.set_user_type("application/immutable").unwrap();
value.set_push_type("call").unwrap();
runner.put(&hash, value, |_| {}, false);
runner.loop_once().unwrap();
let (get_sent, get_received) = mpsc::channel();
runner.get(
&hash,
move |value: Box<Value>| {
let pointer = value.as_slice().as_ptr() as usize;
let _ = get_sent.send((pointer, value));
true
},
|_| {},
);
runner.loop_once().unwrap();
runner.join().unwrap();
for (received, expected_seq) in [(listen_received, 8), (get_received, 7)] {
let (pointer, value) = received.recv_timeout(Duration::from_secs(5)).unwrap();
let value = std::thread::spawn(move || {
let mut value = value;
assert_eq!(value.as_slice().as_ptr() as usize, pointer);
assert_eq!(value.as_slice(), b"immutable\0payload\xff");
assert_eq!(value.id(), 42);
assert_eq!(value.seq(), expected_seq);
let user_type: &str = value.user_type().unwrap();
assert_eq!(user_type, "application/immutable");
let push_type: &str = value.push_type().unwrap();
assert_eq!(push_type, "call");
value.set_seq(9);
assert_eq!(value.as_slice().as_ptr() as usize, pointer);
value
})
.join()
.unwrap();
let mut editable = value.boxed();
editable.set_seq(10);
assert_eq!(value.seq(), 9);
assert_eq!(editable.seq(), 10);
}
}
#[cfg(feature = "async")]
mod asynchronous {
use super::*;
use futures::{executor::block_on, future, Future, StreamExt, TryStreamExt};
use opendht::aio::AsyncDhtRunner;
use std::pin::pin;
use std::time::Instant;
fn nonthreaded(capacity: usize) -> AsyncDhtRunner {
let mut runner = DhtRunner::new();
let mut config = DhtRunnerConfig::new();
config.threaded = false;
runner.run_bind("127.0.0.1", "::1", 0, config).unwrap();
runner.register_type(17, "async test", 1, false).unwrap();
AsyncDhtRunner::with_capacity(runner, capacity).unwrap()
}
fn drive<Output>(runner: &AsyncDhtRunner, operation: impl Future<Output = Output>) -> Output {
let mut operation = pin!(operation);
let deadline = Instant::now() + Duration::from_secs(15);
block_on(future::poll_fn(|context| {
assert!(Instant::now() < deadline, "async operation timed out");
runner.loop_once().unwrap();
context.waker().wake_by_ref();
operation.as_mut().poll(context)
}))
}
fn wait<Output>(operation: impl Future<Output = Output>) -> Output {
let (finished, stopped) = mpsc::channel();
let (expired, deadline) = futures::channel::oneshot::channel();
let watchdog = std::thread::spawn(move || {
if stopped.recv_timeout(Duration::from_secs(20)).is_err() {
let _ = expired.send(());
}
});
let outcome = block_on(future::select(Box::pin(operation), deadline));
let _ = finished.send(());
watchdog.join().unwrap();
match outcome {
future::Either::Left((output, _)) => output,
future::Either::Right(_) => panic!("async operation timed out"),
}
}
#[test]
fn async_listener_preserves_owned_metadata() {
let runner = nonthreaded(8);
let hash = InfoHash::get("async metadata");
let mut listener = runner.listen_filtered(&hash, "WHERE seq=7").unwrap();
let mut value = Value::with_id(b"async\0binary\xff", 42);
value.set_value_type(17);
value.set_seq(7);
value.set_priority(2);
value.set_user_type("application/async").unwrap();
value.set_push_type("call").unwrap();
let _pending = runner.put(&hash, value, false);
let event = drive(&runner, listener.next()).unwrap().unwrap();
assert!(!event.expired);
assert_eq!(event.value.as_slice(), b"async\0binary\xff");
assert_eq!(event.value.id(), 42);
assert_eq!(event.value.value_type(), 17);
assert_eq!(event.value.seq(), 7);
assert_eq!(event.value.priority(), 2);
assert_eq!(event.value.user_type().unwrap(), "application/async");
assert_eq!(event.value.push_type().unwrap(), "call");
assert!(event.value.owner().is_none());
assert!(event.value.recipient().is_zero());
let event = std::thread::spawn(move || event).join().unwrap();
assert_eq!(event.value.as_slice(), b"async\0binary\xff");
drop(listener);
runner.loop_once().unwrap();
runner.join().unwrap();
}
#[test]
fn async_listener_delivers_expiration_events() {
let runner = nonthreaded(8);
let hash = InfoHash::get("async expiration");
let mut listener = runner.listen(&hash).unwrap();
let mut value = Value::with_id(b"expires", 42);
value.set_value_type(17);
let _pending = runner.put(&hash, value, false);
assert!(!drive(&runner, listener.next()).unwrap().unwrap().expired);
let event = drive(&runner, listener.next()).unwrap().unwrap();
assert!(event.expired);
assert_eq!(event.value.id(), 42);
assert_eq!(event.value.as_slice(), b"expires");
drop(listener);
runner.join().unwrap();
}
#[test]
fn async_overflow_is_reported_after_buffered_values() {
let runner = nonthreaded(1);
let hash = InfoHash::get("async overflow");
let mut listener = runner.listen(&hash).unwrap();
for id in 1..=3 {
let _pending = runner.put(&hash, Value::with_id(b"queued", id), false);
runner.loop_once().unwrap();
}
assert_eq!(
drive(&runner, listener.next()).unwrap().unwrap().value.id(),
1
);
let error = drive(&runner, listener.next()).unwrap().err().unwrap();
assert_eq!(error.kind(), std::io::ErrorKind::WouldBlock);
assert!(drive(&runner, listener.next()).is_none());
drop(listener);
runner.join().unwrap();
}
#[test]
fn async_listener_retains_runner_until_drop() {
let released = Arc::new(AtomicUsize::new(0));
let counter = DropCounter(released.clone());
let mut native = DhtRunner::new();
let mut config = DhtRunnerConfig::new();
config.threaded = false;
config.set_status_callback(move |_, _| {
let _ = &counter;
});
native.run_config(0, config).unwrap();
let runner = AsyncDhtRunner::new(native);
let listener = runner.listen(&InfoHash::get("retained runner")).unwrap();
drop(runner);
assert_eq!(released.load(Ordering::SeqCst), 0);
drop(listener);
assert_eq!(released.load(Ordering::SeqCst), 1);
}
#[test]
fn async_join_completes_pending_operations_and_streams() {
let runner = nonthreaded(8);
let hash = InfoHash::get("async shutdown failures");
let put = runner.put(&hash, Value::new("pending"), false);
let mut values = runner.get(&hash);
let mut query = runner.query(&hash, "SELECT id").unwrap();
let mut listener = runner.listen(&hash).unwrap();
let ping = runner.ping("127.0.0.1:9".parse().unwrap());
let bootstrap = runner.bootstrap(&["127.0.0.1:9".parse().unwrap()]);
let shutdown = runner.shutdown();
runner.join().unwrap();
assert!(wait(put).is_err());
assert!(wait(ping).is_err());
assert!(wait(bootstrap).is_err());
assert!(wait(shutdown).is_err());
assert!(wait(values.next()).unwrap().is_err());
assert!(wait(values.next()).is_none());
assert!(wait(query.next()).unwrap().is_err());
assert!(wait(query.next()).is_none());
assert!(wait(listener.next()).unwrap().is_err());
assert!(wait(listener.next()).is_none());
drop(listener);
}
#[test]
fn async_stream_reports_runner_destruction() {
let runner = nonthreaded(8);
let mut values = runner.get(&InfoHash::get("dropped async runner"));
drop(runner);
assert!(wait(values.next()).unwrap().is_err());
assert!(wait(values.next()).is_none());
}
#[test]
fn async_invalid_inputs_and_dropped_operations() {
assert!(AsyncDhtRunner::with_capacity(DhtRunner::new(), 0).is_err());
let runner = nonthreaded(8);
let hash = InfoHash::get("async cancellation");
let error = wait(runner.bootstrap(&[])).unwrap_err();
assert_eq!(error.kind(), std::io::ErrorKind::InvalidInput);
assert!(runner.get_filtered(&hash, "WHERE user_type=\0").is_err());
assert!(runner.listen_filtered(&hash, "WHERE user_type=\0").is_err());
assert!(runner.query(&hash, "SELECT\0 id").is_err());
let mut invalid = runner.get_filtered(&hash, "WHERE invalid=1").unwrap();
assert!(wait(invalid.next()).unwrap().is_err());
assert!(wait(invalid.next()).is_none());
drop(runner.get(&hash));
drop(runner.query(&hash, "SELECT id").unwrap());
drop(runner.put(&hash, Value::new("abandoned"), false));
drop(runner.ping("127.0.0.1:9".parse().unwrap()));
let listener = runner.listen(&hash).unwrap();
drop(listener);
runner.loop_once().unwrap();
runner.join().unwrap();
}
#[test]
fn async_threaded_network_roundtrips() {
let mut nodes = Vec::new();
for name in ["async source", "async destination"] {
let mut runner = DhtRunner::new();
let mut config = DhtRunnerConfig::new();
config.dht_config.id = Some(DhtIdentity::new(name).unwrap());
runner.run_bind("127.0.0.1", "::1", 0, config).unwrap();
nodes.push(runner);
}
let destination_native = nodes.pop().unwrap();
let address = (
[127, 0, 0, 1],
destination_native.bound_port(libc::AF_INET as _),
)
.into();
let recipient = destination_native.public_key().unwrap().long_id();
let signer = nodes.last().unwrap().public_key().unwrap().id();
let source = AsyncDhtRunner::new(nodes.pop().unwrap());
let destination = AsyncDhtRunner::new(destination_native);
wait(source.bootstrap(&["127.0.0.1:9".parse().unwrap(), address])).unwrap();
for mode in ["plain", "signed", "encrypted"] {
let hash = InfoHash::get(mode);
let mut listener = destination.listen(&hash).unwrap();
let mut value = Value::with_id(b"async network\0value", 42);
value.set_seq(7);
let put = match mode {
"signed" => source.put_signed(&hash, value, false),
"encrypted" => source.put_encrypted(&hash, &recipient, value, false),
_ => source.put(&hash, value, false),
};
wait(put).unwrap();
let event = wait(listener.next()).unwrap().unwrap();
assert!(!event.expired);
assert_eq!(event.value.as_slice(), b"async network\0value");
if mode != "plain" {
assert!(event.value.owner().is_some());
assert!(event.value.owner().unwrap().id() == signer);
let event = std::thread::spawn(move || event).join().unwrap();
assert!(event.value.owner().unwrap().id() == signer);
}
drop(listener);
if mode == "signed" {
let values: Vec<_> = wait(
destination
.get_filtered(&hash, "WHERE seq=7")
.unwrap()
.try_collect(),
)
.unwrap();
assert!(values
.iter()
.any(|value| value.id() == 42 && value.seq() == 7));
}
let values: Vec<_> = wait(destination.get(&hash).try_collect()).unwrap();
assert!(values
.iter()
.any(|value| value.as_slice() == b"async network\0value"));
let rows: Vec<_> =
wait(destination.query(&hash, "SELECT id").unwrap().try_collect()).unwrap();
assert!(rows.iter().any(|row| row.id == Some(42)));
}
wait(source.shutdown()).unwrap();
wait(destination.shutdown()).unwrap();
source.join().unwrap();
destination.join().unwrap();
}
#[cfg(feature = "serde")]
mod typed {
use super::*;
use opendht::typed::TypedError;
use serde::{Deserialize, Serialize};
use std::error::Error;
use std::marker::PhantomPinned;
#[derive(Debug, PartialEq, Serialize, Deserialize)]
struct Record {
version: u32,
name: String,
ports: Vec<u16>,
#[serde(with = "serde_bytes")]
payload: Vec<u8>,
}
fn record() -> Record {
Record {
version: 1,
name: "node-\u{03bb}\0".into(),
ports: vec![4222, 8443],
payload: vec![0, 0x80, 0xff, 1],
}
}
#[test]
fn typed_listener_decodes_compact_msgpack_binary() {
#[derive(Debug, PartialEq, Deserialize)]
struct Packet {
#[serde(with = "serde_bytes")]
payload: Vec<u8>,
}
let runner = nonthreaded(8);
let hash = InfoHash::get("typed MessagePack binary fixture");
let mut listener = runner.listen_typed::<Packet>(&hash).unwrap();
let fixture = [0x91, 0xc4, 0x04, 0x00, 0x80, 0xff, 0x01];
let _pending = runner.put(&hash, Value::with_id(&fixture, 42), false);
let event = drive(&runner, listener.next()).unwrap().unwrap();
assert_eq!(event.value.raw.as_slice(), fixture);
assert_eq!(event.value.data.payload, [0, 0x80, 0xff, 1]);
drop(listener);
runner.join().unwrap();
}
#[test]
fn typed_listener_preserves_metadata_and_expiration() {
let runner = nonthreaded(8);
let hash = InfoHash::get("typed metadata");
let mut listener = runner.listen_typed::<Record>(&hash).unwrap();
let mut value = Value::with_id(&rmp_serde::to_vec(&record()).unwrap(), 42);
value.set_value_type(17);
value.set_seq(7);
value.set_user_type("application/typed-test").unwrap();
let _pending = runner.put(&hash, value, false);
let event = drive(&runner, listener.next()).unwrap().unwrap();
assert!(!event.expired);
assert_eq!(event.value.data, record());
assert_eq!(event.value.raw.id(), 42);
assert_eq!(event.value.raw.value_type(), 17);
assert_eq!(event.value.raw.seq(), 7);
assert_eq!(
event.value.raw.user_type().unwrap(),
"application/typed-test"
);
assert_eq!(
rmp_serde::from_slice::<Record>(event.value.raw.as_slice()).unwrap(),
record()
);
let event = drive(&runner, listener.next()).unwrap().unwrap();
assert!(event.expired);
assert_eq!(event.value.data, record());
assert_eq!(event.value.raw.id(), 42);
drop(listener);
runner.join().unwrap();
}
#[test]
fn typed_decode_errors_do_not_end_listener() {
let runner = nonthreaded(8);
let hash = InfoHash::get("typed mixed payloads");
let mut listener = runner.listen_typed::<Record>(&hash).unwrap();
for (id, payload) in [(1, [0xc1].as_slice()), (2, [0x11].as_slice())] {
let _pending = runner.put(&hash, Value::with_id(payload, id), false);
let error = drive(&runner, listener.next()).unwrap().err().unwrap();
assert!(error.source().is_some());
assert!(
matches!(error, TypedError::Deserialize { id: value_id, .. } if value_id == id)
);
}
let _pending = runner.put(
&hash,
Value::with_id(&rmp_serde::to_vec(&record()).unwrap(), 3),
false,
);
let event = drive(&runner, listener.next()).unwrap().unwrap();
assert_eq!(event.value.raw.id(), 3);
assert_eq!(event.value.data, record());
runner.join().unwrap();
let error = wait(listener.next()).unwrap().err().unwrap();
assert!(matches!(error, TypedError::Operation(_)));
assert!(wait(listener.next()).is_none());
}
#[test]
fn typed_put_distinguishes_serialization_and_native_errors() {
struct Unserializable;
impl Serialize for Unserializable {
fn serialize<Serializer>(
&self,
_: Serializer,
) -> Result<Serializer::Ok, Serializer::Error>
where
Serializer: serde::Serializer,
{
Err(serde::ser::Error::custom(
"intentional serialization failure",
))
}
}
let runner = AsyncDhtRunner::new(DhtRunner::new());
let hash = InfoHash::get("typed put errors");
let error = wait(runner.put_typed(&hash, &Unserializable, false)).unwrap_err();
assert!(error.source().is_some());
assert!(matches!(error, TypedError::Serialize(_)));
let error = wait(runner.put_typed(&hash, &record(), false)).unwrap_err();
assert!(error.source().is_some());
assert!(matches!(error, TypedError::Operation(_)));
runner.join().unwrap();
}
#[test]
fn typed_streams_support_non_unpin_records_and_detached_lifetimes() {
#[derive(Serialize, Deserialize)]
struct PinnedRecord {
message: String,
#[serde(skip)]
_pin: PhantomPinned,
}
let runner = nonthreaded(8);
let hash = InfoHash::get("typed pinned records");
let mut listener = runner.listen_typed::<PinnedRecord>(&hash).unwrap();
let value = PinnedRecord {
message: "owned".into(),
_pin: PhantomPinned,
};
let _pending = runner.put(
&hash,
Value::from_bytes(&rmp_serde::to_vec(&value).unwrap()),
false,
);
let event = drive(&runner, listener.next()).unwrap().unwrap();
assert_eq!(event.value.data.message, "owned");
drop(listener);
let mut values = runner.get_typed::<PinnedRecord>(&InfoHash::get("temporary hash"));
drop(runner);
assert!(matches!(
wait(values.next()).unwrap(),
Err(TypedError::Operation(_))
));
assert!(wait(values.next()).is_none());
}
#[test]
fn typed_listener_inherits_bounded_stream_errors() {
let runner = nonthreaded(1);
let hash = InfoHash::get("typed overflow");
let mut listener = runner.listen_typed::<Record>(&hash).unwrap();
for id in 1..=3 {
let _pending = runner.put(
&hash,
Value::with_id(&rmp_serde::to_vec(&record()).unwrap(), id),
false,
);
runner.loop_once().unwrap();
}
assert_eq!(
drive(&runner, listener.next())
.unwrap()
.unwrap()
.value
.raw
.id(),
1
);
let error = drive(&runner, listener.next()).unwrap().err().unwrap();
assert!(
matches!(error, TypedError::Operation(error) if error.kind() == std::io::ErrorKind::WouldBlock)
);
assert!(drive(&runner, listener.next()).is_none());
drop(listener);
runner.join().unwrap();
}
#[test]
fn typed_threaded_put_get_listen_roundtrip() {
let mut source = DhtRunner::new();
let mut destination = DhtRunner::new();
source
.run_bind("127.0.0.1", "::1", 0, DhtRunnerConfig::new())
.unwrap();
destination
.run_bind("127.0.0.1", "::1", 0, DhtRunnerConfig::new())
.unwrap();
let address = ([127, 0, 0, 1], destination.bound_port(libc::AF_INET as _)).into();
let source = AsyncDhtRunner::new(source);
let destination = AsyncDhtRunner::new(destination);
wait(source.bootstrap(&[address])).unwrap();
let hash = InfoHash::get("typed network record");
let mut listener = destination.listen_typed::<Record>(&hash).unwrap();
wait(source.put_typed(&hash, &record(), false)).unwrap();
let event = wait(listener.next()).unwrap().unwrap();
assert_eq!(event.value.data, record());
assert_eq!(event.value.raw.as_slice().first(), Some(&0x94));
assert!(event
.value
.raw
.as_slice()
.ends_with(&[0xc4, 0x04, 0x00, 0x80, 0xff, 0x01]));
let id = event.value.raw.id();
drop(listener);
let values: Vec<_> =
wait(destination.get_typed::<Record>(&hash).try_collect()).unwrap();
assert!(values
.iter()
.any(|value| value.data == record() && value.raw.id() == id));
let values: Vec<_> = wait(destination.get(&hash).try_collect()).unwrap();
assert!(values.iter().any(|value| {
rmp_serde::from_slice::<Record>(value.as_slice()).unwrap() == record()
}));
wait(source.shutdown()).unwrap();
wait(destination.shutdown()).unwrap();
source.join().unwrap();
destination.join().unwrap();
}
}
}
+22 -4
View File
@@ -97,7 +97,7 @@ std::string
DhtProxyServer::ServerStats::toString() const
{
auto ret = fmt::format("Listens: {}, Puts: {}, PushListeners: {}\n"
"Push requests in the last {}: [Android: {}], [iOS: {}], [Unified: {}]\n"
"Push requests in the last {}: [Android: {}], [iOS: {}], [Unified: {}], [Huawei: {}]\n"
"Requests: {} per second.",
listenCount,
putCount,
@@ -106,6 +106,7 @@ DhtProxyServer::ServerStats::toString() const
androidPush.toString(),
iosPush.toString(),
unifiedPush.toString(),
huaweiPush.toString(),
requestRate);
if (nodeInfo) {
auto& ipv4 = nodeInfo->ipv4;
@@ -131,6 +132,7 @@ DhtProxyServer::ServerStats::toJson() const
result["androidPush"] = androidPush.toJson();
result["iosPush"] = iosPush.toJson();
result["unifiedPush"] = unifiedPush.toJson();
result["huaweiPush"] = huaweiPush.toJson();
result["requestRate"] = requestRate;
if (nodeInfo)
result["nodeInfo"] = nodeInfo->toJson();
@@ -639,6 +641,7 @@ DhtProxyServer::updateStats(std::shared_ptr<NodeInfo> info) const
stats.androidPush = androidPush_;
stats.iosPush = iosPush_;
stats.unifiedPush = unifiedPush_;
stats.huaweiPush = huaweiPush_;
}
#endif
stats.serverStartTime = serverStartTime_;
@@ -864,6 +867,8 @@ DhtProxyServer::getTypeFromString(const std::string& type)
return PushType::iOS;
else if (type == "unifiedpush")
return PushType::UnifiedPush;
else if (type == "huawei")
return PushType::Huawei;
return PushType::None;
}
@@ -1275,15 +1280,25 @@ DhtProxyServer::sendPushNotification(
Json::Value tokens(Json::arrayValue);
tokens[0] = token;
notification["tokens"] = std::move(tokens);
notification["platform"] = type == PushType::Android ? 2 : 1;
notification["data"] = std::move(json);
// gorush platform codes: 1 = iOS (APNs), 2 = Android (FCM), 3 = Huawei (HMS)
notification["platform"] = type == PushType::Android ? 2
: type == PushType::Huawei ? 3
: 1;
auto priority = highPriority ? "high" : "normal";
if (type == PushType::Android) {
if (type == PushType::Huawei) {
// HMS carries the payload as a serialized JSON string in its own field:
// the generic "data" object is ignored for this platform by gorush.
notification["huawei_data"] = Json::writeString(jsonBuilder_, json);
notification["huawei_ttl"] = "86400s"; // time to live = 24 hours
notification["priority"] = priority;
} else if (type == PushType::Android) {
notification["data"] = std::move(json);
Json::Value androidConfig(Json::objectValue);
androidConfig["priority"] = priority;
androidConfig["ttl"] = "86400s"; // time to live = 24 hours
notification["android"] = std::move(androidConfig);
} else {
notification["data"] = std::move(json);
notification["priority"] = priority;
const auto expiration = std::chrono::system_clock::now() + std::chrono::hours(24);
uint32_t exp = std::chrono::duration_cast<std::chrono::seconds>(expiration.time_since_epoch()).count();
@@ -1336,6 +1351,9 @@ DhtProxyServer::sendPushNotification(
case PushType::UnifiedPush:
unifiedPush_.increment(highPriority);
break;
case PushType::Huawei:
huaweiPush_.increment(highPriority);
break;
default:
break;
}
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "opendht",
"version-string": "4.4.99",
"version-string": "4.5.0-rc.1",
"port-version": 1,
"description": "OpenDHT: A lightweight C++17 Distributed Hash Table library.",
"homepage": "https://opendht.net",