RxJava using에서 FileInputStream 을 안정적으로 닫습니다.

Observable.using(
    { FileInputStream(File("hoge.txt")) }, // 1. Observableを流すためのresourceを生成
    { fileInputStream -> Observable.just(fileInputStream).flatMap { /* fileInputStreamを使った何かしらの処理 */ } }, // 2. 1を引数にとり、Observableを返す関数。1を使った一連の処理はここに記述する。
    { fileInputStream -> fileInputStream.close() } // 3. 1を引数に取る関数。2の処理が終了したり、onErrorに流れた場合に実行される。
)

fun <T> usingFileInputStream(file: File, mainStream: (FileInputStream) -> Observable<T>) = 
    Observable.using(
        { FileInputStream(file) },
        { mainStream(it) },
        { it.close }
    )

usingFileInputStream(File("hoge.txt")) { fileInputStream ->

   // fileInputStreamを使って何かする
   ...

}
.flatMap { ... }
.subscribe( ... )