From 6e7c6f02e44428ac8defad7ce34a533a1658cca1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=A1=A5=E4=B8=8B=E7=BA=A2=E8=8D=AF?= Date: Fri, 13 Mar 2026 02:15:50 +0800 Subject: [PATCH] jni --- .github/workflows/release.yml | 101 +++++++++-- Cargo.toml | 5 +- README.md | 114 ++++++++---- examples/main.rs | 11 +- jni/callbacks.rs | 56 ++++++ jni/env.rs | 71 ++++++++ jni/exports.rs | 151 ++++++++++++++++ jni/java/README.md | 93 ++++++++++ jni/java/build.gradle | 39 +++++ jni/java/settings.gradle | 1 + .../java/cn/lmcw/dht/DhtCrawlerExample.java | 82 +++++++++ .../main/java/cn/lmcw/dht/DhtCrawlerJni.java | 68 ++++++++ .../main/java/cn/lmcw/dht/DhtListener.java | 25 +++ .../java/cn/lmcw/dht/model/DHTOptions.java | 86 +++++++++ .../main/java/cn/lmcw/dht/model/FileInfo.java | 30 ++++ .../java/cn/lmcw/dht/model/TorrentInfo.java | 70 ++++++++ jni/mod.rs | 5 + jni/server.rs | 76 ++++++++ jni/types.rs | 165 ++++++++++++++++++ src/error.rs | 9 + src/lib.rs | 4 + src/server.rs | 43 ++++- src/sharded.rs | 28 +-- 23 files changed, 1256 insertions(+), 77 deletions(-) create mode 100644 jni/callbacks.rs create mode 100644 jni/env.rs create mode 100644 jni/exports.rs create mode 100644 jni/java/README.md create mode 100644 jni/java/build.gradle create mode 100644 jni/java/settings.gradle create mode 100644 jni/java/src/main/java/cn/lmcw/dht/DhtCrawlerExample.java create mode 100644 jni/java/src/main/java/cn/lmcw/dht/DhtCrawlerJni.java create mode 100644 jni/java/src/main/java/cn/lmcw/dht/DhtListener.java create mode 100644 jni/java/src/main/java/cn/lmcw/dht/model/DHTOptions.java create mode 100644 jni/java/src/main/java/cn/lmcw/dht/model/FileInfo.java create mode 100644 jni/java/src/main/java/cn/lmcw/dht/model/TorrentInfo.java create mode 100644 jni/mod.rs create mode 100644 jni/server.rs create mode 100644 jni/types.rs diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index a98844e..c364452 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -12,6 +12,41 @@ env: CARGO_TERM_COLOR: always jobs: + # ────────────────────────────────────────────────────────────────── + # Java fat JAR（平台无关，只需构建一次） + # 运行方式：java -Djava.library.path= -jar dht-crawler-jni-example-.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 diff --git a/Cargo.toml b/Cargo.toml index 5a75305..bda76bf 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -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" diff --git a/README.md b/README.md index fe4ecc7..26633f5 100644 --- a/README.md +++ b/README.md @@ -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 ## 📜 许可证 diff --git a/examples/main.rs b/examples/main.rs index d2056f2..ef205ba 100644 --- a/examples/main.rs +++ b/examples/main.rs @@ -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; diff --git a/jni/callbacks.rs b/jni/callbacks.rs new file mode 100644 index 0000000..23f00fa --- /dev/null +++ b/jni/callbacks.rs @@ -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, 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, 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, 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}"); + } + }); +} diff --git a/jni/env.rs b/jni/env.rs new file mode 100644 index 0000000..61b185c --- /dev/null +++ b/jni/env.rs @@ -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, + /// listener 对象的全局引用（不可跨线程直接用，需通过 vm 先 attach） + listener: Arc>, +} + +impl JavaCallback { + /// 从当前 JNI 调用线程创建，把 listener jobject 升级为全局引用。 + pub fn new(env: &mut JNIEnv, listener: &JObject) -> jni::errors::Result { + 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(&self, f: F) -> jni::errors::Result + where + F: FnOnce(&mut JNIEnv, &JObject) -> jni::errors::Result, + { + 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 + } + } + }}; +} diff --git a/jni/exports.rs b/jni/exports.rs new file mode 100644 index 0000000..3916a12 --- /dev/null +++ b/jni/exports.rs @@ -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 + }) +} diff --git a/jni/java/README.md b/jni/java/README.md new file mode 100644 index 0000000..b53f8cf --- /dev/null +++ b/jni/java/README.md @@ -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-.jar` — 平台无关的 fat JAR +2. 对应平台的 JNI 动态库 zip（如 `dht_crawler_jni--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-.jar + +# Windows（将 dht_crawler.dll 放在同目录） +java -Djava.library.path=. -jar dht-crawler-jni-example-.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-.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`。 diff --git a/jni/java/build.gradle b/jni/java/build.gradle new file mode 100644 index 0000000..5683cd7 --- /dev/null +++ b/jni/java/build.gradle @@ -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' + } +} diff --git a/jni/java/settings.gradle b/jni/java/settings.gradle new file mode 100644 index 0000000..2f3e755 --- /dev/null +++ b/jni/java/settings.gradle @@ -0,0 +1 @@ +rootProject.name = 'dht-crawler-jni' diff --git a/jni/java/src/main/java/cn/lmcw/dht/DhtCrawlerExample.java b/jni/java/src/main/java/cn/lmcw/dht/DhtCrawlerExample.java new file mode 100644 index 0000000..eb76336 --- /dev/null +++ b/jni/java/src/main/java/cn/lmcw/dht/DhtCrawlerExample.java @@ -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 使用示例。 + * + *

