Extension Function for InputStream that convert it to Observable<String>

InputStreamObservable.kt

fun InputStream.toObservable(): Observable<String> {

    class IOThread(private val inputStream: InputStream,
                   private val subscriber: Subscriber<in String>) : Thread() {
        override fun run(): Unit {
            Scanner(InputStreamReader(inputStream)).use {
                while (it.hasNextLine()) {
                    subscriber.onNext(it.nextLine())
                }
                subscriber.onCompleted()
            }
        }
    }

    return observable { subscriber ->
        IOThread(this, subscriber).start()
    }
}

Main.kt

fun main(args: Array<String>) {
    val processBuilder = ProcessBuilder("ping", "-c", "10", "www.google.com").apply {
        directory(File("/Users/jasoet"))
    }

    val process = processBuilder.start()
    val processInputStream = process.inputStream
    val observableEvent = processInputStream.toObservable()
    observableEvent
        .doOnCompleted {
            println("Finally Completed ${processInputStream.available()}")
        }
        .forEach { println("Observe => $it") }
    process.waitFor()

}