atc atc - 2 months ago 9
Scala Question

RxScala Observable never runs

With the following

build.sbt
:

name := "blah"

version := "1.0"

scalaVersion := "2.11.6"

libraryDependencies ++= Seq("io.reactivex" % "rxscala_2.11" % "0.24.1", "org.scalaj" %% "scalaj-http" % "1.1.4")


and this code:

import rx.lang.scala.Observable
import scala.concurrent.duration._
import scala.language.postfixOps

object Main {

def main(args: Array[String]): Unit = {
println("Ready?")
val o = Observable.interval(200 millis).take(5)
o.subscribe(n => println(s"n = ${n}"))
}

}


When I run it, all that's printed is
Ready?
; I see no
n = ...
at all.

I run using
sbt run
; it's built using Scala 2.6.11 and RxScala 0.24.1, as well as sbt 0.13. Any ideas?

Answer

The problem is that your program exits before o fires. Try the following code:

import rx.lang.scala.Observable
import scala.concurrent.duration._
import scala.language.postfixOps
object Main {

  def main(args: Array[String]): Unit = {
    println("Ready?")
    val o = Observable.interval(200 millis).take(5)
    o.subscribe(n => println(s"n = ${n}"))

    Thread.sleep(5000)
  }

}

Alternatively you can replace Thread.sleep with o.toBlocking.last, which cannot return before o terminates.

Comments