java - 创建 Flux.fromIterable 后如何运行 onClose 操作？

Question

假设我们需要创建一个 Flux基于 Closeable 的内容资源。
为了清楚起见，说有一个 BufferedReader转换为 Flux<String> .

BufferedReader reader = createReader("my_resource_path");
Flux<String> flux = Flux.fromIterable(() -> iteratorOfLines(reader));

让我们假设 iteratorOfLines产生有限的项目集。

我正在寻找关闭 BufferedReader 的方法当 Flux已使用其中的所有数据或由于某种原因不需要剩余数据(即订阅被中止)。

有一个构造函数 reactor.core.publisher.FluxIterable(Iterable iterable, Runnable onClose) ， 但:

似乎无法从 react 器

的公共(public) API 访问(甚至可传递)

我怀疑它是否有帮助，因为它不包括 Flux 在获取可迭代的最后一项之前停止的情况。

Flux.fromIterable 之后清理/关闭资源的正确方法是什么？发布最后一项？

可能有比 fromIterable 更好的方法做类似的事情，所以欢迎所有选择。

Answer 1

对于尝试使用资源的等效项，您可以使用 using

    Flux.using(
            //Set up resource
            () -> createReader("my_resource_path"),
            //Create flux from resource
            reader -> Flux.fromIterable(iteratorOfLines(reader)),
            //Perform action (cleanup/close) 
            //when resource completes/errors/cancelled
            reader -> {
                try{
                    reader.close();
                }catch(IOException e){
                    throw Exceptions.propagate(e);
                }
            }
    );