RxJavaでファイルツリーをつくる

RxJavaの使い方で、再帰的に何かしたい事があるかと思うので、
単純な例としてファイルツリーを作成するものを、RxJavaで書くとどうなるか書いてみました。

RxJava2系なので、Flowableを使ってますが、Observableに読み替えればOKです。

// 再帰呼出しするためのメソッド
static Flowable<String> loadFileStream(String[] paths) {

    // String配列からストリームを作成
    return Flowable.fromArray(paths)
            .flatMap(p -> {
                File f = new File(p);

                // ディレクトリでない場合は、自身のファイルPATHだけをストリームに流す
                Flowable<String> parentStream = Flowable.just(f.getPath());
                if (!f.isDirectory()) {
                    return parentStream;
                }

                // ディレクトリ内にほかのファイルが見つからない場合も、自身のファイルPATHだけをストリームに流す
                String[] children = f.list();
                if (null == children) {
                    return parentStream;
                }

                // ディレクトリ内のファイルに親PATHを結合して文字列生成
                String[] list = Stream.of(children)
                        .map(o -> Paths.get(f.getPath(), o).toString())
                        .toArray(String[]::new);

                // 再帰呼出し
                // (自身のファイルPATHも含めてストリームに流す)
                return Flowable.concat(parentStream, loadFileStream(list));
            });
}

public static void main(String[] args) {

    // 起点となるPATHを渡してストームを生成
    Flowable<String> stream = loadFileStream(new String[]{"."});

    // subscribeすると、ストリームにデータが流れてくる。
    stream.subscribe(
            System.out::println,
            Throwable::printStackTrace,
            () -> System.out.println("complete!"));
}

flatMapの中で、再度Flowableを生成するように、 loadFileStreamを再帰呼出しすれば良いだけです。
更に作成したFlowableにsubscribeOnでスケジューラー設定をしてやれば、
ディレクトリ階層ごとにマルチスレッド処理で動作させる事も簡単です。

余談 Groovyで書くと

ちなみに敢えてGroovyで書くとこんな感じになります。

Flowable<String> loadFileStream(List<String> paths) {
    return Flowable.fromIterable(paths).
            flatMap({ p ->
                File f = new File(p)

                Flowable<String> parentStream = Flowable.just(f.path)
                if (!f.directory) {
                    return parentStream
                }

                List<String> children = f.list().collect({ Paths.get(f.path, it).toString() })
                if (!children) {
                    return parentStream
                }

                return Flowable.concat(parentStream, loadFileStream(children))
            })
}


Flowable<String> stream = loadFileStream(["."])
stream.subscribe(
        { println(it) },
        { it.printStackTrace() },
        { println("complete!") })

出力結果

こんな感じに出力されますよ。

./src
./src/main
./src/main/groovy
./src/main/groovy/sample1.groovy
./src/main/groovy/sample2.groovy
./src/main/groovy/sample3.groovy
./src/main/groovy/sample4.groovy
./src/main/groovy/sample5.groovy
./src/main/groovy/sample6.groovy
./src/main/groovy/sample8.groovy
./src/main/java
./src/main/java/Sample7.java
./src/script
./src/script/groovy
./src/script/groovy/test.groovy