tweak RxScalaDemo (and Olympics example)

Author: samuelgruetter

language-adaptors/rxjava-scala/src/examples/scala/rx/lang/scala/examples/Olympics.scala

@@ -22,6 +22,7 @@ object Olympics {
   case class Medal(val year: Int, val games: String, val discipline: String, val medal: String, val athlete: String, val country: String)
 
   def mountainBikeMedals: Observable[Medal] = Observable.items(
+    duration(100 millis), // a short delay because medals are only awarded some time after the Games began
     Observable.items(
       Medal(1996, "Atlanta 1996", "cross-country men", "Gold", "Bart BRENTJENS", "Netherlands"),
       Medal(1996, "Atlanta 1996", "cross-country women", "Gold", "Paola PEZZO", "Italy"),
@@ -69,18 +70,33 @@ object Olympics {
   ).concat
 
   // speed it up :D
-  val fourYears = 4000.millis
+  val oneYear = 1000.millis
 
-  val neverUsedDummyMedal = Medal(3333, "?", "?", "?", "?", "?")
+  //val neverUsedDummyMedal = Medal(3333, "?", "?", "?", "?", "?")
 
-  def fourYearsEmpty: Observable[Medal] = {
+  /** runs an infinite loop, and returns Bottom type (Nothing) */
+  def getNothing: Nothing = {
+    println("You shouldn't have called this method ;-)")
+    getNothing
+  }
+  
+  /** returns an Observable which emits no elements and completes after a duration of d */
+  def duration(d: Duration): Observable[Nothing] = Observable.interval(d).take(1).filter(_ => false).map(_ => getNothing)
+  
+  def fourYearsEmpty: Observable[Medal] = duration(4*oneYear)
+
+  def yearTicks: Observable[Int] = 
+    (Observable.from(1996 to 2014) zip (Observable.items(-1) ++ Observable.interval(oneYear))).map(_._1)
+  
+  /*
+  def fourYearsEmptyOld: Observable[Medal] = {
     // TODO this should return an observable which emits nothing during fourYears and then completes
     // Because of https://github.com/Netflix/RxJava/issues/388, we get non-terminating tests
     // And this https://github.com/Netflix/RxJava/pull/289#issuecomment-24738668 also causes problems
     // So we don't use this:
-    // Observable.interval(fourYears).take(1).map(i => neverUsedDummyMedal).filter(m => false)
+    Observable.interval(fourYears).take(1).map(i => neverUsedDummyMedal).filter(m => false)
     // But we just return empty, which completes immediately
-    Observable.empty
-  }
+    // Observable.empty
+  }*/
 
 }

language-adaptors/rxjava-scala/src/examples/scala/rx/lang/scala/examples/RxScalaDemo.scala

@@ -32,7 +32,7 @@ import org.scalatest.junit.JUnitSuite
 import rx.lang.scala._
 import rx.lang.scala.schedulers._
 
-@Ignore // Since this doesn't do automatic testing, don't increase build time unnecessarily
+//@Ignore // Since this doesn't do automatic testing, don't increase build time unnecessarily
 class RxScalaDemo extends JUnitSuite {
 
   @Test def intervalExample() {
@@ -78,13 +78,9 @@ class RxScalaDemo extends JUnitSuite {
     val first = Observable.from(List(10, 11, 12))
     val second = Observable.from(List(10, 11, 12))
 
-    val b1 = (first zip second) map (p => p._1 == p._2) forall (b => b)
+    val b = (first zip second) forall { case (a, b) => a == b }
 
-    val equality = (a: Any, b: Any) => a == b
-    val b2 = (first zip second) map (p => equality(p._1, p._2)) forall (b => b)
-
-    assertTrue(b1.toBlockingObservable.single)
-    assertTrue(b2.toBlockingObservable.single)
+    assertTrue(b.toBlockingObservable.single)
   }
 
   @Test def testObservableComparisonWithForComprehension() {
@@ -93,7 +89,7 @@ class RxScalaDemo extends JUnitSuite {
 
     val booleans = for ((n1, n2) <- (first zip second)) yield (n1 == n2)
 
-    val b1 = booleans.forall(_ == true) // without `== true`, b1 is assigned the forall function
+    val b1 = booleans.forall(identity)
 
     assertTrue(b1.toBlockingObservable.single)
   }
@@ -216,18 +212,21 @@ class RxScalaDemo extends JUnitSuite {
     }).flatten.toBlockingObservable.foreach(println(_))
   }
 
-  @Ignore // TODO something's bad here
   @Test def timingTest1() {
     val numbersByModulo3 = Observable.interval(1000 millis).take(9).groupBy(_ % 3)
 
     val t0 = System.currentTimeMillis
 
     (for ((modulo, numbers) <- numbersByModulo3) yield {
       println("Observable for modulo" + modulo + " started at t = " + (System.currentTimeMillis - t0))
-      numbers.take(1) // <- TODO very unexpected
-      //numbers
+      numbers.map(n => s"${n} is in the modulo-$modulo group")
     }).flatten.toBlockingObservable.foreach(println(_))
   }
+  
+  @Test def testOlympicYearTicks() {
+    Olympics.yearTicks.subscribe(println(_))
+    waitFor(Olympics.yearTicks)
+  }
 
   @Test def groupByExample() {
     val medalsByCountry = Olympics.mountainBikeMedals.groupBy(medal => medal.country)
@@ -238,8 +237,10 @@ class RxScalaDemo extends JUnitSuite {
     firstMedalOfEachCountry.subscribe(medal => {
       println(s"${medal.country} wins its first medal in ${medal.year}")
     })
+    
+    Olympics.yearTicks.subscribe(year => println(s"\nYear $year starts."))
 
-    waitFor(firstMedalOfEachCountry)
+    waitFor(Olympics.yearTicks)
   }
 
   @Test def groupByUntilExample() {
@@ -469,7 +470,10 @@ class RxScalaDemo extends JUnitSuite {
 
   def output(s: String): Unit = println(s)
 
-  // blocks until obs has completed
+  /** Subscribes to obs and waits until obs has completed. Note that if you subscribe to
+   *  obs yourself and also call waitFor(obs), all side-effects of subscribing to obs
+   *  will happen twice.
+   */
   def waitFor[T](obs: Observable[T]): Unit = {
     obs.toBlockingObservable.toIterable.last
   }