过滤并重试，直到筛选器通过，不重新测试通过的项目

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! **

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

那么，有没有办法只重新测试那些未通过前一次测试的项(而不使用Observable之外的可变状态)？

Answer 1

在内部序列上执行retryWhen，并将它们flatMap在一起：

source
.flatMap(file -> 
    Observable.fromCallable(() -> {
        if (file.exists()) {
            return file;
        }
        throw new RuntimeException();
    })
    .retryWhen(errors -> errors.delay(1, TimeUnit.SECONDS))
)
.blockingForEach(f -> System.out.println(f.getName() + " exists!"));