SubsribeOnTest

Main.kt

import io.reactivex.Observable
import io.reactivex.ObservableEmitter
import io.vertx.reactivex.core.RxHelper
import io.vertx.reactivex.core.Vertx
import java.util.concurrent.TimeUnit

object Main {

    val vertx = Vertx.vertx()
    val blockingScheduler by lazy { vertx.let(RxHelper::blockingScheduler) }
    val nonBlockingScheduler by lazy { vertx.let(RxHelper::scheduler) }

    val expensiveComputation = { it: ObservableEmitter<Int> ->
        printCurrentThread("expensiveComputation")
        var number = 0
        while(!it.isDisposed) {
            TimeUnit.SECONDS.sleep(1)
            it.onNext(number++)
        }
    }

    val printCurrentThread = { operationName : String ->
        println("[$operationName] " + Thread.currentThread().name)
    }

    @JvmStatic
    fun main(args: Array<String>) {

        Observable.create(expensiveComputation)
                .observeOn(nonBlockingScheduler)
                .filter { it % 2 == 0 }
                .doOnNext { printCurrentThread("onNext afterFilter") }
                .take(3)
                .doOnComplete { println("Done") }
                .subscribeOn(blockingScheduler)
                .subscribe { println("Received Number: " + it) }

    }

}

Output:

[expensiveComputation] vert.x-worker-thread-0
[onNext afterFilter] vert.x-eventloop-thread-0
Received Number: 0
[onNext afterFilter] vert.x-eventloop-thread-0
Received Number: 2
[onNext afterFilter] vert.x-eventloop-thread-0
Received Number: 4
Done