RxScalaDemo.scala

/**
 * Copyright 2013 Netflix, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package examples

import java.io.IOException
import java.util.concurrent.CountDownLatch
import java.util.concurrent.Executors
import java.util.concurrent.TimeUnit

import rx.lang.scala.subjects.SerializedSubject

import scala.concurrent.Await
import scala.collection.immutable.HashMap
import scala.collection.mutable
import scala.concurrent.duration.Duration
import scala.concurrent.duration.DurationInt
import scala.concurrent.duration.DurationLong
import scala.concurrent.ExecutionContext
import scala.language.postfixOps
import scala.language.implicitConversions
import org.junit.Assert.assertEquals
import org.junit.Assert.assertTrue
import org.junit.Assert.assertFalse
import org.junit.Ignore
import org.junit.Test
import org.scalatestplus.junit.JUnitSuite
import rx.lang.scala._
import rx.lang.scala.observables.{AsyncOnSubscribe, SyncOnSubscribe}
import rx.lang.scala.schedulers._

/**
 * Demo how the different operators can be used. In Eclipse, you can right-click
 * a test and choose "Run As" > "Scala JUnit Test".
 * 
 * For each operator added to Observable.java, we add a little usage demo here.
 * It does not need to test the functionality (that's already done by the tests in
 * RxJava core), but it should demonstrate how it can be used, to make sure that
 * the method signature makes sense.
 */
@Ignore // Since this doesn't do automatic testing, don't increase build time unnecessarily
class RxScalaDemo extends JUnitSuite {

  @Test def subscribeExample(): Unit = {
    val o = Observable.just(1, 2, 3)

    // Generally, we have two methods, `subscribe` and `foreach`, to listen to the messages from an Observable.
    // `foreach` is just an alias to `subscribe`.
    o.subscribe(
      n => println(n),
      e => e.printStackTrace(),
      () => println("done")
    )

    o.foreach(
      n => println(n),
      e => e.printStackTrace(),
      () => println("done")
    )

    // For-comprehension is also an alternative, if you are only interested in `onNext`
    for (i <- o) {
      println(i)
    }
  }

  @Test def intervalExample(): Unit = {
    val o = Observable.interval(200 millis).take(5)
    o.subscribe(n => println("n = " + n))

    // need to wait here because otherwise JUnit kills the thread created by interval()
    waitFor(o)

    println("done")
  }

  def msTicks(start: Long, step: Long): Observable[Long] = {
    // will be easier once we have Observable.generate method
    Observable.interval(step millis) map (_ * step + start)
  }

  def prefixedTicks(start: Long, step: Long, prefix: String): Observable[String] = {
    msTicks(start, step).map(prefix + _)
  }

  @Test def testTicks(): Unit = {
    val o = prefixedTicks(5000, 500, "t = ").take(5)
    o.subscribe(output(_))
    waitFor(o)
  }

  @Test def testSwitch(): Unit = {
    // We do not have ultimate precision: Sometimes, 747 gets through, sometimes not
    val o = Observable.interval(1000 millis).map(n => prefixedTicks(0, 249, s"Observable#$n: "))
      .switch.take(16)
    o.subscribe(output(_))
    waitFor(o)
  }

  @Test def testSwitchOnObservableOfInt(): Unit = {
    // Correctly rejected with error
    // "Cannot prove that Observable[Int] <:< Observable[Observable[U]]"
    // val o = Observable(1, 2).switch
  }

  @Test def testObservableComparison(): Unit = {
    val first = Observable.from(List(10, 11, 12))
    val second = Observable.from(List(10, 11, 12))

    val b = (first zip second) forall { case (a, b) => a == b }

    assertTrue(b.toBlocking.single)
  }

  @Test def testObservableComparisonWithForComprehension(): Unit = {
    val first = Observable.from(List(10, 11, 12))
    val second = Observable.from(List(10, 11, 12))

    val booleans = for ((n1, n2) <- (first zip second)) yield (n1 == n2)

    val b1 = booleans.forall(identity)

    assertTrue(b1.toBlocking.single)
  }

  @Test def testStartWithIsUnnecessary(): Unit = {
    val before = List(-2, -1, 0).toObservable
    val source = List(1, 2, 3).toObservable
    println((before ++ source).toBlocking.toList)
  }

  @Test def mergeTwoExample(): Unit = {
    val slowNumbers = Observable.interval(400 millis).take(5).map("slow " + _)
    val fastNumbers = Observable.interval(200 millis).take(10).map("fast " + _)
    val o = (slowNumbers merge fastNumbers)
    o.subscribe(output(_))
    waitFor(o)
  }

  def myInterval(period: Long): Observable[String] = {
    Observable.interval(period.millis).map(n => s"Obs-$period emits $n")
  }

  @Test def flattenManyExample(): Unit = {
    val o = Observable.interval(500 millis).map(n => myInterval((n+1)*100))
    val stopper = Observable.interval(5 seconds)
    o.flatten.takeUntil(stopper).toBlocking.foreach(println(_))
  }

  @Test def flattenSomeExample(): Unit = {
    // To merge some observables which are all known already:
    List(
      Observable.interval(200 millis),
      Observable.interval(400 millis),
      Observable.interval(800 millis)
    ).toObservable.flatten.take(12).toBlocking.foreach(println(_))
  }

  @Test def flattenExample(): Unit = {
    List(
      Observable.interval(200 millis).map(_ => 1).take(5),
      Observable.interval(200 millis).map(_ => 2).take(5),
      Observable.interval(200 millis).map(_ => 3).take(5),
      Observable.interval(200 millis).map(_ => 4).take(5)
    ).toObservable.flatten.toBlocking.foreach(println(_))
  }

  @Test def flattenExample2(): Unit = {
    List(
      Observable.interval(200 millis).map(_ => 1).take(5),
      Observable.interval(200 millis).map(_ => 2).take(5),
      Observable.interval(200 millis).map(_ => 3).take(5),
      Observable.interval(200 millis).map(_ => 4).take(5)
    ).toObservable.flatten(2).toBlocking.foreach(println(_))
  }

  @Test def tumblingBufferExample(): Unit = {
    val o = Observable.from(1 to 18)
    o.tumblingBuffer(5).subscribe((l: Seq[Int]) => println(l.mkString("[", ", ", "]")))
  }

  @Test def tumblingBufferExample2(): Unit = {
    val o = Observable.from(1 to 18).zip(Observable.interval(100 millis)).map(_._1)
    val boundary = Observable.interval(500 millis)
    o.tumblingBuffer(boundary).toBlocking.foreach((l: Seq[Int]) => println(l.mkString("[", ", ", "]")))
  }

  @Test def slidingBufferExample(): Unit = {
    val o = Observable.from(1 to 18).slidingBuffer(4, 2)
    o.subscribe(println(_))
  }

  @Test def slidingBufferExample2(): Unit = {
    val open = Observable.interval(300 millis)
    val closing = Observable.interval(600 millis)
    val o = Observable.interval(100 millis).take(20).slidingBuffer(open)(_ => closing)
    o.zipWithIndex.toBlocking.foreach {
      case (seq, i) => println(s"Observable#$i emits $seq")
    }
  }

  @Test def slidingBufferExample3(): Unit = {
    val o = Observable.from(1 to 18).zip(Observable.interval(100 millis)).map(_._1)
    o.slidingBuffer(500 millis, 200 millis).toBlocking.foreach((l: Seq[Int]) => println(l.mkString("[", ", ", "]")))
  }

  @Test def tumblingExample(): Unit = {
    (for ((o, i) <- Observable.from(1 to 18).tumbling(5).zipWithIndex; n <- o)
      yield s"Observable#$i emits $n"
    ).subscribe(output(_))
  }

  @Test def tumblingExample2(): Unit = {
    val windowObservable = Observable.interval(500 millis)
    val o = Observable.from(1 to 20).zip(Observable.interval(100 millis)).map(_._1)
    (for ((o, i) <- o.tumbling(windowObservable).zipWithIndex; n <- o)
      yield s"Observable#$i emits $n"
    ).toBlocking.foreach(println)
  }

  @Test def slidingExample(): Unit = {
    val o = Observable.from(1 to 18).sliding(4, 2)
    (for ((o, i) <- o.zipWithIndex;
          n <- o)
      yield s"Observable#$i emits $n"
    ).toBlocking.foreach(println)
  }

  @Test def slidingExample2(): Unit = {
    val o = Observable.interval(100 millis).take(20).sliding(500 millis, 200 millis)
    (for ((o, i) <- o.zipWithIndex;
          n <- o)
      yield s"Observable#$i emits $n"
    ).toBlocking.foreach(println)
  }

  @Test def slidingExample3(): Unit = {
    val open = Observable.interval(300 millis)
    val closing = Observable.interval(600 millis)
    val o = Observable.interval(100 millis).take(20).sliding(open)(_ => closing)
    (for ((o, i) <- o.zipWithIndex;
          n <- o)
      yield s"Observable#$i emits $n"
    ).toBlocking.foreach(println)
  }

