RxScala example

RxScalaBasedApproach.scala

import java.util.concurrent.{Executors, TimeUnit}

import rx.lang.scala.Observable

import scala.concurrent.duration._

//import scala.concurrent.ExecutionContext.Implicits.global
import rx.lang.scala.schedulers.{ComputationScheduler, ExecutionContextScheduler}

import scala.concurrent.{ExecutionContext, Future}

object RxScalaBasedApproach {

  def main(args: Array[String]): Unit = {
    val t0 = System.currentTimeMillis()

    def queryData1 = () => {
      println(s"thread - ${Thread.currentThread().getName}")
      Array(6, 7, 8, 9, 10)
    }

    def queryData2 = () => {
      println(s"thread - ${Thread.currentThread().getName}")
      Array(1, 2, 3, 4, 5)
    }

    val service = Executors.newFixedThreadPool(4)
    val ec = ExecutionContext.fromExecutor(service)
    val s = ExecutionContextScheduler(ec)

    val o1 = Observable.from(queryData1())
    val o2 = Observable.from(queryData2())

    Seq(o1, o2).foreach(_
        .subscribeOn(s)
        .subscribe(x => println(s"thread - ${Thread.currentThread().getName}, value - $x"))
    )
    service.shutdown()
    service.awaitTermination(10, TimeUnit.SECONDS)

    val t1 = System.currentTimeMillis()
    println(s"Elapsed time:${(t1 - t0) / 1000f}")
  }

}