Experimentation with Rx

Main.kt

import rx.Observable
import java.io.File

/** Basic data structure to add information to a file reference. */
data class FileInfo(val file: File, var bytes: Long = 0, var filesCount: Long = 1)

/** Main function. */
fun main(args: Array<String>) {
    Observable.just(File("/Users/joan/Dropbox"))
            .flatMap({ fileToFileInfo(it) })
            .subscribe(
                    { fileInfo -> println(fileInfo) },
                    { throwable -> throwable.printStackTrace() })
}

/** Recursive function to compute additional information on a file. */
fun fileToFileInfo(file: File): Observable<FileInfo> {
    if (file.isFile()) {
        return Observable.just(FileInfo(file, file.length()))
    } else if (file.isDirectory()) {
        return Observable.from(file.listFiles())
                .flatMap({ fileToFileInfo(it) })
                .reduce(FileInfo(file), { acc, b ->
                    acc.bytes += b.bytes
                    acc.filesCount += b.filesCount
                    acc
                })
    }
    return Observable.error(IllegalArgumentException("$file is neither a file or a directory"))
}