Flux.using 不发射项目答案

【问题标题】：Flux.using is not emiting itemsFlux.using 不发射项目
【发布时间】：2019-10-26 15:05:51
【问题描述】：

我正在尝试创建基于拉取的项目来源。出于某种原因，在推送一件商品后，没有任何东西到达消费者手中。整个管道卡住了。我错过了什么？

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);
        }
    });
}

【问题讨论】：

标签： java project-reactor

【解决方案1】：

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

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);
        }
    });
}

【讨论】：