RecursiveDirList.java

import java.io.File;
import java.util.*;
import java.util.concurrent.*;

import rx.*;
import rx.Observable;
import rx.schedulers.Schedulers;


public class RecursiveDirList {
    public static Observable<File[]> getFiles(File parent) {
        return Observable.create(s -> {
            System.out.println(Thread.currentThread() + ": " + parent);
            s.onNext(parent.listFiles());
            s.onCompleted();
        });
    }
    public static Observable<File> getAllFiles(File root) {
        Observable<File[]> ofs = getFiles(root);
        return ofs.flatMap(fs -> {
            Observable<File> result = Observable.empty();
            List<File> regularFiles = new ArrayList<>();
            for (File f : fs) {
                if (f.isDirectory()) {
                    result = result.mergeWith(getAllFiles(f));
                } else {
                    regularFiles.add(f);
                }
            }
            
            return result.mergeWith(Observable.from(regularFiles));
        });
    }
    public static void main(String[] args) throws Exception {
        ExecutorService exec = Executors.newFixedThreadPool(1);
        try {
            Scheduler single = Schedulers.from(exec);

            CountDownLatch cdl = new CountDownLatch(1);
            
            getAllFiles(new File("."))
            .subscribeOn(single)
            .observeOn(Schedulers.trampoline())
            .subscribe(System.out::println, Throwable::printStackTrace, () -> { System.out.println("---"); cdl.countDown(); });
            
            cdl.await();
            
        } finally {
            exec.shutdown();
        }
    }
}