运行前准备

+ *
    + *
  1. 在仓库根目录编译 Rust JNI 产物： + * 
    cargo build --release --features jni
    + *
    2. + *
  2. 在 {@code jni/java/} 目录下运行（{@code lib.path} 指向 so/dll/dylib 所在目录）： + * 
    gradle run -Plib.path=../../target/release
    + * 或直接用默认路径（{@code ../../../target/release}）： + * 
    gradle run
    + *
    3. + *
+ */ +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()); + } + } +} diff --git a/jni/java/src/main/java/cn/lmcw/dht/DhtCrawlerJni.java b/jni/java/src/main/java/cn/lmcw/dht/DhtCrawlerJni.java new file mode 100644 index 0000000..4800414 --- /dev/null +++ b/jni/java/src/main/java/cn/lmcw/dht/DhtCrawlerJni.java @@ -0,0 +1,68 @@ +package cn.lmcw.dht; + +import cn.lmcw.dht.model.DHTOptions; + +/** + * Rust DHT-Crawler 库的 JNI 绑定入口。 + * + *

所有方法均为 {@code static native}，通过 JNI 调用 Rust 实现。 + * 加载顺序由静态初始化块保证，使用前无需手动调用。

+ * + *

生命周期

+ * 
{@code
+ * long handle = DhtCrawlerJni.createServer(options, listener);
+ * DhtCrawlerJni.startServer(handle);
+ * // ... 运行中 ...
+ * DhtCrawlerJni.stopServer(handle);
+ * DhtCrawlerJni.destroyServer(handle); // 必须调用，否则 Rust 资源泄漏
+ * }
+ */ +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、连接等）。 + * 必须调用，否则发生内存泄漏。调用后不得再使用该句柄。 + * + * @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); +} diff --git a/jni/java/src/main/java/cn/lmcw/dht/DhtListener.java b/jni/java/src/main/java/cn/lmcw/dht/DhtListener.java new file mode 100644 index 0000000..fb8a335 --- /dev/null +++ b/jni/java/src/main/java/cn/lmcw/dht/DhtListener.java @@ -0,0 +1,25 @@ +package cn.lmcw.dht; + +import cn.lmcw.dht.model.TorrentInfo; + +/** + * DHT 爬虫事件回调接口。 + *

实现此接口并传入 {@link DhtCrawlerJni#createServer} 即可接收事件通知。 + * 回调在 Rust 内部的 tokio 工作线程触发，实现方需自行保证线程安全。

+ */ +public interface DhtListener { + + /** + * 成功获取到 torrent metadata 时触发。 + * + * @param info 完整的 torrent 信息对象，由 Rust JNI 层映射构造 + */ + void onTorrent(TorrentInfo info); + + /** + * 内部发生错误时触发。 + * + * @param message Rust 侧错误的文本描述 + */ + void onError(String message); +} diff --git a/jni/java/src/main/java/cn/lmcw/dht/model/DHTOptions.java b/jni/java/src/main/java/cn/lmcw/dht/model/DHTOptions.java new file mode 100644 index 0000000..dad39fd --- /dev/null +++ b/jni/java/src/main/java/cn/lmcw/dht/model/DHTOptions.java @@ -0,0 +1,86 @@ +package cn.lmcw.dht.model; + +/** + * 与 Rust {@code DHTOptions} 一一对应的配置对象。 + *

通过 JNI 传入 Rust 侧，由 Rust 读取各字段构造 {@code DHTOptions}。

+ * + *

netMode 取值

+ *
    + *
  • 0 - IPv4 Only
    • + *
  • 1 - IPv6 Only
    • + *
  • 2 - Dual Stack（默认）
    • + *
+ */ +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 + + '}'; + } +} diff --git a/jni/java/src/main/java/cn/lmcw/dht/model/FileInfo.java b/jni/java/src/main/java/cn/lmcw/dht/model/FileInfo.java new file mode 100644 index 0000000..46decf9 --- /dev/null +++ b/jni/java/src/main/java/cn/lmcw/dht/model/FileInfo.java @@ -0,0 +1,30 @@ +package cn.lmcw.dht.model; + +/** + * 与 Rust {@code FileInfo} 一一对应的 POJO。 + *

表示一个种子内单个文件的路径与大小。

+ */ +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 + '}'; + } +} diff --git a/jni/java/src/main/java/cn/lmcw/dht/model/TorrentInfo.java b/jni/java/src/main/java/cn/lmcw/dht/model/TorrentInfo.java new file mode 100644 index 0000000..0b16eb6 --- /dev/null +++ b/jni/java/src/main/java/cn/lmcw/dht/model/TorrentInfo.java @@ -0,0 +1,70 @@ +package cn.lmcw.dht.model; + +import java.util.List; + +/** + * 与 Rust {@code TorrentInfo} 一一对应的 POJO。 + *

由 Rust JNI 层通过 {@code NewObject} / {@code SetField} 构造并填充， + * 通过 {@link cn.lmcw.dht.DhtListener#onTorrent(TorrentInfo)} 回调给 Java 层。

+ */ +public final class TorrentInfo { + + private final String infoHash; + private final String magnetLink; + private final String name; + private final long totalSize; + private final List files; + private final long pieceLength; + private final List 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 files, + long pieceLength, + List 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 getFiles() { return files; } + public long getPieceLength() { return pieceLength; } + public List 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) + + '}'; + } +} diff --git a/jni/mod.rs b/jni/mod.rs new file mode 100644 index 0000000..a85f488 --- /dev/null +++ b/jni/mod.rs @@ -0,0 +1,5 @@ +mod callbacks; +mod env; +mod exports; +mod server; +mod types; diff --git a/jni/server.rs b/jni/server.rs new file mode 100644 index 0000000..c062056 --- /dev/null +++ b/jni/server.rs @@ -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, +} + +impl ServerHandle { + /// 在新建的 tokio runtime 里初始化 DHTServer。 + pub fn new(options: DHTOptions) -> Result { + 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 = 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) }); + } +} diff --git a/jni/types.rs b/jni/types.rs new file mode 100644 index 0000000..e685418 --- /dev/null +++ b/jni/types.rs @@ -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> { + 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 { + 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> { + 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`。 +pub fn torrent_info_to_java<'local>( + env: &mut JNIEnv<'local>, + ti: &TorrentInfo, +) -> jni::errors::Result> { + let cls = env.find_class(CLASS_TORRENT_INFO)?; + + // 构造 files list + let list = build_file_list(env, &ti.files)?; + + // 构造 peers list（List） + 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> { + 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> { + 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 { + 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 { + if obj.is_null() { + Ok(DHTOptions::default()) + } else { + java_to_dht_options(env, obj) + } +} diff --git a/src/error.rs b/src/error.rs index 4599c8e..7563444 100644 --- a/src/error.rs +++ b/src/error.rs @@ -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), } diff --git a/src/lib.rs b/src/lib.rs index b2303df..64f2e62 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -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; diff --git a/src/server.rs b/src/server.rs index f3939df..03d1659 100644 --- a/src/server.rs +++ b/src/server.rs @@ -40,6 +40,7 @@ pub struct HashDiscovered { type TorrentCallback = Arc; type FilterCallback = Arc bool + Send + Sync>; +type ErrorCallback = Arc; #[derive(Clone)] pub struct DHTServer { @@ -51,6 +52,7 @@ pub struct DHTServer { callback: Arc>>, filter: Arc>>, on_metadata_fetch: Arc>>, + on_error_cb: Arc>>, node_queue: Arc, hash_tx: mpsc::Sender, metadata_queue_len: Arc, @@ -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 + 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(&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(&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, mut workers: Vec, 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 { diff --git a/src/sharded.rs b/src/sharded.rs index 0caf00c..767ec17 100644 --- a/src/sharded.rs +++ b/src/sharded.rs @@ -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(), } }