Список содержимого каталога с помощью Project Reactor и DirectoryStream

Question

Я хотел бы использовать DirectoryStream с Project Reactor для вывода списка всех файлов в каталоге.

Моя попытка:

Path myDir = Paths.get("C:\\Users\\r.dacanal\\Documents\\Reply\\EDA\\logging-consumer\\input");
DirectoryStream<Path> directoryStream = Files.newDirectoryStream(myDir);
Flux.fromIterable(directoryStream).doOnNext(s -> System.out.println(s)).subscribe();

Но я получаю следующее исключение:

Caused by: java.lang.IllegalStateException: Iterator already obtained
        at sun.nio.fs.WindowsDirectoryStream.iterator(WindowsDirectoryStream.java:117) ~[na:1.8.0_171]
        at reactor.core.publisher.FluxIterable.subscribe(FluxIterable.java:79) ~[reactor-core-3.3.4.RELEASE.jar:3.3.4.RELEASE]
        at reactor.core.publisher.Flux.subscribe(Flux.java:8264) ~[reactor-core-3.3.4.RELEASE.jar:3.3.4.RELEASE]
        at reactor.core.publisher.Flux.subscribeWith(Flux.java:8428) ~[reactor-core-3.3.4.RELEASE.jar:3.3.4.RELEASE]
        at reactor.core.publisher.Flux.subscribe(Flux.java:8235) ~[reactor-core-3.3.4.RELEASE.jar:3.3.4.RELEASE]
        at reactor.core.publisher.Flux.subscribe(Flux.java:8162) ~[reactor-core-3.3.4.RELEASE.jar:3.3.4.RELEASE]
        at reactor.core.publisher.Flux.subscribe(Flux.java:8080) ~[reactor-core-3.3.4.RELEASE.jar:3.3.4.RELEASE]
        at com.example.loggingconsumer.LoggingConsumerApplication.filterDirectory(LoggingConsumerApplication.java:46) ~[classes/:na]
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:1.8.0_171]
        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[na:1.8.0_171]
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:1.8.0_171]
        at java.lang.reflect.Method.invoke(Method.java:498) ~[na:1.8.0_171]
        at org.springframework.scheduling.support.ScheduledMethodRunnable.run(ScheduledMethodRunnable.java:84) ~[spring-context-5.2.5.RELEASE.jar:5.2.5.RELEASE]
        at org.springframework.scheduling.support.DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:54) ~[spring-context-5.2.5.RELEASE.jar:5.2.5.RELEASE]
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [na:1.8.0_171]
        at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308) [na:1.8.0_171]
        at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180) [na:1.8.0_171]
        at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294) [na:1.8.0_171]
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [na:1.8.0_171]
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [na:1.8.0_171]
        at java.lang.Thread.run(Thread.java:748) [na:1.8.0_171]

Может ли кто-нибудь привести пример кода использования DirectoryStream с Project Reactor?

Answer 1

Subscribe неоднократно вызывает iterator (), но DirectoryStream имеет проверку, чтобы убедиться, что итератор должен быть нулевым, прежде чем назначить его собственный

@Override
public Iterator<Path> iterator() {
    if (!isOpen) {
        throw new IllegalStateException("Directory stream is closed");
    }
    synchronized (this) {
        if (iterator != null)
            throw new IllegalStateException("Iterator already obtained");
        iterator = new WindowsDirectoryIterator(firstName);
        return iterator;
    }
}

Хотя, если вы используете Java 8+, есть 0 причин использовать newDirectoryStream(dir), так как вы можете использовать list(dir) для обеспечения фактического потока

Следующее должно работать

Path myDir = Paths.get("C:\\Users\\r.dacanal\\Documents\\Reply\\EDA\\logging-consumer\\input");
Stream<Path> directoryStream = Files.list(myDir);
Flux.fromStream(directoryStream).doOnNext(System.out::println).subscribe();