java - Flux.using 不发射元素

Question

我正在尝试创建基于拉动的项目源。由于某种原因，在推出一件商品后，消费者没有收到任何东西。整个管道被卡住了。我缺少什么？

private static Flux<ByteBuffer> testRun(String path) {
    return Flux.using(() -> {
        FileInputStream in = new FileInputStream(path);
        FileChannel channel = in.getChannel();
        return Tuple.of(in, channel);
    }, t -> s -> {
        ByteBuffer bb = ByteBuffer.allocate(1000);
        try {
            if (t._2.read(bb) > 0)
                s.onNext(bb.rewind());
            else
                s.onComplete();
        } catch (IOException ex) {
            s.onError(ex);
        }
    }, t -> {
        try {
            t._1.close();
            t._2.close();
        } catch (IOException ex) {
            throw new RuntimeException(ex);
        }
    });
}

Answer 1

我明白了。 using 的 sourceSupplier 仅调用一次。

private static Flux<ByteBuffer> testRun(String path) {

    return Flux.using(() -> {
        final FileInputStream in = new FileInputStream(path);
        final FileChannel channel = in.getChannel();
        return Tuple.of(in, channel);
    }, t -> {
        return Flux.generate(s -> {
            ByteBuffer bb = ByteBuffer.allocate(1000);
            try {
                if (t._2.read(bb) > 0)
                    s.next(bb.rewind());
                else
                    s.complete();
            } catch (IOException ex) {
                s.error(ex);
            }
        });
    }, t -> {
        try {
            t._1.close();
            t._2.close();
        } catch (IOException ex) {
            throw new RuntimeException(ex);
        }
    });
}