tools/dht
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

jni

Browse Source
This commit is contained in:
桥下红药
2026-03-13 02:15:50 +08:00
parent 2ab9103e5b
commit 6e7c6f02e4
23 changed files with 1256 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/release.yml
+84 -17
View File
@@ -12,6 +12,41 @@ env:
CARGO_TERM_COLOR: always
jobs:
# ──────────────────────────────────────────────────────────────────
# Java fat JAR（平台无关，只需构建一次）
# 运行方式：java -Djava.library.path=<so/dll 目录> -jar dht-crawler-jni-example-<ver>.jar
# ──────────────────────────────────────────────────────────────────
build-java-jar:
name: Build Java Example JAR
runs-on: ubuntu-latest
permissions:
contents: write
steps:
- name: Checkout code
uses: actions/checkout@v3
- name: Set up JDK 11
uses: actions/setup-java@v3
with:
distribution: temurin
java-version: '11'
- name: Build fat JAR
working-directory: jni/java
run: |
chmod +x gradlew 2>/dev/null || true
gradle shadowJar -PprojectVersion=${{ github.event.inputs.version }} --no-daemon
shell: bash
- name: Upload JAR to Release
uses: softprops/action-gh-release@v1
with:
tag_name: ${{ github.event.inputs.version }}
files: jni/java/build/libs/dht-crawler-jni-example-${{ github.event.inputs.version }}.jar
# ──────────────────────────────────────────────────────────────────
# 各平台 Rust 构建（example 可执行 + JNI so/dll/dylib）
# ──────────────────────────────────────────────────────────────────
build-and-release:
name: Build and Release for ${{ matrix.target }}
runs-on: ${{ matrix.os }}
@@ -26,36 +61,42 @@ jobs:
os: ubuntu-latest
use_cross: false
platform: linux
jni_lib: libdht_crawler.so
# Linux x86_64 (MUSL - 静态链接，无依赖)
- target: x86_64-unknown-linux-musl
os: ubuntu-latest
use_cross: true
platform: linux
jni_lib: libdht_crawler.so
# Linux ARM64
- target: aarch64-unknown-linux-gnu
os: ubuntu-latest
use_cross: true
platform: linux
jni_lib: libdht_crawler.so
# Windows x86_64
- target: x86_64-pc-windows-msvc
os: windows-latest
use_cross: false
platform: windows
jni_lib: dht_crawler.dll
# macOS x86_64
- target: x86_64-apple-darwin
os: macos-latest
use_cross: false
platform: macos
jni_lib: libdht_crawler.dylib
# macOS ARM64 (Apple Silicon)
- target: aarch64-apple-darwin
os: macos-latest
use_cross: false
platform: macos
jni_lib: libdht_crawler.dylib
steps:
- name: Checkout code
@@ -82,12 +123,7 @@ jobs:
path: ~/.cargo/git
key: ${{ runner.os }}-cargo-index-${{ hashFiles('**/Cargo.lock') }}
- name: Cache target directory
uses: actions/cache@v3
with:
path: target
key: ${{ runner.os }}-${{ matrix.target }}-cargo-target-${{ hashFiles('**/Cargo.lock') }}
# release 构建不缓存 target 目录，避免占用 GitHub 缓存配额
- name: Run tests
run: |
if [ "${{ matrix.use_cross }}" = "true" ]; then
@@ -106,24 +142,25 @@ jobs:
fi
shell: bash
- name: Build
- name: Build (example + JNI cdylib)
run: |
if [ "${{ matrix.use_cross }}" = "true" ]; then
cross build --release --target ${{ matrix.target }} --examples --features mimalloc,metrics
cross build --release --target ${{ matrix.target }} --examples --features mimalloc,metrics,jni
else
cargo build --release --target ${{ matrix.target }} --examples --features mimalloc,metrics
cargo build --release --target ${{ matrix.target }} --examples --features mimalloc,metrics,jni
fi
shell: bash
- name: Get version
id: get_version
run: |
# 使用用户输入的版本号
VERSION="${{ github.event.inputs.version }}"
echo "version=$VERSION" >> $GITHUB_OUTPUT
shell: bash
- name: Prepare artifacts (Unix)
# ────────── example 产物打包 ──────────
- name: Prepare example artifacts (Unix)
if: runner.os != 'Windows'
run: |
cd target/${{ matrix.target }}/release/examples
@@ -131,7 +168,7 @@ jobs:
mv dht_crawler_example-${{ steps.get_version.outputs.version }}-${{ matrix.target }}.tar.gz ${{ github.workspace }}/
shell: bash
- name: Prepare artifacts (Windows)
- name: Prepare example artifacts (Windows)
if: runner.os == 'Windows'
run: |
cd target/${{ matrix.target }}/release/examples
@@ -139,9 +176,39 @@ jobs:
Move-Item dht_crawler_example-${{ steps.get_version.outputs.version }}-${{ matrix.target }}.zip ${{ github.workspace }}/
shell: pwsh
# ────────── JNI 动态库打包（仅含单个 so/dll/dylib，体积最小）──────────
- name: Prepare JNI artifacts (Unix)
if: runner.os != 'Windows'
run: |
VERSION=${{ steps.get_version.outputs.version }}
TARGET=${{ matrix.target }}
LIB=${{ matrix.jni_lib }}
WORKDIR=$(mktemp -d)
cp target/${TARGET}/release/${LIB} ${WORKDIR}/
cd ${WORKDIR}
zip dht_crawler_jni-${VERSION}-${TARGET}.zip ${LIB}
mv dht_crawler_jni-${VERSION}-${TARGET}.zip ${{ github.workspace }}/
shell: bash
- name: Prepare JNI artifacts (Windows)
if: runner.os == 'Windows'
run: |
$VERSION = "${{ steps.get_version.outputs.version }}"
$TARGET = "${{ matrix.target }}"
$LIB = "${{ matrix.jni_lib }}"
$SRC = "target\$TARGET\release\$LIB"
$OUT = "dht_crawler_jni-$VERSION-$TARGET.zip"
7z a $OUT $SRC
Move-Item $OUT ${{ github.workspace }}\
shell: pwsh
# ────────── 上传到 Release ──────────
- name: Upload Release Assets
uses: softprops/action-gh-release@v1
with:
tag_name: ${{ github.event.inputs.version }}
files: |
dht_crawler_example-${{ steps.get_version.outputs.version }}-${{ matrix.target }}.*
dht_crawler_jni-${{ steps.get_version.outputs.version }}-${{ matrix.target }}.zip
Cargo.toml
+4 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "dht-crawler"
version = "0.1.1"
version = "0.1.2"
edition = "2024"
authors = ["桥下红药 <1121744186@qq.com>"]
description = "高性能的 Rust DHT (Distributed Hash Table) 爬虫库 | A high-performance Rust DHT crawler library for fetching torrent information from the BitTorrent DHT network"
@@ -14,6 +14,7 @@ readme = "README.md"
[lib]
name = "dht_crawler"
path = "src/lib.rs"
crate-type = ["rlib", "cdylib"]
[dependencies]
tokio = { version = "1.35", features = ["rt", "rt-multi-thread", "net", "sync", "time", "macros"] }
@@ -32,6 +33,7 @@ ahash = "0.8"
serde_bytes = "0.11.19"
metrics = { version = "0.24", optional = true }
async-channel = "2.5.0"
jni = { version = "0.21", optional = true }
[dev-dependencies]
tracing = "0.1"
@@ -46,6 +48,7 @@ metrics-exporter-prometheus = { version = "0.18", default-features = false, feat
default = []
metrics = ["dep:metrics"]
mimalloc = []
jni = ["dep:jni"]
[[example]]
name = "dht_crawler_example"
README.md
+80 -34
View File
@@ -2,7 +2,7 @@
[![Crates.io](https://img.shields.io/crates/v/dht-crawler.svg)](https://crates.io/crates/dht-crawler)
[![Documentation](https://docs.rs/dht-crawler/badge.svg)](https://docs.rs/dht-crawler)
[![License](https://img.shields.io/crates/l/dht-crawler.svg)](https://github.com/yourusername/dht-crawler/blob/master/LICENSE)
[![License](https://img.shields.io/crates/l/dht-crawler.svg)](https://github.com/0xddy/dht-crawler/blob/master/LICENSE)
一个基于 Rust 和 Tokio 实现的高性能分布式哈希表 (DHT) 爬虫库。它能够加入 BitTorrent DHT 网络，监听并自动获取种子的元数据（Metadata/InfoHash）。
@@ -45,6 +45,7 @@ graph TD
%% 业务逻辑分支
Logic -->|Discover Node| NodeMgr[Node Queue]
Logic -->|Discover InfoHash| HashQ[Hash Queue]
Logic -->|On Error| ErrorCb[User on_error Callback]
%% 元数据抓取子系统
subgraph Metadata [Metadata Subsystem]
@@ -68,6 +69,7 @@ graph TD
* N 个 Packet Worker 并行消费 Channel 中的数据。
* 负责 Bencode 解码、KRPC 协议解析、消息路由（Query/Response）。
* 高效处理 `get_peers` 和 `announce_peer` 消息，提取 InfoHash。
* 运行时错误通过 `on_error` 回调上报，不触发 panic，便于 JNI 等集成场景。
3. **元数据调度 (Metadata Subsystem)**:
* 提取出的 InfoHash 进入独立的 Hash Queue。
@@ -96,14 +98,12 @@ dht-crawler = { version = "0.1", features = ["metrics"] }
```rust
use dht_crawler::prelude::*;
use std::sync::Arc;
#[tokio::main]
async fn main() -> Result<()> {
// 1. 配置爬虫参数
let options = DHTOptions {
port: 12313,
auto_metadata: true, // 开启自动元数据获取
..Default::default()
};
@@ -111,12 +111,20 @@ async fn main() -> Result<()> {
let server = DHTServer::new(options).await?;
println!("DHT Server 启动于端口 12313...");
// 3. 注册回调函数：当成功获取到种子元数据时触发
// 3. 注册错误回调：运行时错误通过回调输出，避免 panic（适合 JNI/嵌入式场景）
server.on_error(|err| {
eprintln!("DHT 错误: {}", err);
});
// 4. 注册回调：成功获取到种子元数据时触发
server.on_torrent(move |torrent| {
println!("🎉 抓取成功: {} (文件数: {})", torrent.name, torrent.files.len());
});
// 4. 启动服务
// 5. 可选：在拉取元数据前过滤 info_hash，返回 true 表示允许拉取
server.on_metadata_fetch(|_hash| async move { true });
// 6. 启动服务
server.start().await?;
Ok(())
}
@@ -132,61 +140,99 @@ async fn main() -> Result<()> {
let options = DHTOptions {
// 监听端口
port: 12313,
// 网络模式：Ipv4Only, Ipv6Only, 或 DualStack (默认)
netmode: NetMode::Ipv4Only,
// 是否自动尝试从 peers 获取元数据
auto_metadata: true,
// 元数据获取超时时间 (秒)
metadata_timeout: 5,
// 元数据下载队列大小，建议根据内存大小调整
max_metadata_queue_size: 100000,
// 同时进行元数据下载的并发任务数
max_metadata_worker_count: 1000,
// 节点池容量（DHT 节点队列）
node_queue_capacity: 100000,
// InfoHash 发现队列容量
hash_queue_capacity: 10000,
..Default::default()
};
```
## 🛠️ 性能优化与编译选项
**可选 API**：`server.set_filter(|info_hash_hex| bool)` 可在发现阶段过滤要处理的 info_hash（返回 `true` 表示允许进入元数据队列）。
为了在生产环境中获得最佳性能，本库提供了几个可选的 Feature 和编译建议。
## 错误处理
### 1. 启用 `mimalloc` (内存优化)
库内采用严格的错误处理策略，避免底层异常导致进程崩溃，便于集成 JNI 或嵌入式场景。
在长运行的高并发场景下，使用 `mimalloc` 替代默认内存分配器可以降低 10-30% 的内存占用。
### 错误类型 `DHTError`
```rust
use dht_crawler::{DHTError, Result};
// 错误变体包括：
// - DHTError::Network(io::Error) — 网络/IO 错误
// - DHTError::Init(String) — 初始化错误（如 socket、worker）
// - DHTError::Internal(String) — 内部逻辑错误
// - DHTError::LockPoisoned(String) — 锁中毒（预留）
// - DHTError::Other(String) — 其他
```
### 注册错误回调 `on_error`
运行时错误（如协议处理失败）会通过回调上报，而不会 panic：
```rust
let server = DHTServer::new(options).await?;
// 将错误输出到 stderr 或接入自己的日志/监控
server.on_error(|err| {
log::error!("DHT 运行时错误: {}", err);
});
// JNI 场景示例：将错误传回 Java 层
// server.on_error(|err| {
// jni_callback_on_error(env, err.to_string());
// });
```
### 同步错误：`Result` 传播
`DHTServer::new()` 和 `server.start()` 返回 `Result`，调用方需处理或传播：
```rust
#[tokio::main]
async fn main() -> Result<()> {
let server = DHTServer::new(options).await?; // 初始化失败会返回 Err
server.on_error(|e| eprintln!("{}", e));
server.start().await?; // 启动失败（如 socket 绑定）会返回 Err
Ok(())
}
```
## 编译示例
### 1. 启用 `mimalloc`（内存优化）
在长运行的高并发场景下，使用 `mimalloc` 可降低约 10–30% 内存占用。本库的 `mimalloc` feature 仅用于方便编译/运行示例；若将本库作为依赖使用，请在自己的 bin 项目中单独引入并配置 mimalloc 全局分配器。
**运行示例代码：**
```bash
cargo run --release --example dht_crawler_example --features mimalloc
```
**在项目中使用：**
只需在你的 `Cargo.toml` 和 `main.rs` 中配置全局分配器即可（无需依赖本库的 feature，直接引入 mimalloc crate）。
### 2. 启用 `metrics`（监控）
### 2. 启用 `metrics` (监控)
启用后，可通过 HTTP 接口拉取 Prometheus 格式的监控数据。
启用后，可以通过 HTTP 接口拉取 Prometheus 格式的监控数据。
**启动带监控的示例：**
```bash
cargo run --release --example dht_crawler_example --features metrics
```
*监控地址：http://localhost:9000/metrics*
### 3. 交叉编译 (Linux)
推荐使用以下命令编译 Linux 生产环境版本：
```bash
cargo build --release --target x86_64-unknown-linux-gnu --examples --features mimalloc,metrics
```
> **注意**：`mimalloc` feature 主要是为了方便示例程序 (`examples/`) 的编译。在将其作为库引用时，建议你在自己的 `bin` 项目中独立配置内存分配器。
监控地址：http://localhost:9000/metrics
## 📜 许可证
examples/main.rs
+9 -2
View File
@@ -22,11 +22,13 @@ async fn main() -> Result<()> {
// 初始化 Prometheus metrics 导出器
#[cfg(feature = "metrics")]
{
let addr: SocketAddr = "0.0.0.0:9000".parse().expect("无效的地址");
let addr: SocketAddr = "0.0.0.0:9000"
.parse()
.map_err(|e| DHTError::Init(format!("无效的 metrics 监听地址: {e}")))?;
PrometheusBuilder::new()
.with_http_listener(addr)
.install()
.expect("无法安装 Prometheus metrics 导出器");
.map_err(|e| DHTError::Init(format!("无法安装 Prometheus metrics 导出器: {e}")))?;
log::info!("📊 Prometheus metrics 导出器已启动，访问 http://localhost:9000/metrics");
}
@@ -49,6 +51,11 @@ async fn main() -> Result<()> {
log::info!("🚀 DHT Server 启动，监听端口: {}", options.port);
// 注册错误回调，将运行时错误输出而不是 panic
server.on_error(|err| {
log::error!("DHT 运行时错误: {}", err);
});
// 设置 torrent 回调
server.on_torrent(move |_torrent| {
let _count = torrent_count_clone.fetch_add(1, Ordering::Relaxed) + 1;
jni/callbacks.rs
+56
View File
@@ -0,0 +1,56 @@
use crate::DHTServer;
use crate::jni_bindings::env::JavaCallback;
use crate::jni_bindings::types::torrent_info_to_java;
use jni::objects::JValue;
use std::sync::Arc;
/// 向 `DHTServer` 注册所有 Java 回调（on_torrent、on_error）。
///
/// `callback` 持有 Java listener 的全局引用，可跨线程安全使用。
pub fn register_callbacks(server: &Arc<DHTServer>, callback: JavaCallback) {
register_on_torrent(server, callback.clone());
register_on_error(server, callback);
}
/// 注册 on_torrent 回调：Rust TorrentInfo → Java listener.onTorrent(TorrentInfo)
fn register_on_torrent(server: &Arc<DHTServer>, callback: JavaCallback) {
server.on_torrent(move |info| {
let result = callback.with_env(|env, listener| {
// 映射为 Java TorrentInfo 对象
let j_info = torrent_info_to_java(env, &info).map_err(|e| {
log::error!("TorrentInfo 转换 Java 对象失败: {e}");
e
})?;
env.call_method(
listener,
"onTorrent",
"(Lcn/lmcw/dht/model/TorrentInfo;)V",
&[JValue::Object(&j_info)],
)?;
Ok(())
});
if let Err(e) = result {
log::error!("回调 onTorrent 失败: {e}");
}
});
}
/// 注册 on_error 回调：Rust DHTError → Java listener.onError(String)
fn register_on_error(server: &Arc<DHTServer>, callback: JavaCallback) {
server.on_error(move |err| {
let msg = err.to_string();
let result = callback.with_env(|env, listener| {
let j_msg = env.new_string(&msg)?;
env.call_method(
listener,
"onError",
"(Ljava/lang/String;)V",
&[JValue::Object(&j_msg.into())],
)?;
Ok(())
});
if let Err(e) = result {
log::error!("回调 onError 失败: {e}");
}
});
}
jni/env.rs
+71
View File
@@ -0,0 +1,71 @@
use jni::{JavaVM, JNIEnv};
use jni::objects::{JObject, JValue};
use std::sync::{Arc, Mutex};
/// 跨线程安全的 Java 回调持有者。
/// 保存 JavaVM 指针和 Java 侧 listener 的全局引用，
/// 在任意 Rust 线程中都可 attach 并回调 Java 方法。
#[derive(Clone)]
pub struct JavaCallback {
vm: Arc<JavaVM>,
/// listener 对象的全局引用（不可跨线程直接用，需通过 vm 先 attach）
listener: Arc<Mutex<jni::objects::GlobalRef>>,
}
impl JavaCallback {
/// 从当前 JNI 调用线程创建，把 listener jobject 升级为全局引用。
pub fn new(env: &mut JNIEnv, listener: &JObject) -> jni::errors::Result<Self> {
let vm = env.get_java_vm()?;
let global_ref = env.new_global_ref(listener)?;
Ok(Self {
vm: Arc::new(vm),
listener: Arc::new(Mutex::new(global_ref)),
})
}
/// 在任意线程中 attach JVM，执行 `f(env, listener)` 后自动 detach。
/// `f` 内可调用 Java 方法；出错时 f 应检查 exception 并返回 Err。
pub fn with_env<F, R>(&self, f: F) -> jni::errors::Result<R>
where
F: FnOnce(&mut JNIEnv, &JObject) -> jni::errors::Result<R>,
{
let mut guard = self.vm.attach_current_thread()?;
let listener_guard = self.listener.lock().unwrap();
let result = f(&mut guard, listener_guard.as_obj());
// 检查并清除 Java 侧遗留的异常，避免后续 JNI 调用受污染
if let Err(ref _e) = result {
if guard.exception_check().unwrap_or(false) {
let _ = guard.exception_clear();
}
}
result
}
/// 调用 Java listener 的方法（用于简单通知）。
pub fn call_void_method(
&self,
method_name: &str,
sig: &str,
args: &[JValue],
) -> jni::errors::Result<()> {
self.with_env(|env, obj| {
env.call_method(obj, method_name, sig, args)?;
Ok(())
})
}
}
/// 安全地执行 JNI 代码块，捕获 Rust panic，避免 JVM 崩溃。
/// 若发生 panic，向 Java 抛出 RuntimeException 后返回 `default`。
#[macro_export]
macro_rules! jni_catch {
($env:expr, $default:expr, $body:expr) => {{
match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| $body)) {
Ok(v) => v,
Err(_) => {
let _ = $env.throw_new("java/lang/RuntimeException", "Rust internal panic");
$default
}
}
}};
}
jni/exports.rs
+151
View File
@@ -0,0 +1,151 @@
use crate::jni_bindings::callbacks::register_callbacks;
use crate::jni_bindings::env::JavaCallback;
use crate::jni_bindings::server::{destroy_handle, handle_ref, into_handle_ptr, ServerHandle};
use crate::jni_bindings::types::java_to_dht_options_or_default;
use jni::JNIEnv;
use jni::objects::{JClass, JObject};
use jni::sys::{jint, jlong};
// ──────────────────────────────────────────────────────────────────────────────
// cn.lmcw.dht.DhtCrawlerJni 的 JNI 导出
// ──────────────────────────────────────────────────────────────────────────────
/// 创建 DHTServer 并返回句柄（jlong）。
///
/// Java 签名：`native long createServer(DHTOptions options, DhtListener listener);`
///
/// - `options`：`cn.lmcw.dht.model.DHTOptions` 对象，或 null 则使用默认选项。
/// - `listener`：`cn.lmcw.dht.DhtListener` 实现，或 null 则不注册回调。
/// - 返回：服务器句柄（成功）或 0（失败）。
#[unsafe(no_mangle)]
pub extern "system" fn Java_cn_lmcw_dht_DhtCrawlerJni_createServer(
mut env: JNIEnv,
_class: JClass,
options: JObject,
listener: JObject,
) -> jlong {
crate::jni_catch!(&mut env, 0, {
// 解析选项
let opts = match java_to_dht_options_or_default(&mut env, &options) {
Ok(o) => o,
Err(e) => {
let _ = env.throw_new("java/lang/IllegalArgumentException", &e.to_string());
return 0;
}
};
// 创建 ServerHandle（初始化 runtime + DHTServer）
let handle = match ServerHandle::new(opts) {
Ok(h) => h,
Err(e) => {
let _ = env.throw_new("java/lang/RuntimeException", &e);
return 0;
}
};
// 若提供了 listener，注册回调
if !listener.is_null() {
match JavaCallback::new(&mut env, &listener) {
Ok(cb) => register_callbacks(&handle.server, cb),
Err(e) => {
let _ = env.throw_new("java/lang/RuntimeException", &e.to_string());
return 0;
}
}
}
into_handle_ptr(handle)
})
}
/// 启动 DHTServer（在后台 tokio 任务中运行，不阻塞 JNI 线程）。
///
/// Java 签名：`native void startServer(long handle);`
#[unsafe(no_mangle)]
pub extern "system" fn Java_cn_lmcw_dht_DhtCrawlerJni_startServer(
mut env: JNIEnv,
_class: JClass,
handle: jlong,
) {
crate::jni_catch!(&mut env, (), {
let h = unsafe {
match handle_ref(handle) {
Some(h) => h,
None => {
let _ = env.throw_new("java/lang/IllegalArgumentException", "无效的服务器句柄");
return;
}
}
};
if let Err(e) = h.start() {
let _ = env.throw_new("java/lang/RuntimeException", &e);
}
});
}
/// 停止 DHTServer（发送关闭信号，不释放资源）。
///
/// Java 签名：`native void stopServer(long handle);`
#[unsafe(no_mangle)]
pub extern "system" fn Java_cn_lmcw_dht_DhtCrawlerJni_stopServer(
mut env: JNIEnv,
_class: JClass,
handle: jlong,
) {
crate::jni_catch!(&mut env, (), {
let h = unsafe {
match handle_ref(handle) {
Some(h) => h,
None => {
let _ = env.throw_new("java/lang/IllegalArgumentException", "无效的服务器句柄");
return;
}
}
};
h.stop();
});
}
/// 销毁 DHTServer：停止并释放所有 Rust 资源（包括 tokio runtime）。
/// 调用后 Java 侧不得再使用该句柄。
///
/// Java 签名：`native void destroyServer(long handle);`
#[unsafe(no_mangle)]
pub extern "system" fn Java_cn_lmcw_dht_DhtCrawlerJni_destroyServer(
mut env: JNIEnv,
_class: JClass,
handle: jlong,
) {
crate::jni_catch!(&mut env, (), {
if handle == 0 {
return;
}
// 先停止，再释放
unsafe {
if let Some(h) = handle_ref(handle) {
h.stop();
}
destroy_handle(handle);
}
});
}
/// 获取节点池当前大小。
///
/// Java 签名：`native int getNodePoolSize(long handle);`
#[unsafe(no_mangle)]
pub extern "system" fn Java_cn_lmcw_dht_DhtCrawlerJni_getNodePoolSize(
mut env: JNIEnv,
_class: JClass,
handle: jlong,
) -> jint {
crate::jni_catch!(&mut env, 0, {
let h = unsafe {
match handle_ref(handle) {
Some(h) => h,
None => return 0,
}
};
h.node_pool_size() as jint
})
}
jni/java/README.md
+93
View File
@@ -0,0 +1,93 @@
# dht-crawler-jni Java 示例
本目录是一个 Gradle 管理的 Java 项目，演示如何通过 JNI 使用 `dht-crawler` Rust 库。
## 项目结构
```
java/
├── build.gradle # Gradle 构建脚本
├── settings.gradle
└── src/main/java/cn/lmcw/dht/
├── model/
│ ├── TorrentInfo.java # 与 Rust TorrentInfo 一一对应
│ ├── FileInfo.java # 与 Rust FileInfo 一一对应
│ └── DHTOptions.java # 与 Rust DHTOptions 一一对应
├── DhtListener.java # 事件回调接口
├── DhtCrawlerJni.java # JNI 绑定类（native 方法声明）
└── DhtCrawlerExample.java # 可运行示例
```
## 快速开始
### 方式一：直接下载 Release 中的 JAR 运行（推荐）
从 GitHub Release 页面下载两个文件：
1. `dht-crawler-jni-example-<version>.jar` — 平台无关的 fat JAR
2. 对应平台的 JNI 动态库 zip（如 `dht_crawler_jni-<version>-x86_64-unknown-linux-gnu.zip`），解压得到 `libdht_crawler.so` / `dht_crawler.dll` / `libdht_crawler.dylib`
将 JAR 和动态库放到同一目录，然后执行：
```bash
# Linux / macOS
java -Djava.library.path=. -jar dht-crawler-jni-example-<version>.jar
# Windows（将 dht_crawler.dll 放在同目录）
java -Djava.library.path=. -jar dht-crawler-jni-example-<version>.jar
```
### 方式二：从源码编译并运行
#### 1. 编译 Rust JNI 动态库
在仓库**根目录**执行：
```bash
# 本机（Linux 产出 libdht_crawler.so，Windows 产出 dht_crawler.dll，macOS 产出 libdht_crawler.dylib）
cargo build --release --features jni
```
产物路径：`target/release/`
#### 2. 运行 Java 示例
在本目录（`jni/java/`）执行：
```bash
# 使用默认库路径（../../../target/release）
gradle run
# 自定义库路径
gradle run -Plib.path=/path/to/your/lib
```
#### 3. 构建 fat JAR
```bash
gradle shadowJar
# 产物：build/libs/dht-crawler-jni-example-<version>.jar
```
## 在自己的项目中集成
1. 将 `src/main/java/cn/lmcw/dht/` 下的文件复制到你的项目。
2. 将对应平台的 so/dll/dylib 放入 `java.library.path` 可访问的目录。
3. 确保 JVM 启动时加了 `-Djava.library.path=<路径>`。
## JNI 生命周期
```java
DHTOptions options = new DHTOptions().setPort(6881);
long handle = DhtCrawlerJni.createServer(options, listener);
DhtCrawlerJni.startServer(handle);
// ... 运行 ...
DhtCrawlerJni.stopServer(handle);
DhtCrawlerJni.destroyServer(handle); // 必须调用，释放 Rust 资源
```
## 注意事项
- `destroyServer` 必须在不再使用后调用，否则 Rust 侧的 tokio runtime 和连接不会释放。
- 回调方法（`onTorrent`、`onError`）在 Rust 的 tokio 工作线程中触发，请确保实现是线程安全的。
- 同一个句柄不要在多个线程中并发调用 `destroyServer`。
jni/java/build.gradle
+39
View File
@@ -0,0 +1,39 @@
plugins {
id 'java'
id 'application'
id 'com.github.johnrengelman.shadow' version '8.1.1'
}
group = 'cn.lmcw.dht'
version = project.findProperty('projectVersion') ?: '0.1.0'
java {
sourceCompatibility = JavaVersion.VERSION_11
targetCompatibility = JavaVersion.VERSION_11
}
application {
mainClass = 'cn.lmcw.dht.DhtCrawlerExample'
def libPath = project.findProperty('lib.path') ?: '.'
applicationDefaultJvmArgs = ["-Djava.library.path=${libPath}"]
}
repositories {
mavenCentral()
}
// 运行时将 native 库路径传入 JVM
tasks.withType(JavaExec).configureEach {
def libPath = project.findProperty('lib.path') ?: '../../../target/release'
jvmArgs "-Djava.library.path=${libPath}"
}
// Fat JAR（含所有依赖 + Main-Class）：执行 `gradle shadowJar` 生成
shadowJar {
archiveBaseName = 'dht-crawler-jni-example'
archiveClassifier = ''
archiveVersion = version
manifest {
attributes 'Main-Class': 'cn.lmcw.dht.DhtCrawlerExample'
}
}
jni/java/settings.gradle
+1
View File
@@ -0,0 +1 @@
rootProject.name = 'dht-crawler-jni'
jni/java/src/main/java/cn/lmcw/dht/DhtCrawlerExample.java
+82
View File
@@ -0,0 +1,82 @@
package cn.lmcw.dht;
import cn.lmcw.dht.model.DHTOptions;
import cn.lmcw.dht.model.TorrentInfo;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicLong;
/**
* DHT-Crawler JNI 使用示例。
*
* <h3>运行前准备</h3>
* <ol>
* <li>在仓库根目录编译 Rust JNI 产物：
* <pre>cargo build --release --features jni</pre>
* </li>
* <li>在 {@code jni/java/} 目录下运行（{@code lib.path} 指向 so/dll/dylib 所在目录）：
* <pre>gradle run -Plib.path=../../target/release</pre>
* 或直接用默认路径（{@code ../../../target/release}）：
* <pre>gradle run</pre>
* </li>
* </ol>
*/
public class DhtCrawlerExample {
public static void main(String[] args) throws InterruptedException {
// 1. 构造配置
DHTOptions options = new DHTOptions()
.setPort(6881)
.setNetMode(0) // 0 = IPv4 Only
.setMetadataTimeout(5L)
.setMaxMetadataQueueSize(50_000)
.setMaxMetadataWorkerCount(500);
AtomicLong torrentCount = new AtomicLong();
CountDownLatch shutdown = new CountDownLatch(1);
// 2. 实现回调
DhtListener listener = new DhtListener() {
@Override
public void onTorrent(TorrentInfo info) {
long n = torrentCount.incrementAndGet();
System.out.printf("[#%d] %s (%s) files=%d%n",
n, info.getName(), info.getInfoHash(),
info.getFiles() != null ? info.getFiles().size() : 0);
}
@Override
public void onError(String message) {
System.err.println("[ERROR] " + message);
}
};
// 3. 创建并启动服务器
long handle = DhtCrawlerJni.createServer(options, listener);
if (handle == 0) {
System.err.println("创建服务器失败");
return;
}
System.out.println("DHT 服务器已创建，正在启动...");
DhtCrawlerJni.startServer(handle);
System.out.println("DHT 服务器已启动，端口 " + options.getPort());
// 4. 注册 JVM 关闭钩子，确保资源释放
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
System.out.println("\n收到退出信号，正在停止...");
DhtCrawlerJni.stopServer(handle);
try { Thread.sleep(500); } catch (InterruptedException ignored) {}
DhtCrawlerJni.destroyServer(handle);
System.out.println("服务器已销毁，共发现种子：" + torrentCount.get());
shutdown.countDown();
}));
// 5. 定期打印节点池大小
System.out.println("按 Ctrl+C 退出。每 10 秒打印一次节点池大小...");
while (true) {
Thread.sleep(10_000);
int poolSize = DhtCrawlerJni.getNodePoolSize(handle);
System.out.println("节点池大小: " + poolSize + " 已发现种子: " + torrentCount.get());
}
}
}
jni/java/src/main/java/cn/lmcw/dht/DhtCrawlerJni.java
+68
View File
@@ -0,0 +1,68 @@
package cn.lmcw.dht;
import cn.lmcw.dht.model.DHTOptions;
/**
* Rust DHT-Crawler 库的 JNI 绑定入口。
*
* <p>所有方法均为 {@code static native}，通过 JNI 调用 Rust 实现。
* 加载顺序由静态初始化块保证，使用前无需手动调用。</p>
*
* <h3>生命周期</h3>
* <pre>{@code
* long handle = DhtCrawlerJni.createServer(options, listener);
* DhtCrawlerJni.startServer(handle);
* // ... 运行中 ...
* DhtCrawlerJni.stopServer(handle);
* DhtCrawlerJni.destroyServer(handle); // 必须调用，否则 Rust 资源泄漏
* }</pre>
*/
public final class DhtCrawlerJni {
static {
System.loadLibrary("dht_crawler");
}
private DhtCrawlerJni() {}
/**
* 创建并初始化 DHT 服务器，返回 Rust 侧句柄。
*
* @param options 服务器配置，传 {@code null} 则使用 Rust 侧默认值
* @param listener 事件回调，传 {@code null} 则不注册任何回调
* @return 服务器句柄（非 0 表示成功），或 0（失败，同时会向 JVM 抛出异常）
*/
public static native long createServer(DHTOptions options, DhtListener listener);
/**
* 启动已创建的 DHT 服务器（非阻塞）。
* 服务器在 Rust tokio runtime 后台运行，此方法立即返回。
*
* @param handle {@link #createServer} 返回的句柄
*/
public static native void startServer(long handle);
/**
* 向服务器发送停止信号（非阻塞）。
* 调用后服务器将优雅退出，但资源尚未释放，需继续调用 {@link #destroyServer}。
*
* @param handle {@link #createServer} 返回的句柄
*/
public static native void stopServer(long handle);
/**
* 销毁服务器并释放所有 Rust 侧资源（tokio runtime、连接等）。
* <strong>必须调用</strong>，否则发生内存泄漏。调用后不得再使用该句柄。
*
* @param handle {@link #createServer} 返回的句柄
*/
public static native void destroyServer(long handle);
/**
* 获取当前节点池（routing table）中的节点数量。
*
* @param handle {@link #createServer} 返回的句柄
* @return 节点数量，句柄无效时返回 0
*/
public static native int getNodePoolSize(long handle);
}
jni/java/src/main/java/cn/lmcw/dht/DhtListener.java
+25
View File
@@ -0,0 +1,25 @@
package cn.lmcw.dht;
import cn.lmcw.dht.model.TorrentInfo;
/**
* DHT 爬虫事件回调接口。
* <p>实现此接口并传入 {@link DhtCrawlerJni#createServer} 即可接收事件通知。
* 回调在 Rust 内部的 tokio 工作线程触发，实现方需自行保证线程安全。</p>
*/
public interface DhtListener {
/**
* 成功获取到 torrent metadata 时触发。
*
* @param info 完整的 torrent 信息对象，由 Rust JNI 层映射构造
*/
void onTorrent(TorrentInfo info);
/**
* 内部发生错误时触发。
*
* @param message Rust 侧错误的文本描述
*/
void onError(String message);
}
jni/java/src/main/java/cn/lmcw/dht/model/DHTOptions.java
+86
View File
@@ -0,0 +1,86 @@
package cn.lmcw.dht.model;
/**
* 与 Rust {@code DHTOptions} 一一对应的配置对象。
* <p>通过 JNI 传入 Rust 侧，由 Rust 读取各字段构造 {@code DHTOptions}。</p>
*
* <h3>netMode 取值</h3>
* <ul>
* <li>0 - IPv4 Only</li>
* <li>1 - IPv6 Only</li>
* <li>2 - Dual Stack（默认）</li>
* </ul>
*/
public final class DHTOptions {
/** 监听端口，默认 6881 */
private int port = 6881;
/** 获取 metadata 超时（秒），默认 3 */
private long metadataTimeout = 3L;
/** metadata 队列最大容量，默认 100000 */
private int maxMetadataQueueSize = 100_000;
/** 并发 metadata 拉取 worker 数量，默认 1000 */
private int maxMetadataWorkerCount = 1_000;
/** 节点队列容量，默认 100000 */
private int nodeQueueCapacity = 100_000;
/** hash 队列容量，默认 10000 */
private int hashQueueCapacity = 10_000;
/**
* 网络模式：0=IPv4Only, 1=IPv6Only, 2=DualStack（默认 0 / IPv4Only，
* 与 Rust 侧 {@code NetMode::Ipv4Only} 对应）
*/
private int netMode = 0;
public DHTOptions() {}
public int getPort() { return port; }
public DHTOptions setPort(int port) { this.port = port; return this; }
public long getMetadataTimeout() { return metadataTimeout; }
public DHTOptions setMetadataTimeout(long metadataTimeout) {
this.metadataTimeout = metadataTimeout;
return this;
}
public int getMaxMetadataQueueSize() { return maxMetadataQueueSize; }
public DHTOptions setMaxMetadataQueueSize(int maxMetadataQueueSize) {
this.maxMetadataQueueSize = maxMetadataQueueSize;
return this;
}
public int getMaxMetadataWorkerCount() { return maxMetadataWorkerCount; }
public DHTOptions setMaxMetadataWorkerCount(int maxMetadataWorkerCount) {
this.maxMetadataWorkerCount = maxMetadataWorkerCount;
return this;
}
public int getNodeQueueCapacity() { return nodeQueueCapacity; }
public DHTOptions setNodeQueueCapacity(int nodeQueueCapacity) {
this.nodeQueueCapacity = nodeQueueCapacity;
return this;
}
public int getHashQueueCapacity() { return hashQueueCapacity; }
public DHTOptions setHashQueueCapacity(int hashQueueCapacity) {
this.hashQueueCapacity = hashQueueCapacity;
return this;
}
public int getNetMode() { return netMode; }
public DHTOptions setNetMode(int netMode) { this.netMode = netMode; return this; }
@Override
public String toString() {
return "DHTOptions{"
+ "port=" + port
+ ", metadataTimeout=" + metadataTimeout
+ ", netMode=" + netMode
+ '}';
}
}
jni/java/src/main/java/cn/lmcw/dht/model/FileInfo.java
+30
View File
@@ -0,0 +1,30 @@
package cn.lmcw.dht.model;
/**
* 与 Rust {@code FileInfo} 一一对应的 POJO。
* <p>表示一个种子内单个文件的路径与大小。</p>
*/
public final class FileInfo {
private final String path;
private final long size;
/**
* 供 JNI 层调用的全参构造器。
*
* @param path 文件相对路径
* @param size 文件字节数
*/
public FileInfo(String path, long size) {
this.path = path;
this.size = size;
}
public String getPath() { return path; }
public long getSize() { return size; }
@Override
public String toString() {
return "FileInfo{path='" + path + "', size=" + size + '}';
}
}
jni/java/src/main/java/cn/lmcw/dht/model/TorrentInfo.java
+70
View File
@@ -0,0 +1,70 @@
package cn.lmcw.dht.model;
import java.util.List;
/**
* 与 Rust {@code TorrentInfo} 一一对应的 POJO。
* <p>由 Rust JNI 层通过 {@code NewObject} / {@code SetField} 构造并填充，
* 通过 {@link cn.lmcw.dht.DhtListener#onTorrent(TorrentInfo)} 回调给 Java 层。</p>
*/
public final class TorrentInfo {
private final String infoHash;
private final String magnetLink;
private final String name;
private final long totalSize;
private final List<FileInfo> files;
private final long pieceLength;
private final List<String> peers;
private final long timestamp;
/**
* 供 JNI 层调用的全参构造器。
*
* @param infoHash 十六进制 info-hash 字符串
* @param magnetLink magnet 链接
* @param name 种子名称
* @param totalSize 总字节数
* @param files 文件列表
* @param pieceLength 分片长度（字节）
* @param peers 来源节点地址列表
* @param timestamp 发现时间戳（Unix 秒）
*/
public TorrentInfo(
String infoHash,
String magnetLink,
String name,
long totalSize,
List<FileInfo> files,
long pieceLength,
List<String> peers,
long timestamp) {
this.infoHash = infoHash;
this.magnetLink = magnetLink;
this.name = name;
this.totalSize = totalSize;
this.files = files;
this.pieceLength = pieceLength;
this.peers = peers;
this.timestamp = timestamp;
}
public String getInfoHash() { return infoHash; }
public String getMagnetLink() { return magnetLink; }
public String getName() { return name; }
public long getTotalSize() { return totalSize; }
public List<FileInfo> getFiles() { return files; }
public long getPieceLength() { return pieceLength; }
public List<String> getPeers() { return peers; }
public long getTimestamp() { return timestamp; }
@Override
public String toString() {
return "TorrentInfo{"
+ "infoHash='" + infoHash + '\''
+ ", name='" + name + '\''
+ ", totalSize=" + totalSize
+ ", files=" + (files != null ? files.size() : 0)
+ '}';
}
}
jni/mod.rs
+5
View File
@@ -0,0 +1,5 @@
mod callbacks;
mod env;
mod exports;
mod server;
mod types;
jni/server.rs
+76
View File
@@ -0,0 +1,76 @@
use crate::{DHTServer, DHTOptions};
use std::sync::Arc;
use tokio::runtime::Runtime;
/// JNI 侧持有的服务器句柄，包含 tokio runtime 和 DHTServer 实例。
/// 通过 `Box::into_raw` 转成 `jlong` 句柄传给 Java，
/// 在 destroy 时通过 `Box::from_raw` 恢复并 drop。
pub struct ServerHandle {
pub runtime: Runtime,
pub server: Arc<DHTServer>,
}
impl ServerHandle {
/// 在新建的 tokio runtime 里初始化 DHTServer。
pub fn new(options: DHTOptions) -> Result<Self, String> {
let runtime = Runtime::new().map_err(|e| format!("无法创建 tokio runtime: {e}"))?;
let server = runtime
.block_on(DHTServer::new(options))
.map_err(|e| format!("DHTServer 初始化失败: {e}"))?;
Ok(Self {
runtime,
server: Arc::new(server),
})
}
/// 在 runtime 里 spawn server.start()，不阻塞调用线程。
pub fn start(&self) -> Result<(), String> {
let server: Arc<DHTServer> = Arc::clone(&self.server);
self.runtime.spawn(async move {
if let Err(e) = server.start().await {
log::error!("DHT server 运行错误: {e}");
}
});
Ok(())
}
/// 发送关闭信号（非阻塞）。
pub fn stop(&self) {
self.server.shutdown();
}
/// 返回节点池大小。
pub fn node_pool_size(&self) -> usize {
self.server.get_node_pool_size()
}
}
// ──────────────────────────────────────────────────────────────────────────────
// 句柄指针工具
// ──────────────────────────────────────────────────────────────────────────────
/// 将 `ServerHandle` 装箱并返回原始指针，供 Java 以 `long` 持有。
pub fn into_handle_ptr(handle: ServerHandle) -> i64 {
Box::into_raw(Box::new(handle)) as i64
}
/// 从 Java 传入的 `long` 句柄获取不可变引用。
///
/// # Safety
/// 调用方必须确保句柄未被 destroy，且在单次 JNI 调用生命周期内使用。
pub unsafe fn handle_ref<'a>(ptr: i64) -> Option<&'a ServerHandle> {
if ptr == 0 {
return None;
}
Some(unsafe { &*(ptr as *const ServerHandle) })
}
/// 消费句柄：从裸指针重建 Box 并 drop，释放所有资源（包括 runtime）。
///
/// # Safety
/// 只能调用一次，调用后 Java 侧不得再使用该句柄。
pub unsafe fn destroy_handle(ptr: i64) {
if ptr != 0 {
drop(unsafe { Box::from_raw(ptr as *mut ServerHandle) });
}
}
jni/types.rs
+165
View File
@@ -0,0 +1,165 @@
use crate::{TorrentInfo, FileInfo, DHTOptions, types::NetMode};
use jni::JNIEnv;
use jni::objects::{JObject, JString, JValue};
use jni::sys::jlong;
// ──────────────────────────────────────────────────────────────────────────────
// 常量：Java 类全限定名
// ──────────────────────────────────────────────────────────────────────────────
const CLASS_TORRENT_INFO: &str = "cn/lmcw/dht/model/TorrentInfo";
const CLASS_FILE_INFO: &str = "cn/lmcw/dht/model/FileInfo";
const CLASS_DHT_OPTIONS: &str = "cn/lmcw/dht/model/DHTOptions";
const CLASS_ARRAY_LIST: &str = "java/util/ArrayList";
// ──────────────────────────────────────────────────────────────────────────────
// 基础类型转换
// ──────────────────────────────────────────────────────────────────────────────
/// Rust String → Java String（jstring）
pub fn rust_str_to_jstring<'local>(
env: &mut JNIEnv<'local>,
s: &str,
) -> jni::errors::Result<JObject<'local>> {
let js: JString<'local> = env.new_string(s)?;
Ok(js.into())
}
/// Java String（JObject）→ Rust String
pub fn jstring_to_rust(env: &mut JNIEnv, obj: &JObject) -> jni::errors::Result<String> {
let jstr = JString::from(env.new_local_ref(obj)?);
Ok(env.get_string(&jstr)?.into())
}
// ──────────────────────────────────────────────────────────────────────────────
// FileInfo: Rust → Java
// ──────────────────────────────────────────────────────────────────────────────
/// 将 Rust `FileInfo` 构造为 Java `cn.lmcw.dht.model.FileInfo` 对象。
pub fn file_info_to_java<'local>(
env: &mut JNIEnv<'local>,
fi: &FileInfo,
) -> jni::errors::Result<JObject<'local>> {
let cls = env.find_class(CLASS_FILE_INFO)?;
let path = rust_str_to_jstring(env, &fi.path)?;
let obj = env.new_object(
&cls,
"(Ljava/lang/String;J)V",
&[JValue::Object(&path), JValue::Long(fi.size as jlong)],
)?;
Ok(obj)
}
// ──────────────────────────────────────────────────────────────────────────────
// TorrentInfo: Rust → Java
// ──────────────────────────────────────────────────────────────────────────────
/// 将 Rust `TorrentInfo` 构造为 Java `cn.lmcw.dht.model.TorrentInfo` 对象。
/// files 字段被构造为 `java.util.ArrayList<FileInfo>`。
pub fn torrent_info_to_java<'local>(
env: &mut JNIEnv<'local>,
ti: &TorrentInfo,
) -> jni::errors::Result<JObject<'local>> {
let cls = env.find_class(CLASS_TORRENT_INFO)?;
// 构造 files list
let list = build_file_list(env, &ti.files)?;
// 构造 peers list（List<String>）
let peers_list = build_string_list(env, &ti.peers)?;
let info_hash = rust_str_to_jstring(env, &ti.info_hash)?;
let magnet = rust_str_to_jstring(env, &ti.magnet_link)?;
let name = rust_str_to_jstring(env, &ti.name)?;
let obj = env.new_object(
&cls,
"(Ljava/lang/String;Ljava/lang/String;Ljava/lang/String;JLjava/util/List;JLjava/util/List;J)V",
&[
JValue::Object(&info_hash),
JValue::Object(&magnet),
JValue::Object(&name),
JValue::Long(ti.total_size as jlong),
JValue::Object(&list),
JValue::Long(ti.piece_length as jlong),
JValue::Object(&peers_list),
JValue::Long(ti.timestamp as jlong),
],
)?;
Ok(obj)
}
/// 构造 `java.util.ArrayList` 并填入 FileInfo 对象列表。
fn build_file_list<'local>(
env: &mut JNIEnv<'local>,
files: &[FileInfo],
) -> jni::errors::Result<JObject<'local>> {
let list_cls = env.find_class(CLASS_ARRAY_LIST)?;
let list = env.new_object(&list_cls, "()V", &[])?;
for fi in files {
let jfi = file_info_to_java(env, fi)?;
env.call_method(&list, "add", "(Ljava/lang/Object;)Z", &[JValue::Object(&jfi)])?;
env.delete_local_ref(jfi)?;
}
Ok(list)
}
/// 构造 `java.util.ArrayList` 并填入字符串列表。
fn build_string_list<'local>(
env: &mut JNIEnv<'local>,
strs: &[String],
) -> jni::errors::Result<JObject<'local>> {
let list_cls = env.find_class(CLASS_ARRAY_LIST)?;
let list = env.new_object(&list_cls, "()V", &[])?;
for s in strs {
let js = rust_str_to_jstring(env, s)?;
env.call_method(&list, "add", "(Ljava/lang/Object;)Z", &[JValue::Object(&js)])?;
env.delete_local_ref(js)?;
}
Ok(list)
}
// ──────────────────────────────────────────────────────────────────────────────
// DHTOptions: Java → Rust
// ──────────────────────────────────────────────────────────────────────────────
/// 从 Java `cn.lmcw.dht.model.DHTOptions` 对象读取字段，构造 Rust `DHTOptions`。
pub fn java_to_dht_options(env: &mut JNIEnv, obj: &JObject) -> jni::errors::Result<DHTOptions> {
let port = env.get_field(obj, "port", "I")?.i()? as u16;
let metadata_timeout = env.get_field(obj, "metadataTimeout", "J")?.j()? as u64;
let max_metadata_queue_size =
env.get_field(obj, "maxMetadataQueueSize", "I")?.i()? as usize;
let max_metadata_worker_count =
env.get_field(obj, "maxMetadataWorkerCount", "I")?.i()? as usize;
let node_queue_capacity =
env.get_field(obj, "nodeQueueCapacity", "I")?.i()? as usize;
let hash_queue_capacity =
env.get_field(obj, "hashQueueCapacity", "I")?.i()? as usize;
let netmode_ord = env.get_field(obj, "netMode", "I")?.i()?;
let netmode = match netmode_ord {
0 => NetMode::Ipv4Only,
1 => NetMode::Ipv6Only,
_ => NetMode::DualStack,
};
Ok(DHTOptions {
port,
metadata_timeout,
max_metadata_queue_size,
max_metadata_worker_count,
netmode,
node_queue_capacity,
hash_queue_capacity,
})
}
/// 从 Java `cn.lmcw.dht.model.DHTOptions` 对象读取，或若为 null 则返回默认选项。
pub fn java_to_dht_options_or_default(
env: &mut JNIEnv,
obj: &JObject,
) -> jni::errors::Result<DHTOptions> {
if obj.is_null() {
Ok(DHTOptions::default())
} else {
java_to_dht_options(env, obj)
}
}
src/error.rs
+9
View File
@@ -5,6 +5,15 @@ pub enum DHTError {
#[error("网络错误: {0}")]
Network(#[from] std::io::Error),
#[error("锁中毒: {0}")]
LockPoisoned(String),
#[error("初始化错误: {0}")]
Init(String),
#[error("内部错误: {0}")]
Internal(String),
#[error("{0}")]
Other(String),
}
src/lib.rs
+4
View File
@@ -18,3 +18,7 @@ pub mod prelude {
pub use crate::server::DHTServer;
pub use crate::types::{DHTOptions, FileInfo, NetMode, TorrentInfo};
}
#[cfg(feature = "jni")]
#[path = "../jni/mod.rs"]
pub mod jni_bindings;
src/server.rs
+34 -9
View File
@@ -40,6 +40,7 @@ pub struct HashDiscovered {
type TorrentCallback = Arc<dyn Fn(TorrentInfo) + Send + Sync>;
type FilterCallback = Arc<dyn Fn(&str) -> bool + Send + Sync>;
type ErrorCallback = Arc<dyn Fn(crate::error::DHTError) + Send + Sync>;
#[derive(Clone)]
pub struct DHTServer {
@@ -51,6 +52,7 @@ pub struct DHTServer {
callback: Arc<RwLock<Option<TorrentCallback>>>,
filter: Arc<RwLock<Option<FilterCallback>>>,
on_metadata_fetch: Arc<RwLock<Option<MetadataFetchCallback>>>,
on_error_cb: Arc<RwLock<Option<ErrorCallback>>>,
node_queue: Arc<ShardedNodeQueue>,
hash_tx: mpsc::Sender<HashDiscovered>,
metadata_queue_len: Arc<AtomicUsize>,
@@ -153,6 +155,7 @@ impl DHTServer {
on_metadata_fetch,
node_queue: Arc::new(node_queue),
filter: Arc::new(RwLock::new(None)),
on_error_cb: Arc::new(RwLock::new(None)),
hash_tx,
metadata_queue_len,
max_metadata_queue_size,
@@ -167,7 +170,7 @@ impl DHTServer {
F: Fn(String) -> Fut + Send + Sync + 'static,
Fut: Future<Output = bool> + Send + 'static,
{
*self.on_metadata_fetch.write().unwrap() =
*self.on_metadata_fetch.write().unwrap_or_else(|e| e.into_inner()) =
Some(Arc::new(move |hash| Box::pin(callback(hash))));
}
@@ -175,14 +178,29 @@ impl DHTServer {
where
F: Fn(TorrentInfo) + Send + Sync + 'static,
{
*self.callback.write().unwrap() = Some(Arc::new(callback));
*self.callback.write().unwrap_or_else(|e| e.into_inner()) = Some(Arc::new(callback));
}
pub fn set_filter<F>(&self, filter: F)
where
F: Fn(&str) -> bool + Send + Sync + 'static,
{
*self.filter.write().unwrap() = Some(Arc::new(filter));
*self.filter.write().unwrap_or_else(|e| e.into_inner()) = Some(Arc::new(filter));
}
pub fn on_error<F>(&self, callback: F)
where
F: Fn(crate::error::DHTError) + Send + Sync + 'static,
{
*self.on_error_cb.write().unwrap_or_else(|e| e.into_inner()) = Some(Arc::new(callback));
}
fn emit_error(&self, error: crate::error::DHTError) {
if let Ok(cb) = self.on_error_cb.read() {
if let Some(f) = cb.as_ref() {
f(error);
}
}
}
pub fn get_node_pool_size(&self) -> usize {
@@ -198,7 +216,7 @@ impl DHTServer {
let workers = self.spawn_workers();
for sock in self.socket_providers.values().cloned() {
spawn_udp_listener(sock, workers.clone(), self.shutdown.clone());
spawn_udp_listener(sock, workers.clone(), self.shutdown.clone())?;
}
self.bootstrap().await;
@@ -342,7 +360,9 @@ impl DHTServer {
msg = rx.recv() => {
match msg {
Some((data, remote_addr, local_addr)) => {
let _ = server_clone.handle_message(data.as_ref(), remote_addr, local_addr).await;
if let Err(e) = server_clone.handle_message(data.as_ref(), remote_addr, local_addr).await {
server_clone.emit_error(e);
}
}
None => break,
}
@@ -478,7 +498,7 @@ impl DHTServer {
};
let hash_hex = hex::encode(info_hash_arr);
let filter_cb = self.filter.read().unwrap().clone();
let filter_cb = self.filter.read().unwrap_or_else(|e| e.into_inner()).clone();
if let Some(f) = filter_cb
&& !f(&hash_hex)
{
@@ -886,10 +906,14 @@ fn spawn_udp_listener(
socket: Arc<UdpSocket>,
mut workers: Vec<WorkerHandle>,
shutdown: CancellationToken,
) {
let local_addr = socket.local_addr().expect("socket to have IP address");
) -> crate::error::Result<()> {
let local_addr = socket
.local_addr()
.map_err(|e| crate::error::DHTError::Init(format!("socket 无法获取本地地址: {e}")))?;
if workers.is_empty() {
panic!("No worker supplied for UDP reader")
return Err(crate::error::DHTError::Init(
"spawn_udp_listener: 未提供任何 worker".to_string(),
));
}
tokio::spawn(async move {
let mut buffer = [0u8; 65536];
@@ -922,6 +946,7 @@ fn spawn_udp_listener(
}
}
});
Ok(())
}
enum ProcessUdpPacketError {
src/sharded.rs
+14 -14
View File
@@ -90,10 +90,10 @@ impl ShardedNodeQueue {
let shard_idx = self.addr_to_shard(&node.addr);
if node.addr.is_ipv6() {
let mut shard = self.shards_v6[shard_idx].lock().unwrap();
let mut shard = self.shards_v6[shard_idx].lock().unwrap_or_else(|e| e.into_inner());
shard.push(node);
} else {
let mut shard = self.shards_v4[shard_idx].lock().unwrap();
let mut shard = self.shards_v4[shard_idx].lock().unwrap_or_else(|e| e.into_inner());
shard.push(node);
}
}
@@ -109,7 +109,7 @@ impl ShardedNodeQueue {
if result.len() >= count {
break;
}
let mut s = shard.lock().unwrap();
let mut s = shard.lock().unwrap_or_else(|e| e.into_inner());
let nodes = s.pop_batch(per_shard);
result.extend(nodes);
}
@@ -119,7 +119,7 @@ impl ShardedNodeQueue {
if result.len() >= count {
break;
}
let mut s = shard.lock().unwrap();
let mut s = shard.lock().unwrap_or_else(|e| e.into_inner());
let nodes = s.pop_batch(per_shard);
result.extend(nodes);
}
@@ -130,7 +130,7 @@ impl ShardedNodeQueue {
break;
}
let mut s4 = self.shards_v4[i].lock().unwrap();
let mut s4 = self.shards_v4[i].lock().unwrap_or_else(|e| e.into_inner());
let nodes4 = s4.pop_batch(per_shard / 2);
result.extend(nodes4);
drop(s4);
@@ -139,7 +139,7 @@ impl ShardedNodeQueue {
break;
}
let mut s6 = self.shards_v6[i].lock().unwrap();
let mut s6 = self.shards_v6[i].lock().unwrap_or_else(|e| e.into_inner());
let nodes6 = s6.pop_batch(per_shard / 2);
result.extend(nodes6);
drop(s6);
@@ -185,7 +185,7 @@ impl ShardedNodeQueue {
break;
}
let s = shard.lock().unwrap();
let s = shard.lock().unwrap_or_else(|e| e.into_inner());
let shard_len = s.queue.len();
if shard_len == 0 {
@@ -214,7 +214,7 @@ impl ShardedNodeQueue {
let mut seen = 0usize;
for shard in shards {
let s = shard.lock().unwrap();
let s = shard.lock().unwrap_or_else(|e| e.into_inner());
for node in s.queue.iter() {
seen += 1;
@@ -238,12 +238,12 @@ impl ShardedNodeQueue {
let len_v4: usize = self
.shards_v4
.iter()
.map(|shard| shard.lock().unwrap().len())
.map(|shard| shard.lock().unwrap_or_else(|e| e.into_inner()).len())
.sum();
let len_v6: usize = self
.shards_v6
.iter()
.map(|shard| shard.lock().unwrap().len())
.map(|shard| shard.lock().unwrap_or_else(|e| e.into_inner()).len())
.sum();
len_v4 + len_v6
}
@@ -252,11 +252,11 @@ impl ShardedNodeQueue {
let empty_v4 = self
.shards_v4
.iter()
.all(|shard| shard.lock().unwrap().is_empty());
.all(|shard| shard.lock().unwrap_or_else(|e| e.into_inner()).is_empty());
let empty_v6 = self
.shards_v6
.iter()
.all(|shard| shard.lock().unwrap().is_empty());
.all(|shard| shard.lock().unwrap_or_else(|e| e.into_inner()).is_empty());
empty_v4 && empty_v6
}
@@ -265,11 +265,11 @@ impl ShardedNodeQueue {
Some(true) => self
.shards_v6
.iter()
.all(|shard| shard.lock().unwrap().is_empty()),
.all(|shard| shard.lock().unwrap_or_else(|e| e.into_inner()).is_empty()),
Some(false) => self
.shards_v4
.iter()
.all(|shard| shard.lock().unwrap().is_empty()),
.all(|shard| shard.lock().unwrap_or_else(|e| e.into_inner()).is_empty()),
None => self.is_empty(),
}
}