feat(jni): onMetadataFetch, stop=shutdown+release, DhtCrawler OO API

- JNI: register on_metadata_fetch -> DhtListener.onMetadataFetch - stopServer only: shutdown + destroy (remove destroyServer) - Java: DhtCrawler.createServer/start/stop, package-private DhtCrawlerJni - gitignore examples-jni build/.gradle Made-with: Cursor
2026-03-13 19:48:46 +08:00
parent 2746791205
commit e4740f5fff
8 changed files with 207 additions and 129 deletions
@@ -18,3 +18,7 @@ torrents/

 # 个人脚本（不提交到仓库）
 scripts/
+
+# Java / Gradle（examples-jni）
+examples-jni/build/
+examples-jni/.gradle/
@@ -6,16 +6,14 @@

 ```
 examples-jni/
-├── build.gradle                        # Gradle 构建脚本
+├── build.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          # 可运行示例
+    ├── model/          # DHTOptions, TorrentInfo, FileInfo
+    ├── DhtCrawler.java      # 面向对象入口（推荐）
+    ├── DhtCrawlerJni.java # 包内 native 绑定
+    ├── DhtListener.java
+    └── DhtCrawlerExample.java
 ```

 ## 快速开始
@@ -33,8 +31,8 @@ examples-jni/
 # 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
+# Windows（PowerShell/CMD 需对 -D 参数加引号，否则会报找不到主类）
+java "-Djava.library.path=." -jar dht-crawler-jni-example-<version>.jar
 ```

 ### 方式二：从源码编译并运行
@@ -44,7 +42,6 @@ java -Djava.library.path=. -jar dht-crawler-jni-example-<version>.jar
 在仓库**根目录**执行：

 ```bash
-# 本机（Linux 产出 libdht_crawler.so，Windows 产出 dht_crawler.dll，macOS 产出 libdht_crawler.dylib）
 cargo build --release --features jni
 ```

@@ -55,10 +52,7 @@ cargo build --release --features jni
 在本目录（`examples-jni/`）执行：

 ```bash
-# 使用默认库路径（../target/release）
 gradle run
-
-# 自定义库路径
 gradle run -Plib.path=/path/to/your/lib
 ```

@@ -66,28 +60,28 @@ gradle run -Plib.path=/path/to/your/lib

 ```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=<路径>`。
+1. 复制 `cn/lmcw/dht/` 下源码（含 `model/`、`DhtCrawler`、`DhtCrawlerJni`、`DhtListener`）。
+2. 将对应平台的 so/dll/dylib 放入 `java.library.path`。

-## JNI 生命周期
+## API（面向对象）

 ```java
