text:
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")
      })
  }
}


text:
Home›Lost Decade›Colt Brennan's family and coaches talk about what was lost
Colt Brennan's family and coaches talk about what was lost
We used 2,500 words on Sunday to detail the life and death of Colt Brennan, who grew up in Orange County and became a star quarterback at the University of Hawaii.
He died on May 11 after authorities said he overdosed on something containing fentanyl.
What follows is the best of what didn't make history.
Trying to help their son
Brennan's parents have both said they believe they've done everything they can to help their son, whose battle with drug addiction has lasted for more than a decade.
"We're pretty comfortable with it," said her dad, Terry. "It was one of those insistence that Colt just pushed him, like other kids do." They keep growing, and once you get to that spot, those are the consequences. "
Also from Terry: "We knew he was struggling and his activities showed he was struggling. Everything we tried kind of blew up on us.
A battle in progress
Brennan's behavior – fueled by her drug addiction – often alienated others, especially those close to her.
"He would burn bridges, with friends and coaches," said Terry. "They would become the enemy, in a way, in his mind."
The attacks often took place on social media and by text message. Brennan has apologized repeatedly and profusely for her actions.
"I know I have written horrible texts in the past and I am sorry for that," he expressed in a letter to his family last year. "It was the most embarrassing thing I have done throughout this battle."
His mother, Betsy, said, "He loved us. I mean, he was still sorry for the way he acted sometimes. This is what is so difficult. It had to be something else, something bigger, the demons hitting on him. … He hated it when he got like that. He would say mean things. He couldn't help it.
Find peace
Colt Brennan on Waikiki Beach with Diamond Head Mountain in the background on August 16, 2007.
(Lucy Pemoni / Getty Images)
Brennan's death was a tragic end that brought sadness to those around her but also needed a finality.
"He's at peace," said Terry. "He no longer has to fight battles.
Betsy added, "I think he's at peace because I think he fought so hard just to be happy. We don't know how he felt, how much weight may have been on him all the time. You just don't know.
Former Hawaii assistant coach Rich Miano called it "almost a sense of relief" when Brennan passed away.
"For those [of] you [who] I loved Colt, those who were closest to him, those who tried so hard to change him, influence him and help him, at least it's closing. They no longer have to go to bed thinking about how it's not going to end well. …
He's in a better place. It couldn't be in a worse place.
Everyone's friend
Some of the things Brennan will be remembered for is her easy smile and just as easy demeanor.
"Colt wasn't nervous about a lot of things," said Bruce Rollinson, his coach at Mater Dei High in Santa Ana. "He was a laid back kid with a smile, his big old smile."
Rollinson recalled watching Brennan walk through the halls of Mater Dei exuding the charm that would mark so much of her life.
"Colt was a friend to everyone," Rollinson said. "I used to laugh to myself. I was going to say 'Look at him, man.' It was as if he was carefree, loving every day of his life. And that's what's so hard about it now.
Trying to find answers
Colt Brennan sits on the bench with his teammates during a loss to Georgia at the 2008 Sugar Bowl in New Orleans.
(Matthieu Stockman / Getty Images)
The Brennans continue to question their son's fate, including the impact their attempts at hard love had or did not have on him.
"Friends said to me, 'How can you do that? And some say, 'You're doing the right thing,' "Betsy explained. "You're caught in the middle. You try to have hard sex, but then you want to help them. I mean, he's your son. It is so difficult as a parent.
Signs of hope
As far as Brennan's parents know, he was never diagnosed with depression, although they concluded he was struggling with that as well. He sometimes tried drugs, but always stopped taking them because of how he felt.
"That's what upset us because he would never give her enough time to work," Betsy said.
When asked if, given all of what their son was up against, the Brennans thought it was inevitable that he would die the way he did, Betsy explained that recent events had altered. their perspective.
Brennan had spent four months in a home treatment center in Costa Mesa. His visits to similar places had never lasted more than a few weeks.
"We did at one point [think this was inevitable]"Betsy said." But I think in the last few months… because of this last treatment center, we thought he had made it, turned the corner.… The facility he was in told us that ' he was always positive, always optimistic, they praised him.
Recognizable name
Colt Brennan and his mother, Betsy, after a game in Hawaii.
(Brennan family)
Brennan's parents named him Colt after his father looked at a map of Southern California and noticed "Colton," a town of about 55,000 people in San Bernardino County.
Colton James Brennan was born August 16, 1983. James was one of Brennan's grandfathers. He died of ALS.
Brennan's Hawaii head coach June Jones said that after Brennan's death he received messages from "people in New York, South Carolina and Texas watching every game. [on late-night TV]. … He captivated all of Hawaii and half of the continent. Everyone knows this name.
Trying to stay clean
Betsy on times when Brennan was able to stay sober: "What's really crazy is all those years of drinking and whatever he does… when he was sober his mind was clear. He was funny. He liked to tell stories. He could recite lines from movies. That's why I think it had to be something bigger.
Hawaii Gifts
In addition to the memories, Brennan left her parents with two cats, both of whom he brought back from Hawaii: Toa and Kona.
Rent others
Miano recalled coming home from Rainbow Warriors games listening to the radio and marveling at Brennan's interviews.
In addition to displaying remarkable "charisma," said Miano, Brennan has always congratulated Hawaii, its coaches and teammates, especially its offensive linemen and receivers.
"You couldn't have asked a media consultant to describe him better to say the right things," Miano said. "It was like, 'My God, this guy is a media genius. He just understands. "
What was lost
Colt Brennan, second from right, poses for a photo with his father Terry, sister Chanel, center, and mother Betsy.
Betsy: "We've had a lot of him for the past four months. He had become himself again. We were doing really well. It was the best I had seen. The sparkle in his eyes was back.
Jones: "We were hoping he would hit rock bottom and run away before that happened. … Everyone tried to help him. Other players, teammates, coaches. Unfortunately, the demons were just too much to deal with. … These drugs, these different things, they'll kill you in the end.
Miano: "We have lost a beautiful person. There will never be another Colt Brennan in Hawaiian history. … People say he was the greatest player of all time in Hawaii. How do you argue this? It was."
Offline: it's time to ask questions and learn lessons
Getting misplaced is the entire thought on our "It is Something Can Occur Day" adventures | Tradition & Leisure
Opinion | Is that this the tip of French mental life?
Decade of conflict in Syria prices 1.55 trillion: World Imaginative and prescient report
Integrate ESG for a targeted strategic transformation, ...
China responds to Canada, calls for UN ...
What are the challenges for making more fair economies?
South Sudan's lost decade calls for impact investments
Trump's Washington hotel lost over $ 70 million while he was president
Apple AirTag bug allows "Good Samaritan" attack – Krebs on Security
Sweden's first female prime minister returns days after resigning
Weston reclaims SWC men's tennis title to continue decade of domination
Alternet Systems Inc. (OTCMKTS: ALYI) The stock is gaining ground: how to trade now?
Why Covid shattered parents' sense of risk
Cryptocurrency Bubble Risks Exposed By Recent Bitcoin Slide