  @Test def testReduce(): Unit = {
    assertEquals(10, List(1, 2, 3, 4).toObservable.reduce(_ + _).toBlocking.single)
  }

  @Test def testForeach(): Unit = {
    val numbers = Observable.interval(200 millis).take(3)

    // foreach is not available on normal Observables:
    // for (n <- numbers) println(n+10)

    // but on BlockingObservable, it is:
    for (n <- numbers.toBlocking) println(n+10)
  }

  @Test def testForComprehension(): Unit = {
    val observables = List(List(1, 2, 3).toObservable, List(10, 20, 30).toObservable).toObservable
    val squares = (for (o <- observables; i <- o if i % 2 == 0) yield i*i)
    assertEquals(squares.toBlocking.toList, List(4, 100, 400, 900))
  }

  @Test def nextExample(): Unit = {
    val o = Observable.interval(100 millis).take(20)
    for(i <- o.toBlocking.next) {
      println(i)
      Thread.sleep(200)
    }
  }

  @Test def latestExample(): Unit = {
    val o = Observable.interval(100 millis).take(20)
    for(i <- o.toBlocking.latest) {
      println(i)
      Thread.sleep(200)
    }
  }

  @Test def toFutureExample(): Unit = {
    val o = Observable.interval(500 millis).take(1)
    val r = Await.result(o.toBlocking.toFuture, 2 seconds)
    println(r)
  }

  @Test def testTwoSubscriptionsToOneInterval(): Unit = {
    val o = Observable.interval(100 millis).take(8)
    o.subscribe(
      i => println(s"${i}a (on thread #${Thread.currentThread().getId})")
    )
    o.subscribe(
      i => println(s"${i}b (on thread #${Thread.currentThread().getId})")
    )
    waitFor(o)
  }

  @Test def schedulersExample(): Unit = {
    val o = Observable.interval(100 millis).take(8)
    o.observeOn(NewThreadScheduler()).subscribe(
      i => println(s"${i}a (on thread #${Thread.currentThread().getId})")
    )
    o.observeOn(NewThreadScheduler()).subscribe(
      i => println(s"${i}b (on thread #${Thread.currentThread().getId})")
    )
    waitFor(o)
  }

  @Test def testGroupByThenFlatMap(): Unit = {
    val m = List(1, 2, 3, 4).toObservable
    val g = m.groupBy(i => i % 2)
    val t = g.flatMap((p: (Int, Observable[Int])) => p._2)
    assertEquals(List(1, 2, 3, 4), t.toBlocking.toList)
  }

  @Test def testGroupByThenFlatMapByForComprehension(): Unit = {
    val m = List(1, 2, 3, 4).toObservable
    val g = m.groupBy(i => i % 2)
    val t = for ((i, o) <- g; n <- o) yield n
    assertEquals(List(1, 2, 3, 4), t.toBlocking.toList)
  }

  @Test def testGroupByThenFlatMapByForComprehensionWithTiming(): Unit = {
    val m = Observable.interval(100 millis).take(4)
    val g = m.groupBy(i => i % 2)
    val t = for ((i, o) <- g; n <- o) yield n
    assertEquals(List(0, 1, 2, 3), t.toBlocking.toList)
  }

  @Test def timingTest(): Unit = {
    val firstOnly = false
    val numbersByModulo3 = Observable.interval(1000 millis).take(9).groupBy(_ % 3)

    (for ((modulo, numbers) <- numbersByModulo3) yield {
      println("Observable for modulo" + modulo + " started")

      if (firstOnly) numbers.take(1) else numbers
    }).flatten.toBlocking.foreach(println(_))
  }

