RxMultiThreadCheck.kt

package com.schrmnnj.sample

import rx.Observable
import rx.schedulers.Schedulers

/**
 * Created on 08.01.16.
 */

class RxMultiThreadCheck {

    @Suppress("PLATFORM_CLASS_MAPPED_TO_KOTLIN")
    fun check(): Unit {
        val monitor: Object = Object();

        getStrings()
                .flatMap({ Observable.from(it) })
                .subscribeOn(Schedulers.io())
                .doOnNext({ printElementAndThread(it) })
                .map({ it.toUpperCase() })
                .observeOn(Schedulers.computation())
                .doOnNext({ printElementAndThread(it) })
                .map({ it.toLowerCase() })
                .observeOn(Schedulers.newThread())
                .subscribe({
                    printElementAndThread(it)
                }, {
                    println(it.message)
                }, {
                    synchronized (monitor) {
                        monitor.notifyAll();
                    }
                })

        synchronized (monitor) {
            monitor.wait();
        }
    }

    companion object {
        fun getStrings(): Observable<List<String>> {
            return Observable.range(0, 100).map({ "String $it" }).toList();
        }

        fun printElementAndThread(s: String): Unit {
            println("$s : ${Thread.currentThread().name}");
        }
    }
}

fun main(args: Array<String>) {
    RxMultiThreadCheck().check()
}