redpajama_set_name: RedPajamaCommonCrawl


Children might find that school doesn't suit their specific approach of learning, so dad and mom who develop into unhappy with the education which is offered in faculties may additionally determine to dwelling educate. In April 2016, MCEC and Second Woman Biden marked the 5-year anniversary of Operation Educate the Educators by convening a group of the partnership's energetic signatories in Washington, DC, to reflect on classes discovered, finest practices, and strategies for transferring the work forward.
Educate Professional has superior performance to alter house page components using Easy Drag-n-Drop ! But they do not have to be boxed up in a room crammed with books, pencils, and whiteboards to be educated particularly with the certain age and developmental stage they are in.
Texting and Fb are uncontrolled, children watch method too many R-rated movies, One Path is worshiped by 10-yr-outdated-women, and Michael Jackson died 3 half years in the past. Although I suppose the real problem shouldn't be crucial the fact that some choose ineffective degrees but that they suppose that their masters in ineffective it higher than someone elses BA. We attended the Cultural Survival Bazaar as a vendor, which Educate Malawi has executed for a couple of years now, and Mathew was able to attend with us. Above he is seen speaking with a customer in regards to the organization and about Malawi. The Kaira's have been very helpful to Educate Malawi in sharing their expertise and information about Malawi and the United States. Nationally, Ashford College has obtained $110 million to educate Iraq and Afghanistan veterans in the final 5 years.
The Latin phrase ducere, which means to guide," and its form ductus give us the roots duc and duct. She acknowledged that these weren't her goals however that she didn't really know what an well educated youngster could be for her. My aim was easy: Amass, synthesize, and current knowledge on what it means to be smart, educated and clever in order that it can be understood and utilized by anyone for his or her profit. The Educate Women DIB program works with 166 faculties across one hundred forty villages in Bhilwara, India, and goals to learn 15,000 youngsters!
The thesaurus of synonyms and related phrases is totally integrated into the dictionary. Educate Pro has awesome performance to change residence page parts using Easy Drag-n-Drop ! However they don't have to be boxed up in a room full of books, pencils, and whiteboards to be educated especially with the certain age and developmental stage they are in.
The woman with the BA in Sociology won't date the carpenter (assuming that hypergamy holds true) since she's higher up the formal schooling ladder, although chances are high that he is really more educated than her (in the training that matters – I'll go on file: I consider most blue collar workers are more educated than the occupy wall street crowd – and I say this wearing a white collar).
We act in a rights-based method to help kids overcome obstacles to training. Educate Professional has superior functionality to alter dwelling web page components using Easy Drag-n-Drop ! However they don't have to be boxed up in a room full of books, pencils, and whiteboards to be educated especially with the sure age and developmental stage they are in.
Mathew Kasonjola, a young man from Malawi, has arrived today in America for a go to with Educate Malawi and a few of our good friends and beneficiant sponsors and those who assist us! Working with one hundred seventy five partnership colleges across the UK, Educate presents school-led routes to Certified Trainer Standing (QTS). Final comes the most important half – INSTRUCTING; I suppose the one tense applicable for this verb is 'present steady'. From the very begin, we have considered and constructed Educate as an entire new product, not merely a brand new feature.
Michigan State University will host the 15th annual Graduate Pupil Topology and Geometry Conference on April 8-April 9, 2017. An alternate view is that sure scientific fields (akin to theoretical physics ) are mathematics with axioms which might be intended to correspond to actuality. Because the profession goals of undergraduate mathematics majors are so various, each undergraduate's program is individually organized via collaboration between the student and his or her faculty advisor. They had been acknowledged for excellent instructing and/or advising practices inside the faculty.
Dame Zaha Hadid skilled first in mathematics earlier than finding out structure, and the work of her observe is strongly knowledgeable by ideas about geometry. The American Mathematics Competitions are a series of examinations and curriculum supplies that build drawback-solving expertise and mathematical information in center and high school college students. Mathematical Structures , checklist details about lessons of mathematical constructions.
During the last 15 years from my talking, working in faculties and observing lecturers, I have learned that the number one obstacle that stops the academics from creating an engaged studying classroom is the beliefs of students. In April 2016, MCEC and Second Girl Biden marked the 5-12 months anniversary of Operation Educate the Educators by convening a bunch of the partnership's active signatories in Washington, DC, to mirror on lessons discovered, greatest practices, and strategies for shifting the work ahead.
Educate Pro has superior functionality to vary home web page elements using Simple Drag-n-Drop ! But they do not have to be boxed up in a room full of books, pencils, and whiteboards to be educated especially with the sure age and developmental stage they're in.
With Educate you can create a public help heart where customers can discover solutions on their own, and your assist staff can shortly search for and share help articles immediately in conversations with clients. The aim is to create a robust community of best practice which helps the members of Educate Plus.
When making a purchase order on an item for the first time few persons are capable of make an educated choice. This may be helpful as it'll help to educate children prepared for when they're older and would possibly want to have a profession in finance or in the event that they want to start up a business. A socially motivated investor – the UBS Optimus Foundation (UBSOF) places within the working capital, so the service supplier Educate Women can start its work on the bottom.
Educate Pro has awesome performance to alter home web page components utilizing Easy Drag-n-Drop ! However they don't have to be boxed up in a room stuffed with books, pencils, and whiteboards to be educated particularly with the certain age and developmental stage they're in.
My final phrases of advice to you might be educate, agitate and organize; place confidence in your self. The Krause Innovation Studio and EDUCATE would like to announce that they are persevering with their Innovation Talks: Educating First, Technology Second" series this spring 2015. From the brief description above, it is clear that a good father or mother, must educate their kids at home than ship them to a proper college. After many journeys by Shannon, Stephen and I to Malawi, Mathew has come to Malawi to study Educate Malawi in America and what it means to fundraise to accomplish what we have now in these handed 6 years. Educate Professional comes with several awesome Google fonts choices to make your website look trendy. This blog was impressed and created because of the non-profit orgnanization, Educate Malawi, Inc.


Going for a walk in the woods may lead to an aha! moment. Many people figure out creative, new ways to solve problems by allowing their minds to wander.
Ask most people to identify a creative person, and they'll probably describe an artist — Picasso, Shakespeare or even Lady Gaga.
But what about a Nobel prize–winning chemist? Or a team of engineers that figures out how to make a car engine operate more efficiently?
Creativity, it turns out, is not only the domain of painters, singers and playwrights, says Robert DeHaan, a retired Emory University cell biologist who now studies how to teach creative thinking.
That can mean composing a piece of music that's pleasing to the ear or painting a mural on a city street for pedestrians to admire. Or, DeHaan says, it can mean dreaming up a solution to a challenge encountered in the lab.
But creative thinking, DeHaan and others say, isn't always the focus of teaching in science classrooms.
"A lot of kids think that science is a body of knowledge, a collection of facts they need to memorize," says Bill Wallace, a science teacher at Georgetown Day School in Washington, D.C.
That approach to learning about science, however, emphasizes only facts and concepts. It leaves little room for the creative thinking central to science, Wallace says.
"If instead, you teach science as a process of learning, of observing and of gathering information about the way that nature works, then there's more room for incorporating creativity," Wallace says.
And the approach to a scientific question and its analysis provide additional avenues for creativity.
"In the best science investigations, it's not the questions that are most creative, but rather how the experiment is measured and how the data are interpreted, given meaning and how students see the investigation as a component in understanding a scientific problem," says Carmen Andrews, a science specialist at Thurgood Marshall Middle School in Bridgeport, Conn.
Indeed, scientists themselves describe science not as a set of facts and vocabulary to memorize or a lab report with one "right" answer, but as an ongoing journey, a quest for knowledge about the natural world.
In the quest to make sense of the natural world, scientists think of new ways to approach problems, figure out how to collect meaningful data and explore what those data could mean, explains Deborah Smith, an education professor at Penn State University in State College, Penn.
In other words, they develop ideas that are both new and useful — the very definition of creativity.
Imagining possibilities requires people to use what scientists who study how the brain works call "associative thinking." This is a process in which the mind is free to wander, making possible connections between unrelated ideas.
The process runs counter to what most people would expect to do when tackling a challenge. Most would probably think the best way to solve a problem would be to focus on it — to think analytically — and then to keep reworking the problem.
In fact, the opposite approach is better, DeHaan argues. "The best time to come to a solution to a complex, high-level problem is to go for a hike in the woods or do something totally unrelated and let you mind wander," he explains.
When scientists allow their minds to roam and reach beyond their immediate research fields, they often stumble onto their most creative insights — that "aha" moment, when suddenly a new idea or solution to a problem presents itself.
Herschbach, for example, made an important discovery in chemistry shortly after he learned of a technique in physics called molecular beams. This technique allows researchers to study the motion of molecules in a vacuum, an environment free of the gas molecules that make up air.
Physicists had been using the technique for decades, but Herschbach, a chemist, hadn't heard of it before — nor had he been told what couldn't be done with crossed molecular beams. He reasoned that by crossing two beams of different molecules, he might learn more about how quickly reactions occur as molecules collide with one another.
Initially, Herschbach says, "People thought it would not be feasible. It was called the lunatic fringe of chemistry, which I just loved." He ignored his critics, and set out to see what would happen if he crossed a beam of molecules such as chlorine with a beam of hydrogen atoms.
He spent several years collecting his data, which in the end uncovered new insights into the ways colliding molecules behave. It was an important enough advance in chemistry that in 1986 Herschbach and a colleague were awarded science's top honor: the Nobel Prize.
Herschbach makes an important point. Naïveté — a lack of experience, knowledge or training — can actually be a boon to finding creative insights, DeHaan says. When you're new to a scientific field, he explains, you're less likely to have learned what other people claim is impossible. So you come to the field fresh, without any expectations, sometimes called preconceptions.
In the classroom, broadening your thinking can mean emphasizing something called problem-based learning. In this approach, a teacher presents a problem or question with no clear or obvious solution. Students are then asked to think broadly about how to solve it.
Problem-based learning can help students think like scientists, Wallace says. He cites an example from his own classroom. Last fall, he had students read about fruit flies that lack an enzyme — a molecule that speeds up chemical reactions — to break down alcohol.
He asked his students to find out whether these flies would feel the effects of alcohol, or even become inebriated, sooner than would flies that possess the enzyme.
Talking with others and teamwork also help with associative thinking — allowing thoughts to wander and freely associating one thing with another — that DeHaan says contributes to creativity. Working on a team, he says, introduces a concept called distributed reasoning. Sometimes called brainstorming, this type of reasoning is spread out and conducted by a group of people.
"It's been known or thought for a long time that teams generally are more creative than individuals," DeHaan explains. While researchers who study creativity don't yet know how to explain this, DeHaan says it could be that by hearing different ideas from different people, members of a team begin to see new connections between concepts that didn't initially seem related.
Asking questions such as, "Is there some way to pose the problem other than the way it was presented?" and "What are the parts of this problem?" also can help students stay in this brainstorming mode, he says.
Smith cautions against confusing artistic or visual representations of science with scientific creativity.
Although using arts and crafts to represent ideas can be helpful, Smith says, it is not the same as recognizing the creativity inherent in science. "What we've been missing is that science itself is creative," she explains.
"It's a creativity of ideas and representations and finding things out, which is different from making a papier-mâché globe and painting it to represent the Earth," she says.
In the end, educators and scientists agree that anyone can learn how to think like a scientist. "Too often in school, students get the impression that science is for a specially gifted subspecies of humanity," Herschbach says. But he insists just the opposite is true.


Mike Sheehan is Professor of International Relations at the Swansea University, the co-Director of ISCAS, and Director of the Callaghan Centre for the Study of Conflict.
He previously worked at the University of Aberdeen, where he was Director of the Scottish Centre for International Security, and at the International Institute for Strategic Studies in London. He joined the Department in 2004. His areas of expertise are critical security, military space, balance of power theory, NE Asian security and indigenous security issues in the European Arctic.
David is a Professor of International Relations and Public Policy within the Department of Political and Cultural Studies. He is also Director of the Global Drug Policy Observatory, based at Swansea University. He has been researching various aspects of drug policy for over twenty years with his main areas of interest being US drug policy, the UN and international drug policy and more recently counter narcotic strategies in Afghanistan. David has published widely on many of these and other topics, has collaborated with and produced policy reports for a range of drug policy organizations beyond academia and at present is an Associate of the International Drug Policy Consortium and a Associate Fellow of the Transnational Institute's Drugs and Democracy Programme. He is editor-in-chief of the GDPO Policy Reports, Policy Briefs and Situation Analyses.
View Professor Bewley-Taylor's full Swansea University staff profile here.
Mark Bevir is a Professor of Political Science and Director of the Center for British Studies, University of California, Berkeley. He is also a Professor in the Graduate School of Governance, United Nations University (MERIT), and a Distinguished Research Professor in the College of Arts and Humanities, Swansea University.
Born and raised in London, Mark received his doctorate from the University of Oxford. Before moving to Berkeley, he held posts at the University of Madras, India, and the University of Newcastle, UK. He has held visiting fellowships in Australia, Finland, France, Germany, Italy, Norway, the UK, and the US. He has done policy work for governmental and non-governmental organizations in Europe and the US as well as the United Nations.
Mark's research interests in political theory include moral philosophy, political philosophy, and the history of political thought. His work on public policy focuses on organization theory, democratic theory, and governance. His methodological interests cover philosophy of social science, history of social science, and interpretive theory.
Gerard joined the university in 1994 while completing his PhD, and was based in the Centre for Development Studies for 16 years until its closure in 2010. He is now Associate Professor in Politics and International Development, and Head of the Department of Political and Cultural Studies at Swansea. While in CDS, Gerard worked on consultancy assignments for a range of organizations including the Department for International Development, the World Bank and the National Audit Office. Prior to that, he taught at Middlesex Universityand was a Visiting Research Fellow at the University of the Philippines in Manila. Gerard completed his PhD on Non-Governmental Organisations and Philippine Politics at the School of Oriental and African Studies in London, after completing a masters degree in Comparative Government at the London School of Economics.
Emel conducts research on state, societal and urban transformations at the age of globalisation, social uprisings, forced migration, refugee governmentality and conflict transformation/s. She is especially interested in the wider Mediterranean geography but, likes to compare this region with other parts of the world, as well. More specifically, Emel works on the refinement of the concepts of subjectivity, hybridity, governmentality, resistance. She is interested in finding out what type of subjectivities and structures emerge as a result of various forms of local and global governmentalities. Emel's work has been published in Political Geography, Security Dialogue, Eurasian Geography and Economics, Antipode, Annals of the American Geographers, South European Society and Politics, Geopolitics and Asian Journal of Social Sciences.
Dr Luca Trenta is a Senior lecturer in International Relations in the Department of Political and Cultural Studies, Swansea University. He previously worked as Teaching Associate at the University of Nottingham. Dr Trenta received his PhD from Durham University in 2014. He is the holder of a 2017 British Academy Rising Star Engagement Award for his project: 'Out of the Shadows: understanding, researching and teaching covert action.' He previously received a British Academy Small Research grant for his project on 'Targeted killing? The recurrence of assassination is US foreign policy.' He has published a monograph on risk and presidential decision-making with Routledge. He has also published articles on US foreign policy and presidential decision-making in The European Journal of International Security, Diplomacy and Statecraft and in The Journal of Transatlantic Studies. His research interests include US foreign policy, intelligence, covert action, assassination, and the use of drones. He is a regular contributor to The Conversation, Talk Radio and BBC Radio Wales.
Stephen McVeigh is Associate Professor in War and Society and Director of Learning and Teaching for the College of Arts and Humanities. His areas of research are diverse and include American War and Society, 19th and 20th century American cultural history, American film, 20th century American literature and American pulp fictions. He delivers teaching in topics as diverse as Frontier mythology in 20th century American history and culture, the Spanish Civil War, total war in the modern era, international perspectives on propaganda film, American masculinities, the 'American Way of War', terrorism and culture, representations of war in art, literature and film and he also contributes to a number of modules in undergraduate and postgraduate programmes across the College of Arts and Humanities.
Dr. McVeigh is the series editor of War, Culture and Society, a research monograph series published by Bloomsbury. He is also a member of the editorial team of the Journal of War and Culture Studies.
Krijn is Associate Professor in the department, having joined in 2010 after serving as a lecturer at the Centre for Development Studies, Swansea University since 2005. He obtained his PhD at Wageningen University, the Netherlands, undertaking an investigation into armed conflict and post-war reintegration trajectories of youthful ex-combatants in Sierra Leone, West Africa. His BSc and MSc were in rural development sociology, again at Wageningen University, for which he undertook field work in Sierra Leone and Cambodia. In 2000 Krijn worked a year for Save the Children UK evaluating their ex-child soldier reintegration project in Liberia. He has provided consultancy services for a number of international organisations, including the World Bank, Plan International, the Institute for Security Studies and the Dutch Royal Institute of the Tropics.
Martin is a Research Officer and Crypto-Drug Markets project lead with the Global Drug Policy Observatory. Martin's Ph.D. is entitled 'Crypto-Drug Markets and the Global Drug Prohibition Regime: A Unique Threat?' The study investigates the transnational crypto-drug market phenomena and the associated challenges for drug control policy, including at the domestic and international levels. Martin's research operates within - and contributes to - international relations regime theory.He holds an MA (Distinction) in International Politics from Hull, and a BScEcon (Hons) in International Politics and Strategic Studies from Aberystwyth.


To help understand some key trends about what millennials believe and why it matters for banks, we asked Kristen Soltis Anderson, a pollster and expert in millennial public opinion and behavior, to highlight some top trends.
On the other hand, they're pretty optimistic and creative, and they want to be able to blur the lines between their personal lives and their work lives. The lines between work and play have blurred a lot for millennials, and that's something for people who work in the financial sector to know. At the margins, you're seeing a slightly different set of values when millennials ask, "What does fulfillment in life mean to me, and how can I make my career support that?" Family is hugely important. It's defined a little more flexibly than previous generations, but this is a very pro-family generation. This is a generation that puts a lot of pressure on themselves—you've got to thrive at home, you've got to thrive at work and you've got to put it all together.
Kristen Soltis Anderson is a co-founder of Echelon Insights, a columnist for the Washington Examiner and the author of The Selfie Vote: Where Millennials Are Leading America.