创建Flux.fromIterable后如何运行on关闭操作？

Question

假设我们需要基于可关闭资源的内容创建通量。为了清楚起见，假设有一个BufferedReader要转换为通量

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

让我们假设iteratorOfLines生成一组有限的项。

我正在寻找一种关闭BufferedReader的方法，当Flux已消耗其中的所有数据或由于某种原因不需要剩余数据时（即订阅中止）。

有一个构造函数reactor.core.publisher.FLuxIterable（Iterable iterable， Runnable onClole），但是：

1. 似乎无法从Reactor的公共API获取（甚至是传递性的）

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

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

Answer 1

对于使用资源的等效方法，您可以使用

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