  @Test def timingTest1(): Unit = {
    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.map(n => s"${n} is in the modulo-$modulo group")
    }).flatten.toBlocking.foreach(println(_))
  }
  
  @Test def testOlympicYearTicks(): Unit = {
    Olympics.yearTicks.subscribe(println(_))
    waitFor(Olympics.yearTicks)
  }

  @Test def groupByExample(): Unit = {
    val medalsByCountry = Olympics.mountainBikeMedals.groupBy(medal => medal.country)

    val firstMedalOfEachCountry =
      for ((country, medals) <- medalsByCountry; firstMedal <- medals.take(1)) yield firstMedal

    firstMedalOfEachCountry.subscribe(medal => {
      println(s"${medal.country} wins its first medal in ${medal.year}")
    })
    
    Olympics.yearTicks.subscribe(year => println(s"\nYear $year starts."))

    waitFor(Olympics.yearTicks)
  }

  @Test def groupByExample2(): Unit = {
    val medalByYear = Olympics.mountainBikeMedals.groupBy(medal => medal.year, medal => medal.country)

    for ((year, countries) <- medalByYear; country <- countries) {
      println(s"${year}: ${country}")
    }

    Olympics.yearTicks.subscribe(year => println(s"\nYear $year starts."))

    waitFor(Olympics.yearTicks)
  }

  @Test def combineLatestExample(): Unit = {
    val firstCounter = Observable.interval(250 millis)
    val secondCounter = Observable.interval(550 millis)
    val combinedCounter = firstCounter.combineLatestWith(secondCounter)(List(_, _)) take 10

    combinedCounter subscribe {x => println(s"Emitted group: $x")}
    waitFor(combinedCounter)
  }

  @Test def combineLatestExample2(): Unit = {
    val firstCounter = Observable.interval(250 millis)
    val secondCounter = Observable.interval(550 millis)
    val thirdCounter = Observable.interval(850 millis)
    val sources = Iterable(firstCounter, secondCounter, thirdCounter)
    val combinedCounter = Observable.combineLatest(sources)(_.toList).take(10)

    combinedCounter subscribe {x => println(s"Emitted group: $x")}
    waitFor(combinedCounter)
  }

  @Test def combineLatestDelayErrorExample(): Unit = {
    val firstCounter = Observable.just(1) ++ Observable.error(new RuntimeException("Oops!"))
    val secondCounter = Observable.interval(550 millis).take(3)
    val sources = Iterable(firstCounter, secondCounter)

    Observable.combineLatest(sources)(_.toList).subscribe(
      i => println("combineLatest: " + i),
      e => println("combineLatest: " + e.getMessage))
    Thread.sleep(2000)
    Observable.combineLatestDelayError(sources)(_.toList).subscribe(
      i => println("combineLatestDelayError: " + i),
      e => println("combineLatestDelayError: " + e.getMessage))
    Thread.sleep(2000)
  }

  @Test def olympicsExampleWithoutPublish(): Unit = {
    val medals = Olympics.mountainBikeMedals.doOnEach(_ => println("onNext"))
    medals.subscribe(println(_)) // triggers an execution of medals Observable
    waitFor(medals) // triggers another execution of medals Observable
  }

  @Test def olympicsExampleWithPublish(): Unit = {
    val medals = Olympics.mountainBikeMedals.doOnEach(_ => println("onNext")).publish
    medals.subscribe(println(_)) // triggers an execution of medals Observable
    medals.connect
    waitFor(medals) // triggers another execution of medals Observable
  }

  @Test def exampleWithoutPublish(): Unit = {
    val unshared = Observable.from(1 to 4)
    unshared.subscribe(n => println(s"subscriber 1 gets $n"))
    unshared.subscribe(n => println(s"subscriber 2 gets $n"))
  }

  @Test def exampleWithPublish(): Unit = {
    val unshared = Observable.from(1 to 4)
    val shared = unshared.publish
    shared.subscribe(n => println(s"subscriber 1 gets $n"))
    shared.subscribe(n => println(s"subscriber 2 gets $n"))
    shared.connect
  }

  @Test def exampleWithPublish2(): Unit = {
    val o = Observable.interval(100 millis).take(5).publish((o: Observable[Long]) => o.map(_ * 2))
    o.subscribe(n => println(s"subscriber 1 gets $n"))
    o.subscribe(n => println(s"subscriber 2 gets $n"))
    Thread.sleep(1000)
  }

  def doLater(waitTime: Duration, action: () => Unit): Unit = {
    Observable.interval(waitTime).take(1).subscribe(_ => action())
  }

  @Test def exampleWithoutReplay(): Unit = {
    val numbers = Observable.interval(1000 millis).take(6)
    val sharedNumbers = numbers.publish
    sharedNumbers.subscribe(n => println(s"subscriber 1 gets $n"))
    sharedNumbers.connect
    // subscriber 2 misses 0, 1, 2!
    doLater(3500 millis, () => { sharedNumbers.subscribe(n => println(s"subscriber 2 gets $n")) })
    waitFor(sharedNumbers)
  }

  @Test def exampleWithReplay(): Unit = {
    val numbers = Observable.interval(1000 millis).take(6)
    val sharedNumbers = numbers.replay
    sharedNumbers.subscribe(n => println(s"subscriber 1 gets $n"))
    sharedNumbers.connect
    // subscriber 2 subscribes later but still gets all numbers
    doLater(3500 millis, () => { sharedNumbers.subscribe(n => println(s"subscriber 2 gets $n")) })
    waitFor(sharedNumbers)
  }

  @Test def exampleWithReplay2(): Unit = {
    val numbers = Observable.interval(100 millis).take(10)
    val sharedNumbers = numbers.replay(3)
    sharedNumbers.subscribe(n => println(s"subscriber 1 gets $n"))
    sharedNumbers.connect
    // subscriber 2 subscribes later but only gets the 3 buffered numbers and the following numbers
    Thread.sleep(700)
    sharedNumbers.subscribe(n => println(s"subscriber 2 gets $n"))
    waitFor(sharedNumbers)
  }

  @Test def exampleWithReplay3(): Unit = {
    val numbers = Observable.interval(100 millis).take(10)
    val sharedNumbers = numbers.replay(300 millis)
    sharedNumbers.subscribe(n => println(s"subscriber 1 gets $n"))
    sharedNumbers.connect
    // subscriber 2 subscribes later but only gets the buffered numbers and the following numbers
    Thread.sleep(700)
    sharedNumbers.subscribe(n => println(s"subscriber 2 gets $n"))
    waitFor(sharedNumbers)
  }

  @Test def exampleWithReplay4(): Unit = {
    val numbers = Observable.interval(100 millis).take(10)
    val sharedNumbers = numbers.replay(2, 300 millis)
    sharedNumbers.subscribe(n => println(s"subscriber 1 gets $n"))
    sharedNumbers.connect
    // subscriber 2 subscribes later but only gets the buffered numbers and the following numbers
    Thread.sleep(700)
    sharedNumbers.subscribe(n => println(s"subscriber 2 gets $n"))
    waitFor(sharedNumbers)
  }

  @Test def exampleWithReplay5(): Unit = {
    val numbers = Observable.interval(100 millis).take(10)
    val sharedNumbers = numbers.replay(o => o.map(_ * 2))
    sharedNumbers.subscribe(n => println(s"subscriber gets $n"))
    waitFor(sharedNumbers)
  }

  @Test def testSingleOption(): Unit = {
    assertEquals(Some(1), List(1).toObservable.toBlocking.singleOption)
    assertEquals(None,    List().toObservable.toBlocking.singleOption)
  }

  // We can't put a general average method into Observable.scala, because Scala's Numeric
  // does not have scalar multiplication (we would need to calculate (1.0/numberOfElements)*sum)
  def doubleAverage(o: Observable[Double]): Observable[Double] = {
    for ((finalSum, finalCount) <- o.foldLeft((0.0, 0))({case ((sum, count), elem) => (sum+elem, count+1)}))
    yield finalSum / finalCount
  }

  @Test def averageExample(): Unit = {
    println(doubleAverage(Observable.empty).toBlocking.single)
    println(doubleAverage(List(0.0).toObservable).toBlocking.single)
    println(doubleAverage(List(4.44).toObservable).toBlocking.single)
    println(doubleAverage(List(1.0, 2.0, 3.5).toObservable).toBlocking.single)
  }

  @Test def testSum(): Unit = {
    assertEquals(10, List(1, 2, 3, 4).toObservable.sum.toBlocking.single)
    assertEquals(6, List(4, 2).toObservable.sum.toBlocking.single)
    assertEquals(0, List[Int]().toObservable.sum.toBlocking.single)
  }

  @Test def testProduct(): Unit = {
    assertEquals(24, List(1, 2, 3, 4).toObservable.product.toBlocking.single)
    assertEquals(8, List(4, 2).toObservable.product.toBlocking.single)
    assertEquals(1, List[Int]().toObservable.product.toBlocking.single)
  }

  @Test def mapWithIndexExample(): Unit = {
    // We don't need mapWithIndex because we already have zipWithIndex, which we can easily
    // combine with map:
    List("a", "b", "c").toObservable.zipWithIndex.map(pair => pair._1 + " has index " + pair._2)
      .toBlocking.foreach(println(_))

    // Or even nicer with for-comprehension syntax:
    (for ((letter, index) <- List("a", "b", "c").toObservable.zipWithIndex) yield letter + " has index " + index)
      .toBlocking.foreach(println(_))
  }

  // source Observables are all known:
  @Test def zip3Example(): Unit = {
    val o = Observable.zip(List(1, 2).toObservable, List(10, 20).toObservable, List(100, 200).toObservable)
    (for ((n1, n2, n3) <- o) yield s"$n1, $n2 and $n3")
      .toBlocking.foreach(println(_))
  }

  // source Observables are in an Observable:
  @Test def zipManyObservableExample(): Unit = {
    val observables = List(List(1, 2).toObservable, List(10, 20).toObservable, List(100, 200).toObservable).toObservable
    (for (seq <- Observable.zip(observables)) yield seq.mkString("(", ", ", ")"))
      .toBlocking.foreach(println(_))
  }

  /**
   * This is a bad way of using `zip` with an `Iterable`: even if the consumer unsubscribes,
   * some elements may still be pulled from `Iterable`.
   */
  @Test def zipWithIterableBadExample(): Unit = {
    val o1 = Observable.interval(100 millis, IOScheduler()).map(_ * 100).take(3)
    val o2 = Observable.from(0 until Int.MaxValue).doOnEach(i => println(i + " from o2"))
    o1.zip(o2).toBlocking.foreach(println(_))
  }

  /**
   * This is a good way of using `zip` with an `Iterable`: if the consumer unsubscribes,
   * no more elements will be pulled from `Iterable`.
   */
  @Test def zipWithIterableGoodExample(): Unit = {
    val o1 = Observable.interval(100 millis, IOScheduler()).map(_ * 100).take(3)
    val iter = (0 until Int.MaxValue).view.map {
      i => {
        println(i + " from iter")
        i
      }
    }
    o1.zip(iter).toBlocking.foreach(println(_))
  }

  @Test def zipWithExample(): Unit = {
    val xs = Observable.just(1, 3, 5, 7)
    val ys = Observable.just(2, 4, 6, 8)
    xs.zipWith(ys)(_ * _).subscribe(println(_))
  }

  @Test def takeFirstWithCondition(): Unit = {
    val condition: Int => Boolean = _ >= 3
    assertEquals(3, List(1, 2, 3, 4).toObservable.filter(condition).first.toBlocking.single)
  }

  @Test def firstOrDefaultWithCondition(): Unit = {
    val condition: Int => Boolean = _ >= 3
    assertEquals(3, List(1, 2, 3, 4).toObservable.filter(condition).firstOrElse(10).toBlocking.single)
    assertEquals(10, List(-1, 0, 1).toObservable.filter(condition).firstOrElse(10).toBlocking.single)
  }

  @Test def firstLastSingleExample(): Unit = {
    assertEquals(1, List(1, 2, 3, 4).toObservable.head.toBlocking.single)
    assertEquals(1, List(1, 2, 3, 4).toObservable.first.toBlocking.single)
    assertEquals(4, List(1, 2, 3, 4).toObservable.last.toBlocking.single)
    assertEquals(1, List(1).toObservable.single.toBlocking.single)

    assertEquals(1, List(1, 2, 3, 4).toObservable.toBlocking.head)
    assertEquals(1, List(1, 2, 3, 4).toObservable.toBlocking.first)
    assertEquals(4, List(1, 2, 3, 4).toObservable.toBlocking.last)
    assertEquals(1, List(1).toObservable.toBlocking.single)
  }

  @Test def dropExample(): Unit = {
    val o = List(1, 2, 3, 4).toObservable
    assertEquals(List(3, 4), o.drop(2).toBlocking.toList)
  }

  @Test def dropWithTimeExample(): Unit = {
    val o = List(1, 2, 3, 4).toObservable.zip(
      Observable.interval(500 millis, IOScheduler())).map(_._1) // emit every 500 millis
    println(
      o.drop(1250 millis, IOScheduler()).toBlocking.toList // output List(3, 4)
    )
  }

  @Test def dropRightExample(): Unit = {
    val o = List(1, 2, 3, 4).toObservable
    assertEquals(List(1, 2), o.dropRight(2).toBlocking.toList)
  }

  @Test def dropRightWithTimeExample(): Unit = {
    val o = List(1, 2, 3, 4).toObservable.zip(
      Observable.interval(500 millis, IOScheduler())).map(_._1) // emit every 500 millis
    println(
      o.dropRight(750 millis, IOScheduler()).toBlocking.toList // output List(1, 2)
    )
  }

  @Test def dropUntilExample(): Unit = {
    val o = List("Alice", "Bob", "Carlos").toObservable.zip(
      Observable.interval(700 millis, IOScheduler())).map(_._1) // emit every 700 millis
    val other = List(1).toObservable.delay(1 seconds)
    println(
      o.dropUntil(other).toBlocking.toList // output List("Bob", "Carlos")
    )
  }

  def square(x: Int): Int = {
    println(s"$x*$x is being calculated on thread ${Thread.currentThread().getId}")
    Thread.sleep(100) // calculating a square is heavy work :)
    x*x
  }

  @Test def toSortedList(): Unit = {
    assertEquals(Seq(7, 8, 9, 10), List(10, 7, 8, 9).toObservable.toSeq.map(_.sorted).toBlocking.single)
    val f = (a: Int, b: Int) => b < a
    assertEquals(Seq(10, 9, 8, 7), List(10, 7, 8, 9).toObservable.toSeq.map(_.sortWith(f)).toBlocking.single)
  }

  @Test def timestampExample(): Unit = {
    val timestamped = Observable.interval(100 millis).take(6).timestamp.toBlocking
    for ((millis, value) <- timestamped if value > 0) {
      println(value + " at t = " + millis)
    }
  }

  @Test def materializeExample1(): Unit = {
    def printObservable[T](o: Observable[T]): Unit = {
      import Notification._
      o.materialize.subscribe(n => n match {
        case OnNext(v) => println("Got value " + v)
        case OnCompleted => println("Completed")
        case OnError(err) => println("Error: " + err.getMessage)
      })
    }

    val o1 = Observable.interval(100 millis).take(3)
    val o2 = Observable.error(new IOException("Oops"))
    printObservable(o1)
    printObservable(o2)
    Thread.sleep(500)
  }

  @Test def materializeExample2(): Unit = {
    import Notification._
    List(1, 2, 3).toObservable.materialize.subscribe(n => n match {
      case OnNext(v) => println("Got value " + v)
      case OnCompleted => println("Completed")
      case OnError(err) => println("Error: " + err.getMessage)
    })
  }

  @Test def notificationSubtyping(): Unit = {
    import Notification._
    val oc1: Notification[Nothing] = OnCompleted
    val oc2: Notification[Int] = OnCompleted
    val oc3: rx.Notification[_ <: Int] = JavaConversions.toJavaNotification(oc2)
    val oc4: rx.Notification[_ <: Any] = JavaConversions.toJavaNotification(oc2)
  }

  @Test def takeWhileWithIndexAlternative: Unit = {
    val condition = true
    List("a", "b").toObservable.zipWithIndex.takeWhile{case (elem, index) => condition}.map(_._1)
  }
  
  def calculateElement(index: Int): String = {
    println("omg I'm calculating so hard")
    index match {
      case 0 => "a"
      case 1 => "b"
      case _ => throw new IllegalArgumentException
    }
  }

  /**
    * Using AsyncOnSubscribe of SyncOnSubscribe ensures that your Observable
    * does not calculate elements after the consumer unsubscribes and correctly does not overload
    * the subscriber with data (responds to backpressure).
    *
    * Here we create a SyncOnSubscribe without state, this means that it performs the same action every time downstream can handle more data.
    */
  @Test def createExample(): Unit = {
    val o = Observable.create(SyncOnSubscribe.stateless(
      next = () => Notification.OnNext(math.random),
      onUnsubscribe = () => println("I have stopped generating random numbers for this subscriber")
    ))
    o.take(10).foreach(r => println(s"Next random number: $r"))
  }

  /**
   * You can also add state to (A)SyncOnSubscribe, you generate a state on each subscription and can alter that state in each next call
   * Here we use it to count to a specific number
   */
  @Test def createExampleWithState(): Unit = {
    // Starts with state `0`
    val o = Observable.create(SyncOnSubscribe(() => 0)(i => {
      if(i < 2)
        // Check if the state has reached 2 yet, if not, we emit the current state and add 1 to the state
        (Notification.OnNext(i), i+1)
      else
        // Otherwise we signal completion
        (Notification.OnCompleted, i)
    }))
    o.take(1).subscribe(println(_))
  }

  /**
    * This example shows how to read a (potentially blocking) data source step-by-step (line by line) using SyncOnSubscribe.
    */
  @Test def createExampleFromInputStream(): Unit = {
    import scala.io.{Codec, Source}

    val rxscalaURL = "http://reactivex.io/rxscala/"
    // We use the `singleState` helper here, since we only want to generate one state per subscriber
    // and do not need to modify it afterwards
    val rxscala = Observable.create(SyncOnSubscribe.singleState(
      // This is our `generator`, which generates a state
      generator = () => {
        val input = new java.net.URL(rxscalaURL).openStream()
        (input, Source.fromInputStream(input)(Codec.UTF8).getLines())
      })(
      // This is our `next` function, which gets called whenever the subscriber can handle more data
      next = {
        case (_, lines) => {
          if(lines.hasNext)
            // Here we provide the next line
            Notification.OnNext(lines.next())
          else
            // Here we signal that the stream has completed
            Notification.OnCompleted
          }
        },
      // This is our `onUnsubscribe` function, which gets called after the subscriber unsubscribes, usually to perform cleanup
      onUnsubscribe = {
        case (input, _) => scala.util.Try { input.close() }
      }
    )).subscribeOn(IOScheduler())

    val count = rxscala.flatMap(_.split("\\W+").toSeq.toObservable)
      .map(_.toLowerCase)
      .filter(_ == "rxscala")
      .size
    println(s"RxScala appears ${count.toBlocking.single} times in ${rxscalaURL}")
  }

  /** This example show how to generate an Observable using AsyncOnSubscribe, which can be more efficient than SyncOnSubscribe.
    * Using AsyncOnSubscribe has the same advantages as SyncOnSubscribe, and furthermore it allows us to generate more than one
    * result at a time (which can be more efficient) and allows us to generate the results asynchronously.
    */
  @Test def createExampleAsyncOnUnsubscribe(): Unit = {
    // We are going to count to this number
    val countTo = 200L

    val o = Observable.create(AsyncOnSubscribe(() => 0L)(
      (count, demand) => {
        // Stop counting if we're past the number we were going to count to
        if(count > countTo)
          (Notification.OnCompleted, count)
        else {
          // Generate an observable that contains [count,count+demand) and thus contains exactly `demand` items
          val to = math.min(count + demand, countTo+1)
          val range = count until to
          val resultObservable = Observable.from(range)
          println(s"Currently at $count, received a demand of $demand. Next range [$count,$to)")
          (Notification.OnNext(resultObservable), to)
        }
      }
    ))
    o.subscribe(new Subscriber[Long] {
      override def onStart(): Unit = request(10)
      override def onNext(i: Long): Unit = {
        request(scala.util.Random.nextInt(10)+1)
      }
    })
  }

  def output(s: String): Unit = println(s)

  /** 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.toBlocking.toIterable.last
  }

  @Test def doOnCompletedExample(): Unit = {
    val o = List("red", "green", "blue").toObservable.doOnCompleted { println("onCompleted") }
    o.subscribe(v => println(v), e => e.printStackTrace)
    // red
    // green
    // blue
    // onCompleted
  }

  @Test def doOnSubscribeExample(): Unit = {
    val o = List("red", "green", "blue").toObservable.doOnSubscribe { println("subscribed") }
    o.subscribe(v => println(v), e => e.printStackTrace, () => println("onCompleted"))
    // subscribed
    // red
    // green
    // blue
    // onCompleted
  }

  @Test def doOnTerminateExample(): Unit = {
    val o = List("red", "green", "blue").toObservable.doOnTerminate { println("terminate") }
    o.subscribe(v => println(v), e => e.printStackTrace, () => println("onCompleted"))
    // red
    // green
    // blue
    // terminate
    // onCompleted
  }

  @Test def doOnUnsubscribeExample(): Unit = {
    val o = List("red", "green", "blue").toObservable.doOnUnsubscribe { println("unsubscribed") }
    o.subscribe(v => println(v), e => e.printStackTrace, () => println("onCompleted"))
    // red
    // green
    // blue
    // onCompleted
    // unsubscribed
  }

  @Test def doAfterTerminateExample(): Unit = {
    val o = List("red", "green", "blue").toObservable.doAfterTerminate { println("finally") }
    o.subscribe(v => println(v), e => e.printStackTrace, () => println("onCompleted"))
    // red
    // green
    // blue
    // onCompleted
    // finally
  }

  @Test def timeoutExample(): Unit = {
    val other = List(100L, 200L, 300L).toObservable
    val result = Observable.interval(100 millis).timeout(50 millis, other).toBlocking.toList
    println(result)
  }

  @Test def timeoutExample2(): Unit = {
    val firstTimeoutSelector = () => {
      Observable.interval(10 seconds, 10 seconds, ComputationScheduler()).take(1)
    }
    val timeoutSelector = (t: Long) => {
      Observable.interval(
        (500 - t * 100) max 1 millis,
        (500 - t * 100) max 1 millis,
        ComputationScheduler()).take(1)
    }
    val other = List(100L, 200L, 300L).toObservable
    val result = Observable.interval(100 millis).timeout(firstTimeoutSelector, timeoutSelector, other).toBlocking.toList
    println(result)
  }

  @Test def ambExample(): Unit = {
    val o1 = List(100L, 200L, 300L).toObservable.delay(4 seconds)
    val o2 = List(1000L, 2000L, 3000L).toObservable.delay(2 seconds)
    val result = o1.amb(o2).toBlocking.toList
    println(result)
  }

  @Test def ambWithVarargsExample(): Unit = {
    val o1 = List(100L, 200L, 300L).toObservable.delay(4 seconds)
    val o2 = List(1000L, 2000L, 3000L).toObservable.delay(2 seconds)
    val o3 = List(10000L, 20000L, 30000L).toObservable.delay(4 seconds)
    val result = Observable.amb(o1, o2, o3).toBlocking.toList
    println(result)
  }

  @Test def ambWithSeqExample(): Unit = {
    val o1 = List(100L, 200L, 300L).toObservable.delay(4 seconds)
    val o2 = List(1000L, 2000L, 3000L).toObservable.delay(2 seconds)
    val o3 = List(10000L, 20000L, 30000L).toObservable.delay(4 seconds)
    val o = Seq(o1, o2, o3)
    val result = Observable.amb(o: _*).toBlocking.toList
    println(result)
  }

  @Test def delayExample(): Unit = {
    val o = List(100L, 200L, 300L).toObservable.delay(2 seconds)
    val result = o.toBlocking.toList
    println(result)
  }

  @Test def delayExample2(): Unit = {
    val o = List(100L, 200L, 300L).toObservable.delay(2 seconds, IOScheduler())
    val result = o.toBlocking.toList
    println(result)
  }

  @Test def delayExample3(): Unit = {
    val o = List(100, 500, 200).toObservable.delay(
      (i: Int) => Observable.just(i).delay(i millis)
    )
    o.toBlocking.foreach(println(_))
  }

  @Test def delayExample4(): Unit = {
    val o = List(100, 500, 200).toObservable.delay(
      () => Observable.interval(500 millis).take(1),
      (i: Int) => Observable.just(i).delay(i millis)
    )
    o.toBlocking.foreach(println(_))
  }

  @Test def delaySubscriptionExample(): Unit = {
    val o = List(100L, 200L, 300L).toObservable.delaySubscription(2 seconds)
    val result = o.toBlocking.toList
    println(result)
  }

  @Test def delaySubscriptionExample2(): Unit = {
    val o = List(100L, 200L, 300L).toObservable.delaySubscription(2 seconds, IOScheduler())
    val result = o.toBlocking.toList
    println(result)
  }

  @Test def elementAtExample(): Unit = {
    val o = List("red", "green", "blue").toObservable
    println(o.elementAt(2).toBlocking.single)
  }

  @Test def elementAtOrDefaultExample(): Unit = {
    val o : Observable[Seq[Char]] = List("red".toList, "green".toList, "blue".toList).toObservable.elementAtOrDefault(3, "black".toSeq)
    println(o.toBlocking.single)
  }

  @Test def toMapExample1(): Unit = {
    val o : Observable[String] = List("alice", "bob", "carol").toObservable
    val keySelector = (s: String) => s.head
    val m = o.toMap(keySelector)
    println(m.toBlocking.single)
  }

  @Test def toMapExample2(): Unit = {
    val o : Observable[String] = List("alice", "bob", "carol").toObservable
    val keySelector = (s: String) => s.head
    val valueSelector = (s: String) => s.tail
    val m = o.toMap(keySelector, valueSelector)
    println(m.toBlocking.single)
  }

  @Test def toMapExample3(): Unit = {
    val o : Observable[String] = List("alice", "bob", "carol").toObservable
    val keySelector = (s: String) => s.head
    val valueSelector = (s: String) => s.tail
    val m: Observable[HashMap[Char, String]] = o.to[HashMap, Char, String](keySelector, valueSelector)
    println(m.toBlocking.single)
  }

  @Test def toMultimapExample1(): Unit = {
    val o : Observable[String] = List("alice", "bob", "carol", "allen", "clarke").toObservable
    val keySelector = (s: String) => s.head
    val m = o.toMultiMap(keySelector)
    println(m.toBlocking.single)
  }

  @Test def toMultimapExample2(): Unit = {
    val o : Observable[String] = List("alice", "bob", "carol", "allen", "clarke").toObservable
    val keySelector = (s: String) => s.head
    val valueSelector = (s: String) => s.tail
    val m = o.toMultiMap(keySelector, valueSelector)
    println(m.toBlocking.single)
  }

  @Test def toMultimapExample3(): Unit = {
    val o: Observable[String] = List("alice", "bob", "carol", "allen", "clarke").toObservable
    val keySelector = (s: String) => s.head
    val valueSelector = (s: String) => s.tail
    val m = o.toMultiMap(keySelector, valueSelector, {
      new mutable.HashMap[Char, mutable.Set[String]] with mutable.MultiMap[Char, String] addBinding('d', "oug")
    })
    println(m.toBlocking.single.mapValues(_.toList))
  }

  @Test def toMultimapExample4(): Unit = {
    val o : Observable[String] = List("alice", "bob", "carol", "allen", "clarke").toObservable
    val keySelector = (s: String) => s.head
    val valueSelector = (s: String) => s.tail
    val m = o.toMultiMap(keySelector, valueSelector, {
      new mutable.HashMap[Char, mutable.Set[String]] with mutable.MultiMap[Char, String] {
        override def makeSet = new mutable.TreeSet[String]
      }
    })
    println(m.toBlocking.single)
  }

  @Test def containsExample(): Unit = {
    val o1 = List(1, 2, 3).toObservable.contains(2)
    assertTrue(o1.toBlocking.single)

    val o2 = List(1, 2, 3).toObservable.contains(4)
    assertFalse(o2.toBlocking.single)
  }

  @Test def repeatExample1(): Unit = {
    val o : Observable[String] = List("alice", "bob", "carol").toObservable.repeat.take(6)
    assertEquals(List("alice", "bob", "carol", "alice", "bob", "carol"), o.toBlocking.toList)
  }

  @Test def repeatExample2(): Unit = {
    val o : Observable[String] = List("alice", "bob", "carol").toObservable.repeat(2)
    assertEquals(List("alice", "bob", "carol", "alice", "bob", "carol"), o.toBlocking.toList)
  }

  @Test def retryExample1(): Unit = {
    val o : Observable[String] = List("alice", "bob", "carol").toObservable
    assertEquals(List("alice", "bob", "carol"), o.retry.toBlocking.toList)
  }

  @Test def retryExample2(): Unit = {
    val o : Observable[String] = List("alice", "bob", "carol").toObservable
    assertEquals(List("alice", "bob", "carol"), o.retry(3).toBlocking.toList)
  }

  @Test def retryExample3(): Unit = {
    var isFirst = true
    val o = Observable {
      (subscriber: Subscriber[String]) =>
        if (isFirst) {
          subscriber.onNext("alice")
          subscriber.onError(new IOException("Oops"))
          isFirst = false
        }
        else {
          subscriber.onNext("bob")
          subscriber.onError(new RuntimeException("Oops"))
        }
    }
    o.retry {
      (times, e) => e match {
        case e: IOException => times <= 3
        case _ => false
      }
    }.subscribe(s => println(s), e => e.printStackTrace())
  }

  @Test def retryWhenExample(): Unit = {
    Observable[String]({ subscriber =>
      println("subscribing")
      subscriber.onError(new RuntimeException("always fails"))
    }).retryWhen({ throwableObservable =>
      throwableObservable.zipWith(Observable.from(1 to 3))((t, i) => i).flatMap(i => {
        println("delay retry by " + i + " second(s)")
        Observable.timer(Duration(i, TimeUnit.SECONDS))
      })
    }).toBlocking.foreach(s => println(s))
  }

  @Test def retryWhenDifferentExceptionsExample(): Unit = {
    var observableCreateCount = 1 // Just to support switching which Exception is produced
    Observable[String]({ subscriber =>
      println("subscribing")
      if (observableCreateCount <= 2) {
        subscriber.onError(new IOException("IO Fail"))
      } else {
        subscriber.onError(new RuntimeException("Other failure"))
      }
      observableCreateCount += 1
    }).retryWhen({ throwableObservable =>
      throwableObservable.zip(Observable.from(1 to 3)).flatMap({ case (error, retryCount) =>
        error match {
          // Only retry 2 times if we get a IOException and then error out with the third IOException.
          // Let the other Exception's pass through and complete the Observable.
          case _: IOException =>
            if (retryCount <= 3) {
              println("IOException delay retry by " + retryCount + " second(s)")
              Observable.timer(Duration(retryCount, TimeUnit.SECONDS))
            } else {
              Observable.error(error)
            }

          case _ =>
            println("got error " + error + ", will stop retrying")
            Observable.empty
        }
      })
    }).toBlocking.foreach(s => println(s))
  }

  @Test def repeatWhenExample(): Unit = {
    Observable[String]({ subscriber =>
      println("subscribing")
      subscriber.onCompleted()
    }).repeatWhen({ unitObservable =>
      unitObservable.zipWith(Observable.from(1 to 3))((u, i) => i).flatMap(i => {
        println("delay repeat by " + i + " second(s)")
        Observable.timer(Duration(i, TimeUnit.SECONDS))
      })
    }, NewThreadScheduler()).toBlocking.foreach(s => println(s))
  }

  @Test def liftExample1(): Unit = {
    // Add "No. " in front of each item
    val o = List(1, 2, 3).toObservable.lift {
      subscriber: Subscriber[String] =>
        new Subscriber[Int](subscriber) {
          override def onStart(): Unit = {
            request(Long.MaxValue)
          }

          override def onNext(v: Int): Unit = {
            subscriber.onNext("No. " + v)
          }

          override def onError(e: Throwable): Unit = {
            subscriber.onError(e)
          }

          override def onCompleted(): Unit = {
            subscriber.onCompleted
          }
        }
    }.toBlocking.toList
    println(o)
  }

  @Test def liftExample2(): Unit = {
    // Split the input Strings with " "
    val splitStringsWithSpace = (subscriber: Subscriber[String]) => {
      new Subscriber[String](subscriber) {
        override def onStart(): Unit = {
          request(Long.MaxValue)
        }

        override def onNext(v: String): Unit = {
          v.split(" ").foreach(subscriber.onNext(_))
        }

        override def onError(e: Throwable): Unit = {
          subscriber.onError(e)
        }

        override def onCompleted(): Unit = {
          subscriber.onCompleted
        }
      }
    }

    // Convert the input Strings to Chars
    val stringsToChars = (subscriber: Subscriber[Char]) => {
      new Subscriber[String](subscriber) {
        override def onStart(): Unit = {
          request(Long.MaxValue)
        }

        override def onNext(v: String): Unit = {
          v.foreach(subscriber.onNext(_))
        }

        override def onError(e: Throwable): Unit = {
          subscriber.onError(e)
        }

        override def onCompleted(): Unit = {
          subscriber.onCompleted
        }
      }
    }

    // Skip the first n items. If the length of source is less than n, throw an IllegalArgumentException
    def skipWithException[T](n: Int) = (subscriber: Subscriber[T]) => {
      new Subscriber[T](subscriber) {

        override def onStart(): Unit = {
          request(Long.MaxValue)
        }

        var count = 0

        override def onNext(v: T): Unit = {
          if (count >= n) subscriber.onNext(v)
          count += 1
        }


        override def onError(e: Throwable): Unit = {
          subscriber.onError(e)
        }

        override def onCompleted(): Unit = {
          if (count < n) {
            subscriber.onError(new IllegalArgumentException("There is no enough items"))
          } else {
            subscriber.onCompleted
          }
        }
      }
    }

    val o = List("RxJava – Reactive Extensions for the JVM").toObservable
      .lift(splitStringsWithSpace)
      .map(_.toLowerCase)
      .lift(stringsToChars)
      .filter(_.isLetter)
      .lift(skipWithException(100))
    try {
      o.toBlocking.toList
    }
    catch {
      case e: IllegalArgumentException => println("IllegalArgumentException from skipWithException")
    }
  }

  @Test def startWithExample(): Unit = {
    val o1 = List(3, 4).toObservable
    val o2 = 1 +: 2 +: o1
    assertEquals(List(1, 2, 3, 4), o2.toBlocking.toList)
  }

  @Test def appendExample(): Unit = {
    val o = List(1, 2).toObservable :+ 3 :+ 4
    assertEquals(List(1, 2, 3, 4), o.toBlocking.toList)
  }

  @Test def sequenceEqualExampe(): Unit = {
    val o1 = List(1, 2, 3).toObservable
    val o2 = List(1, 2, 3).toObservable
    val o3 = List(1, 2).toObservable
    val o4 = List(1.0, 2.0, 3.0).toObservable
    assertTrue(o1.sequenceEqual(o2).toBlocking.single)
    assertFalse(o1.sequenceEqual(o3).toBlocking.single)
    assertTrue(o1.sequenceEqual(o4).toBlocking.single)
  }

  @Test def takeExample(): Unit = {
    val o = (1 to 20).toObservable
      .zip(Observable.interval(300 millis))
      .map(_._1)
      .take(2 seconds)
    println(o.toBlocking.toList)
  }

  @Test def takeRightExample(): Unit = {
    val o = (1 to 6).toObservable.takeRight(3)
    assertEquals(List(4, 5, 6), o.toBlocking.toList)
  }

  @Test def takeRightExample2(): Unit = {
    val o = (1 to 10).toObservable
      .zip(Observable.interval(100 millis))
      .map(_._1)
      .takeRight(300 millis)
    println(o.toBlocking.toList)
  }

  @Test def takeRightExample3(): Unit = {
    val o = (1 to 10).toObservable
      .zip(Observable.interval(100 millis))
      .map(_._1)
      .takeRight(2, 300 millis)
    println(o.toBlocking.toList)
  }

  @Test def timeIntervalExample(): Unit = {
    val o = (1 to 10).toObservable
      .zip(Observable.interval(100 millis))
      .map(_._1)
      .timeInterval
    println(o.toBlocking.toList)
  }

  @Test def schedulerExample1(): Unit = {
    val latch = new CountDownLatch(1)
    val worker = IOScheduler().createWorker
    worker.schedule {
      println("Hello from Scheduler")
      latch.countDown()
    }
    latch.await(5, TimeUnit.SECONDS)
  }

  @Test def schedulerExample2(): Unit = {
    val latch = new CountDownLatch(1)
    val worker = IOScheduler().createWorker
    worker.schedule(1 seconds) {
      println("Hello from Scheduler after 1 second")
      latch.countDown()
    }
    latch.await(5, TimeUnit.SECONDS)
  }

  @Test def schedulerExample3(): Unit = {
    val worker = IOScheduler().createWorker
    var no = 1
    val subscription = worker.schedulePeriodically(initialDelay = 1 seconds, period = 100 millis) {
      println(s"Hello(${no}) from Scheduler")
      no += 1
    }
    TimeUnit.SECONDS.sleep(2)
    subscription.unsubscribe()
  }

  @Test def schedulerExample4(): Unit = {
    val worker = IOScheduler().createWorker
    var no = 1
    def hello: Unit = {
      println(s"Hello(${no}) from Scheduler")
      no += 1
      worker.schedule(100 millis)(hello)
    }
    val subscription = worker.schedule(1 seconds)(hello)
    TimeUnit.SECONDS.sleep(2)
    subscription.unsubscribe()
  }

  def sayHelloFromExecutionContext(ec: ExecutionContext): Unit = {
    val scheduler = ExecutionContextScheduler(ec)
    val latch = new CountDownLatch(1)
    Observable.just("Hello").subscribeOn(scheduler).subscribe(
      v => println(s"$v from ${Thread.currentThread.getName}"),
      e => e.printStackTrace(),
      () => latch.countDown()
    )
    latch.await(5, TimeUnit.SECONDS)
  }

  @Test def executionContextSchedulerExample(): Unit = {
    sayHelloFromExecutionContext(scala.concurrent.ExecutionContext.Implicits.global)
  }

  @Test def executionContextSchedulerExample2(): Unit = {
    sayHelloFromExecutionContext(ExecutionContext.fromExecutor(Executors.newCachedThreadPool()))
  }

  def createAHotObservable: Observable[String] = {
    var first = true
    Observable[String] {
      subscriber =>
        if (first) {
          subscriber.onNext("1st: First")
          subscriber.onNext("1st: Last")
          first = false
        }
        else {
          subscriber.onNext("2nd: First")
          subscriber.onNext("2nd: Last")
        }
        subscriber.onCompleted()
    }
  }

  @Test def withoutPublishLastExample(): Unit = {
    val hot = createAHotObservable
    hot.takeRight(1).subscribe(n => println(s"subscriber 1 gets $n"))
    hot.takeRight(1).subscribe(n => println(s"subscriber 2 gets $n"))
  }

  @Test def unsubscribeOnExample(): Unit = {
    val o = Observable[String] {
      subscriber =>
        subscriber.add(Subscription {
          println("unsubscribe on " + Thread.currentThread().getName())
        })
        subscriber.onNext("RxScala")
        subscriber.onCompleted()
    }
    o.unsubscribeOn(NewThreadScheduler()).subscribe(println(_))
  }

  @Test def debounceExample(): Unit = {
    val o = Observable.interval(100 millis).take(20).debounce {
      n =>
        if (n % 2 == 0) {
          Observable.interval(50 millis)
        }
        else {
          Observable.interval(150 millis)
        }
    }
    o.toBlocking.foreach(println(_))
  }

  @Test def flatMapExample(): Unit = {
    val o = Observable.just(10, 100)
    o.flatMap(n => Observable.interval(200 millis).map(_ * n))
      .take(20)
      .toBlocking.foreach(println)
  }

  @Test def flatMapExample2(): Unit = {
    val o = Observable.just(10, 100)
    val o1 = for (n <- o;
                  i <- Observable.interval(200 millis)) yield i * n
    o1.take(20).toBlocking.foreach(println)
  }

  @Test def flatMapExample3(): Unit = {
    val o = Observable[Int] {
      subscriber =>
        subscriber.onNext(10)
        subscriber.onNext(100)
        subscriber.onError(new IOException("Oops"))
    }
    o.flatMap(
      (n: Int) => Observable.interval(200 millis).map(_ * n),
      e => Observable.interval(200 millis).map(_ * -1),
      () => Observable.interval(200 millis).map(_ * 1000)
    ).take(20)
      .toBlocking.foreach(println)
  }

  @Test def flatMapExample4(): Unit = {
    val o = Observable.just(10, 100)
    o.flatMap(
      (n: Int) => Observable.interval(200 millis).map(_ * n),
      e => Observable.interval(200 millis).map(_ * -1),
      () => Observable.interval(200 millis).map(_ * 1000)
    ).take(20)
      .toBlocking.foreach(println)
  }

  @Test def flatMapExample5(): Unit = {
    val o = Observable.just(1, 10, 100, 1000)
    o.flatMapWith(_ => Observable.interval(200 millis).take(5))(_ * _).toBlocking.foreach(println)
  }

  @Test def flatMapIterableExample(): Unit = {
    val o = Observable.just(10, 100)
    o.flatMapIterable(n => (1 to 20).map(_ * n))
      .toBlocking.foreach(println)
  }

  @Test def flatMapIterableExample2(): Unit = {
    val o = Observable.just(1, 10, 100, 1000)
    o.flatMapIterableWith(_=> (1 to 5))(_ * _).toBlocking.foreach(println)
  }

  @Test def concatMapExample(): Unit = {
    val o = Observable.just(10, 100)
    o.concatMap(n => Observable.interval(200 millis).map(_ * n).take(10))
      .take(20)
      .toBlocking.foreach(println)
  }

  @Test def concatMapDelayErrorExample(): Unit = {
    println("=== concatMap ===")
    (1 to 10).toObservable
      .concatMap { i =>
        if (i == 2) Observable.error(new IOException("Oops")) else Observable.just(i)
      }.subscribe(println(_), _.printStackTrace)

    println("=== delayError.concatMap ===")
    (1 to 10).toObservable
      .delayError.concatMap { i =>
        if (i == 2) Observable.error(new IOException("Oops")) else Observable.just(i)
      }.subscribe(println(_), _.printStackTrace)
  }

  @Test def onErrorResumeNextExample(): Unit = {
    val o = Observable {
      (subscriber: Subscriber[Int]) =>
        subscriber.onNext(1)
        subscriber.onNext(2)
        subscriber.onError(new IOException("Oops"))
    }
    o.onErrorResumeNext(_ => Observable.just(10, 11, 12)).subscribe(println(_))
  }

  @Test def onErrorResumeNextCaseExample(): Unit = {
    val o = Observable {
      (subscriber: Subscriber[Int]) =>
        subscriber.onNext(1)
        subscriber.onNext(2)
        subscriber.onError(new IOException("Oops"))
    }
    o.onErrorResumeNext {
      case e: IOException => Observable.just(20, 21, 22)
      case _ => Observable.just(10, 11, 12)
    }.subscribe(println(_))
  }

  @Test def switchExample(): Unit = {
    Observable.interval(300 millis).take(3).map { n =>
      Observable.interval(100 millis)
        .map(l => s"o$n emit $l")
        .take(3)
        .doOnSubscribe(println(s"subscribe to o$n"))
    }.switch.take(5).subscribe(println(_))
  }

  @Test def switchDelayErrorExample(): Unit = {
    println("=== switch ===")
    Observable.interval(300 millis).take(3).map { n =>
      if (n == 0) {
        Observable.error(new RuntimeException("Oops!"))
      } else {
        Observable.interval(100 millis)
          .map(l => s"o$n emit $l")
          .take(3)
          .doOnSubscribe(println(s"subscribe to o$n"))
      }
    }.switch.subscribe(println(_), _.printStackTrace())

    Thread.sleep(2000)

    println("=== delayError.switch ===")
    Observable.interval(300 millis).take(3).map { n =>
      if (n == 0) {
        Observable.error(new RuntimeException("Oops!"))
      } else {
        Observable.interval(100 millis)
          .map(l => s"o$n emit $l")
          .take(3)
          .doOnSubscribe(println(s"subscribe to o$n"))
      }
    }.delayError.switch.subscribe(println(_), _.printStackTrace())
  }

  @Test def switchMapExample(): Unit = {
    val o = Observable.interval(300 millis).take(5).switchMap[String] {
      n => Observable.interval(50 millis).take(10).map(i => s"Seq ${n}: ${i}")
    }
    o.toBlocking.foreach(println)
  }

  @Test def switchMapDelayErrorExample(): Unit = {
    println("=== switchMap ===")
    Observable.interval(300 millis).take(3).switchMap { n =>
      if (n == 0) {
        Observable.error(new RuntimeException("Oops!"))
      } else {
        Observable.interval(100 millis)
          .map(l => s"o$n emit $l")
          .take(3)
          .doOnSubscribe(println(s"subscribe to o$n"))
      }
    }.subscribe(println(_), _.printStackTrace())

    Thread.sleep(2000)

    println("=== delayError.switchMap ===")
    Observable.interval(300 millis).take(3).delayError.switchMap { n =>
      if (n == 0) {
        Observable.error(new RuntimeException("Oops!"))
      } else {
        Observable.interval(100 millis)
          .map(l => s"o$n emit $l")
          .take(3)
          .doOnSubscribe(println(s"subscribe to o$n"))
      }
    }.subscribe(println(_), _.printStackTrace())
  }

  @Test def joinExample(): Unit = {
    val o1 = Observable.interval(500 millis).map(n => "1: " + n)
    val o2 = Observable.interval(100 millis).map(n => "2: " + n)
    val o = o1.join(o2)(_ => Observable.timer(300 millis), _ => Observable.timer(200 millis), (_, _))
    o.take(10).toBlocking.foreach(println)
  }

  @Test def groupJoinExample(): Unit = {
    val o1 = Observable.interval(500 millis).map(n => "1: " + n)
    val o2 = Observable.interval(100 millis).map(n => "2: " + n)
    val o = o1.groupJoin(o2)(_ => Observable.timer(300 millis), _ => Observable.timer(200 millis), (t1, t2) => (t1, t2.toSeq.toBlocking.single))
    o.take(3).toBlocking.foreach(println)
  }

  @Test def collectExample(): Unit = {
    val o = Observable.just(1, 1.0, "a", 2, 2.0, "b")
    o.collect { case s: String => "Item: " + s }.foreach(println(_))
  }

  @Test def usingExample(): Unit = {
    import scala.io.{Codec, Source}

    Observable.using { new java.net.URL("http://rxscala.github.io/").openStream() }(
      input => Source.fromInputStream(input)(Codec.UTF8).getLines().toList.toObservable,
      input => input.close
    ).foreach(println(_))
  }

  def createFastObservable: Observable[Int] = {
    Observable {
      subscriber: Subscriber[Int] => {
        var i = 0
        while (i < 2000 && !subscriber.isUnsubscribed) {
          subscriber.onNext(i)
          i += 1
        }
        if (!subscriber.isUnsubscribed) {
          subscriber.onCompleted()
        }
      }
    }
  }

  @Test def withoutBackpressureExample(): Unit = {
    val o = createFastObservable
    val l = new CountDownLatch(1)
    o.observeOn(NewThreadScheduler()).subscribe(new Subscriber[Int] {
      override def onStart(): Unit = {
        request(1)
      }

      override def onNext(n: Int): Unit = {
        println(n)
        Thread.sleep(10) // emulate a slow subscriber
        request(1)
      }

      override def onError(e: Throwable): Unit = {
        e.printStackTrace()
        l.countDown()
      }

      override def onCompleted(): Unit = {
        l.countDown()
      }
    })
    l.await()
  }

  @Test def onBackpressureDropExample(): Unit = {
    val o = createFastObservable.onBackpressureDrop
    val l = new CountDownLatch(1)
    // Use a small buffer to demonstrate the drop behavior
    o.observeOn(NewThreadScheduler()).subscribe(new Subscriber[Int] {
      override def onStart(): Unit = {
        request(1)
      }

      override def onNext(n: Int): Unit = {
        println(n)
        Thread.sleep(10) // emulate a slow subscriber
        request(1)
      }

      override def onError(e: Throwable): Unit = {
        e.printStackTrace()
        l.countDown()
      }

      override def onCompleted(): Unit = {
        l.countDown()
      }
    })
    l.await()
  }

  @Test def onBackpressureBufferExample(): Unit = {
    val o = createFastObservable.onBackpressureBuffer
    val l = new CountDownLatch(1)
    o.observeOn(NewThreadScheduler()).subscribe(new Subscriber[Int] {
      override def onStart(): Unit = {
        request(1)
      }

      override def onNext(n: Int): Unit = {
        println(n)
        Thread.sleep(10) // emulate a slow subscriber
        request(1)
      }

      override def onError(e: Throwable): Unit = {
        e.printStackTrace()
        l.countDown()
      }

      override def onCompleted(): Unit = {
        l.countDown()
      }
    })
    l.await()
  }
  
  @Test def eventBusExample(): Unit = {
    val eventBus = Subject[String]()
    val threadsafeEventBus = SerializedSubject[String](eventBus)
    
    val event1ProducerSubscription = Observable.interval(10.milliseconds)
      .map(i => s"Event 1 producer: $i")
      .subscribeOn(NewThreadScheduler())
      .subscribe(
        { ev => threadsafeEventBus.onNext(ev) },
        (e: Throwable) => { println("Event 1 producer exception: $e") }
      )

    val event2ProducerSubscription = Observable.interval(10.milliseconds)
      .map(i => s"Event 2 producer: $i")
      .subscribeOn(NewThreadScheduler())
      .subscribe(
        { ev => threadsafeEventBus.onNext(ev) },
        (e: Throwable) => { println("Event 2 producer exception: $e") }
      )

    val l = new CountDownLatch(1)
    println("Event bus values:")
    threadsafeEventBus.take(10).subscribe(
      (ev) => { println(ev) },
      (e: Throwable) => {
        println(s"Event bus exception: $e")
        l.countDown()
      },
      () => { l.countDown() }
    )
    
    l.await(2, TimeUnit.SECONDS)
    
    event1ProducerSubscription.unsubscribe()
    event2ProducerSubscription.unsubscribe()
  }

  @Test def errorExample(): Unit = {
    val o1 = Observable.just(1, 2, 3)
    val o2 = Observable.error(new RuntimeException("Oops"))
    (o1 ++ o2).subscribe(v => println(v), e => e.printStackTrace(), () => println("completed"))
  }

  @Test def concatEagerExample(): Unit = {
    val o1 = Observable.interval(100 millis).take(3).map(l => s"o1 emit $l").doOnSubscribe(println("subscribe to o1"))
    val o2 = Observable.interval(100 millis).take(3).map(l => s"o2 emit $l").doOnSubscribe(println("subscribe to o2"))
    o1.concatEager(o2).subscribe(println(_))
  }

  @Test def concatEagerExample2(): Unit = {
    (0 until 10).map { i =>
      Observable.interval(100 millis).take(3).map(l => s"o$i emit $l").doOnSubscribe(println(s"subscribe to o$i"))
    }.toObservable.concatEager.subscribe(println(_))
  }

  @Test def concatEagerExample3(): Unit = {
    (0 until 10).map { i =>
      Observable.interval(100 millis).take(3).map(l => s"o$i emit $l").doOnSubscribe(println(s"subscribe to o$i"))
    }.toObservable.concatEager(capacityHint = 3).subscribe(println(_))
  }

  @Test def concatMapEagerExample(): Unit = {
    (0 until 10).toObservable.concatMapEager { i =>
      Observable.interval(100 millis).take(3).map(l => s"o$i emit $l").doOnSubscribe(println(s"subscribe to o$i"))
    }.subscribe(println(_))
  }

  @Test def concatMapEagerExample2(): Unit = {
    (0 until 10).toObservable.concatMapEager(capacityHint = 10, i => {
      Observable.interval(100 millis).take(3).map(l => s"o$i emit $l").doOnSubscribe(println(s"subscribe to o$i"))
    }).subscribe(println(_))
  }

  @Test def concatMapEagerExample3(): Unit = {
    (0 until 10).toObservable.concatMapEager(capacityHint = 10, maxConcurrent = 3, i => {
      Observable.interval(100 millis).take(3).map(l => s"o$i emit $l").doOnSubscribe(println(s"subscribe to o$i"))
    }).subscribe(println(_))
  }

  @Test def flattenDelayErrorExample(): Unit = {
    val o1 = Observable.just(1).delay(200 millis).
      flatMap(i => Observable.error(new RuntimeException("Oops!"))).doOnSubscribe(println(s"subscribe to o1"))
    val o2 = Observable.interval(100 millis).map(l => s"o2 emit $l").take(3).doOnSubscribe(println(s"subscribe to o2"))
    val o3 = Observable.interval(100 millis).map(l => s"o3 emit $l").take(3).doOnSubscribe(println(s"subscribe to o3"))
    val o4 = Observable.interval(100 millis).map(l => s"o4 emit $l").take(3).doOnSubscribe(println(s"subscribe to o4"))
    Observable.just(o1, o2, o3, o4).delayError.flatten.subscribe(println(_), _.printStackTrace())
  }

  @Test def flattenDelayErrorExample2(): Unit = {
    val o1 = Observable.just(1).delay(200 millis).
      flatMap(i => Observable.error(new RuntimeException("Oops!"))).doOnSubscribe(println(s"subscribe to o1"))
    val o2 = Observable.interval(100 millis).map(l => s"o2 emit $l").take(3).doOnSubscribe(println(s"subscribe to o2"))
    val o3 = Observable.interval(100 millis).map(l => s"o3 emit $l").take(3).doOnSubscribe(println(s"subscribe to o3"))
    val o4 = Observable.interval(100 millis).map(l => s"o4 emit $l").take(3).doOnSubscribe(println(s"subscribe to o4"))
    Observable.just(o1, o2, o3, o4).delayError.flatten(2).subscribe(println(_), _.printStackTrace())
  }

  @Test def blockingObservableSubscribeExample(): Unit = {
    val b = Observable.just(1, 2, 3).toBlocking
    println("---b.subscribe()---")
    b.subscribe()
    println("---b.subscribe(onNext)---")
    b.subscribe(println(_))
    println("---b.subscribe(onNext, onError)---")
    b.subscribe(println(_), _.printStackTrace())
    println("---b.subscribe(onNext, onError, onCompleted)---")
    b.subscribe(println(_), _.printStackTrace(), () => println("onCompleted"))
    println("---b.subscribe(Observer)---")
    b.subscribe(new Observer[Int] {
      override def onNext(v: Int): Unit = println(v)

      override def onError(e: Throwable): Unit = e.printStackTrace()

      override def onCompleted(): Unit = println("onCompleted")
    })
    println("---b.subscribe(Subscriber)---")
    b.subscribe(new Subscriber[Int] {
      override def onNext(v: Int): Unit = println(v)

      override def onError(e: Throwable): Unit = e.printStackTrace()

      override def onCompleted(): Unit = println("onCompleted")
    })
  }

  def autoConnectExample(): Unit = {
    val o = Observable.just(1, 2, 3).doOnSubscribe {
      println("Start to emit items")
    }.publish.autoConnect
    println("1st Observer is subscribing")
    o.subscribe(println(_))
  }

  def autoConnectExample2(): Unit = {
    val o = Observable.just(1, 2, 3).doOnSubscribe {
      println("Start to emit items")
    }.publish.autoConnect(3)
    println("1st Observer is subscribing")
    o.subscribe(i => println(s"s1: $i"))
    println("2nd Observer is subscribing")
    o.subscribe(i => println(s"s2: $i"))
    println("3rd Observer is subscribing")
    o.subscribe(i => println(s"s3: $i"))
  }

  def concatDelayErrorExample(): Unit = {
    val o1 = Observable.error(new RuntimeException("Oops!"))
    val o2 = Observable.just(1, 2, 3)
    val os = Observable.just(o1, o2)
    os.concat.subscribe(i => println("concat: " + i), e => println("concat: " + e.getMessage))
    os.delayError.concat.subscribe(i => println("concatDelayError: " + i), e => println("concatDelayError: " + e.getMessage))
  }

  def onTerminateDetachExample(): Unit = {
    {
      var o = new Object
      val weakReference = new scala.ref.WeakReference[Object](o)
      // s will have a reference to "o" without onTerminateDetach
      val s = Observable.just(o).size.subscribe(println(_))
      o = null
      System.gc() // It's fine for a demo even if it's just a best effort
      Thread.sleep(2000)
      println(s"without onTerminateDetach: isUnsubscribed=${s.isUnsubscribed}, weakReference=${weakReference.get}")
    }
    {
      var o = new Object
      val weakReference = new scala.ref.WeakReference[Object](o)
      // s won't have a reference to "o" since they are detached.
      val s = Observable.just(o).size.onTerminateDetach.subscribe(println(_))
      o = null
      System.gc() // It's fine for a demo even if it's just a best effort
      Thread.sleep(2000)
      println(s"onTerminateDetach: isUnsubscribed=${s.isUnsubscribed}, weakReference=${weakReference.get}")
    }
  }

  @Test def rebatchRequestsExample(): Unit = {
    val o = (1 to 100).toObservable
    o.doOnRequest(r => println(s"Requesting $r via rebatchRequests"))
      .rebatchRequests(10)
      .doOnRequest(r => println(s"Requesting $r via Subscriber"))
      .subscribe(new Subscriber[Int]() {
        override def onStart(): Unit = request(1)

        override def onNext(value: Int): Unit = {
          println(s"Receive $value")
          request(1)
        }

        override def onError(error: Throwable): Unit = error.printStackTrace()

        override def onCompleted(): Unit = println("Done")
      })
  }
}