-DHTOptions options = new DHTOptions().setPort(6881);
-long handle = DhtCrawlerJni.createServer(options, listener);
-DhtCrawlerJni.startServer(handle);
-// ... 运行 ...
-DhtCrawlerJni.stopServer(handle);
-DhtCrawlerJni.destroyServer(handle); // 必须调用，释放 Rust 资源
+DhtCrawler crawler = DhtCrawler.createServer(options, listener);
+crawler.start();
+// ...
+crawler.stop();   // 或 try-with-resources
 ```

+- **`DhtCrawler.createServer(options, listener)`**：创建会话（未启动 DHT；Java 不能用方法名 `new`，故不用 `open`）。
+- **`start()`**：后台启动 DHT，非阻塞；同一会话多次 `start()` 仅首次生效。
+- **`stop()` / `close()`**：停止并释放 Rust 资源，幂等。
+- **`getNodePoolSize()`**：routing table 节点数。
+
 ## 注意事项

- `destroyServer` 必须在不再使用后调用，否则 Rust 侧的 tokio runtime 和连接不会释放。
- 回调方法（`onTorrent`、`onError`）在 Rust 的 tokio 工作线程中触发，请确保实现是线程安全的。
- 同一个句柄不要在多个线程中并发调用 `destroyServer`。
+- 回调在 Rust 工作线程触发，实现需线程安全。
+- `onMetadataFetch` 在阻塞线程池调用，宜快速返回。
@@ -0,0 +1,100 @@
+package cn.lmcw.dht;
+
+import cn.lmcw.dht.model.DHTOptions;
+
+/**
+ * DHT 爬虫会话：封装 native 句柄与生命周期，推荐使用的面向对象入口。
+ *
+ * <pre>{@code
+ * try (DhtCrawler crawler = DhtCrawler.createServer(options, listener)) {
+ *     crawler.start();
+ *     // ... 运行 ...
+ * } // close → stop，释放 Rust 资源
+ *
+ * // 或手动：
+ * DhtCrawler c = DhtCrawler.createServer(options, listener);
+ * c.start();
+ * // ...
+ * c.stop();
+ * }</pre>
+ */
+public final class DhtCrawler implements AutoCloseable {
+
+    private long handle;
+    private volatile boolean started;
+
+    private DhtCrawler(long handle) {
+        this.handle = handle;
+    }
+
+    /**
+     * 创建 DHT 服务器会话（尚未 {@link #start()}，与 native createServer 对应）。
+     * Java 中方法不能命名为 {@code new}，故用 {@code createServer}。
+     *
+     * @param options  配置，{@code null} 使用 Rust 默认
+     * @param listener 回调，{@code null} 不注册回调
+     * @return 已绑定 native 资源的会话
+     * @throws IllegalStateException 创建失败（句柄为 0，通常伴随 JVM 异常）
+     */
+    public static DhtCrawler createServer(DHTOptions options, DhtListener listener) {
+        long h = DhtCrawlerJni.createServer(options, listener);
+        if (h == 0) {
+            throw new IllegalStateException("DHT 服务器创建失败（createServer 返回 0）");
+        }
+        return new DhtCrawler(h);
+    }
+
+    /**
+     * 在后台启动 DHT（非阻塞）。可多次调用，仅首次生效。
+     *
+     * @throws IllegalStateException 已 {@link #stop()} 关闭后不能再启动
+     */
+    public synchronized void start() {
+        if (handle == 0) {
+            throw new IllegalStateException("会话已关闭，无法 start");
+        }
+        if (started) {
+            return;
+        }
+        DhtCrawlerJni.startServer(handle);
+        started = true;
+    }
+
+    /**
+     * 停止并释放全部 Rust 资源（tokio runtime 等）。幂等：重复调用安全。
+     */
+    public synchronized void stop() {
+        if (handle == 0) {
+            return;
+        }
+        DhtCrawlerJni.stopServer(handle);
+        handle = 0;
+        started = false;
+    }
+
+    /** 是否已调用过 {@link #start()} 且尚未 {@link #stop()}。 */
+    public synchronized boolean isStarted() {
+        return started && handle != 0;
+    }
+
+    /** 会话仍有效（未 stop）时为 true。 */
+    public synchronized boolean isOpen() {
+        return handle != 0;
+    }
+
+    /**
+     * 当前 routing table 节点数；已关闭时返回 0。
+     */
+    public int getNodePoolSize() {
+        long h = handle;
+        if (h == 0) {
+            return 0;
+        }
+        return DhtCrawlerJni.getNodePoolSize(h);
+    }
+
+    @Override
+    public void close() {
+        stop();
+    }
+}
@@ -24,10 +24,9 @@ import java.util.concurrent.atomic.AtomicLong;
 public class DhtCrawlerExample {

    public static void main(String[] args) throws InterruptedException {
-        // 1. 构造配置
        DHTOptions options = new DHTOptions()
                .setPort(6881)
-                .setNetMode(0)              // 0 = IPv4 Only
+                .setNetMode(0)
                .setMetadataTimeout(5L)
                .setMaxMetadataQueueSize(50_000)
                .setMaxMetadataWorkerCount(500);
@@ -35,7 +34,6 @@ public class DhtCrawlerExample {
        AtomicLong torrentCount = new AtomicLong();
        CountDownLatch shutdown = new CountDownLatch(1);

-        // 2. 实现回调
        DhtListener listener = new DhtListener() {
            @Override
            public void onTorrent(TorrentInfo info) {
@@ -49,34 +47,29 @@ public class DhtCrawlerExample {
            public void onError(String message) {
                System.err.println("[ERROR] " + message);
            }
+
+            @Override
+            public boolean onMetadataFetch(String infoHash) {
+                return true;
+            }
        };

-        // 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());
+        final DhtCrawler crawler = DhtCrawler.createServer(options, listener);
+        crawler.start();
+        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());
+            crawler.stop();
+            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());
+            System.out.println("节点池: " + crawler.getNodePoolSize()
+                    + "  已发现种子: " + torrentCount.get());
        }
    }
 }
@@ -3,21 +3,9 @@ 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>
+ * 包内 JNI 绑定（由 {@link DhtCrawler} 使用）。业务代码请用 {@link DhtCrawler}。
 */
-public final class DhtCrawlerJni {
+final class DhtCrawlerJni {

    static {
        System.loadLibrary("dht_crawler");
@@ -25,44 +13,11 @@ public final class DhtCrawlerJni {

    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);
+    static native long createServer(DHTOptions options, DhtListener listener);

-    /**
-     * 启动已创建的 DHT 服务器（非阻塞）。
-     * 服务器在 Rust tokio runtime 后台运行，此方法立即返回。
-     *
-     * @param handle {@link #createServer} 返回的句柄
-     */
-    public static native void startServer(long handle);
+    static native void startServer(long handle);

-    /**
-     * 向服务器发送停止信号（非阻塞）。
-     * 调用后服务器将优雅退出，但资源尚未释放，需继续调用 {@link #destroyServer}。
-     *
-     * @param handle {@link #createServer} 返回的句柄
-     */
-    public static native void stopServer(long handle);
+    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);
+    static native int getNodePoolSize(long handle);
 }
@@ -22,4 +22,17 @@ public interface DhtListener {
     * @param message Rust 侧错误的文本描述
     */
    void onError(String message);
+
+    /**
+     * 在即将对某 info_hash 拉取 BitTorrent metadata 之前调用。
+     * 与 Rust {@code DHTServer::on_metadata_fetch} 对应：返回 {@code true} 才会真正去拉取；
+     * 返回 {@code false} 则跳过该 hash，不会触发 {@link #onTorrent}。
+     * <p>在 Rust 的 metadata 工作线程中通过阻塞线程池调用，应快速返回，避免长时间阻塞。</p>
+     *
+     * @param infoHash 40 字符小写十六进制 info_hash
+     * @return 是否允许拉取 metadata
+     */
+    default boolean onMetadataFetch(String infoHash) {
+        return true;
+    }
 }
@@ -4,12 +4,13 @@ use crate::jni_bindings::types::torrent_info_to_java;
 use jni::objects::JValue;
 use std::sync::Arc;

-/// 向 `DHTServer` 注册所有 Java 回调（on_torrent、on_error）。
+/// 向 `DHTServer` 注册所有 Java 回调（on_torrent、on_error、on_metadata_fetch）。
 ///
 /// `callback` 持有 Java listener 的全局引用，可跨线程安全使用。
 pub fn register_callbacks(server: &Arc<DHTServer>, callback: JavaCallback) {
    register_on_torrent(server, callback.clone());
-    register_on_error(server, callback);
+    register_on_error(server, callback.clone());
+    register_on_metadata_fetch(server, callback);
 }

 /// 注册 on_torrent 回调：Rust TorrentInfo → Java listener.onTorrent(TorrentInfo)
@@ -54,3 +55,37 @@ fn register_on_error(server: &Arc<DHTServer>, callback: JavaCallback) {
        }
    });
 }
+
+/// 注册 on_metadata_fetch：在拉取 metadata 前询问 Java `onMetadataFetch(infoHash)`。
+/// 返回 `true` 才继续拉取；在阻塞线程池中执行 JNI，避免阻塞 tokio worker。
+fn register_on_metadata_fetch(server: &Arc<DHTServer>, callback: JavaCallback) {
+    server.on_metadata_fetch(move |info_hash| {
+        let cb = callback.clone();
+        async move {
+            let r = tokio::task::spawn_blocking(move || {
+                cb.with_env(|env, listener| {
+                    let j_s = env.new_string(&info_hash)?;
+                    let out = env.call_method(
+                        listener,
+                        "onMetadataFetch",
+                        "(Ljava/lang/String;)Z",
+                        &[JValue::Object(&j_s.into())],
+                    )?;
+                    Ok(out.z()?)
+                })
+            })
+            .await;
+            match r {
+                Ok(Ok(allow)) => allow,
+                Ok(Err(e)) => {
+                    log::error!("回调 onMetadataFetch 失败: {e}，默认允许拉取");
+                    true
+                }
+                Err(e) => {
+                    log::error!("onMetadataFetch spawn_blocking 失败: {e}，默认允许拉取");
+                    true
+                }
+            }
+        }
+    });
+}
@@ -83,7 +83,8 @@ pub extern "system" fn Java_cn_lmcw_dht_DhtCrawlerJni_startServer(
    });
 }

-/// 停止 DHTServer（发送关闭信号，不释放资源）。
+/// 停止并销毁 DHTServer：发关闭信号后释放 tokio runtime 等全部 Rust 资源。
+/// 调用后句柄失效，勿再传入任何 native 方法。
 ///
 /// Java 签名：`native void stopServer(long handle);`
 #[unsafe(no_mangle)]
@@ -92,39 +93,22 @@ pub extern "system" fn Java_cn_lmcw_dht_DhtCrawlerJni_stopServer(
    _class: JClass,
    handle: jlong,
 ) {
-    crate::jni_catch!(&mut env, (), {
-        let h = unsafe {
+    jni_shutdown_and_release(&mut env, handle);
+}
+
+fn jni_shutdown_and_release(env: &mut JNIEnv, handle: jlong) {
+    crate::jni_catch!(env, (), {
+        if handle == 0 {
+            return;
+        }
+        unsafe {
            match handle_ref(handle) {
-                Some(h) => h,
+                Some(h) => h.stop(),
                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);
        }
    });