过滤并重试直到合格者通过，并且不对通过的项目进行重新测试

Question

给出代表某些外部可变资源的项目列表，如何过滤列表以仅发出某些项目并等待所有项目服从过滤器？

更具体地说：从文件列表中构建一个Flowable，该文件按存在筛选，仅允许存在的文件通过。如果不存在，请等待5秒钟以使文件存在。

这是我的第一次尝试：

Flowable.fromArray(new File("/tmp/file-1"), new File("/tmp/file-2"))
    .map(f -> {
        boolean exists = f.exists();
        System.out.println(f.getName() + " exists? " + exists);
        if(exists) {
            return f;
        } else {
            throw new RuntimeException(f.getName() + " doesn't exist");
        }
    })
    .retryWhen(ft -> {
        return ft.flatMap(error -> Flowable.timer(1, TimeUnit.SECONDS));
    })
    .blockingForEach(f -> System.out.println(f.getName() + " exists!"));

但是这给了：

file-1 exists? false
file-1 exists? false
file-1 exists? false
file-1 exists? false  ** $ touch /tmp/file-1 **
file-1 exists? true
file-2 exists? false
file-1 exists!
file-1 exists? true   ** BAD we are retesting! **
file-2 exists? false
file-1 exists!        ** BAD the subscriber got a duplicate event! **

即使我在distinct之后添加了retryWhen，文件仍然会重新测试。

因此，有一种方法仅是重新测试未通过先前测试的项目（不使用Observable外部的可变状态？]

Answer 1

对内部序列进行retryWhen并一起进行flatMap：