Фильтрация и повторная попытка, пока установщик не пройдет, пропущенные элементы не будут повторно проверены - PullRequest
Новая
       15

Фильтрация и повторная попытка, пока установщик не пройдет, пропущенные элементы не будут повторно проверены

0 голосов
/ 23 января 2020

При наличии списка элементов, представляющих некоторый внешне изменяемый ресурс, как я могу отфильтровать список, чтобы он излучал только определенные элементы, и подождать, пока все элементы подчинятся фильтру?

Более конкретно: из списка файлов, построить 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)

1 Ответ

1 голос
/ 23 января 2020

Сделайте 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!"));
...