rx-parallel-calls.kt

import io.reactivex.Observable
import io.reactivex.Single
import io.reactivex.observers.DisposableObserver
import io.reactivex.schedulers.Schedulers

fun main(args: Array<String>) {
  val list = listOf(3, 2, 1)
  val list2 = listOf(1f, 7f, 8f)

  Observable.fromIterable(list)
      .flatMap { value -> longRunningAsync(value.toString()).toObservable() }
      .mergeWith(Observable.fromIterable(list2)
            .flatMap { value2 -> longRunningAsync(value2.toString()).toObservable() }
      )
      .subscribe(object : DisposableObserver<String>() {
        override fun onComplete() {
          println("onComplete")
        }

        override fun onError(e: Throwable?) {
          println("onError $e")
        }

        override fun onNext(t: String?) {
          println("onNext $t")
        }

      })

  Thread.sleep(1_000_000L)
}

fun longRunningAsync(value: String): Single<String> {
  return Single.create<String>({
    emitter ->
      Thread.sleep(value.toFloat().toInt() * 100L)
      emitter.onSuccess("Result: $value")
  }).subscribeOn(Schedulers.io())
}