Some examples from the Reactive Programming course for discussion at the OSFP meetup.

SwingApi.scala

// How to improve this?
// Any way to get rid of the default case without MatchErrors at runtime on other UI events?

def clicks: Observable[Button] = Observable[Button]((observer: Observer[Button]) => {
  val r = Reaction {
    case ButtonClicked(b) => observer.onNext(b)
    case _ =>
  }
  button.subscribe(r)
  Subscription {
    button.unsubscribe(r)
  }
})

WikipediaApi.scala

/** Given an observable that can possibly be completed with an error, returns a new observable
 * with the same values wrapped into `Success` and the potential error wrapped into `Failure`.
 *
 * E.g. `1, 2, 3, !Exception!` should become `Success(1), Success(2), Success(3), Failure(Exception), !TerminateStream!`
 * 
 * IMPROVE: Implement without creating a new Observable (observer) ?
 */
def recovered: Observable[Try[T]] = Observable[Try[T]]((observer: Observer[Try[T]]) => {
  obs.materialize.subscribe(
      term => term match {
        case OnNext(v) => observer.onNext(Success(v))
        case OnError(e) => {
          observer.onNext(Failure(e))
          observer.onCompleted
        }
        case OnCompleted() =>
      },
      error => observer.onError(error),
      () => observer.onCompleted
  )
  Subscription {}
}) 

WIkipediaSuggest.scala

// I've experienced hangs in this code - after a while button clicks don't lead to new selections and page loads
// Any idea what is going wrong?

// How to implement without channels?

val selections: Observable[String] = {
  val channel = PublishSubject[String]("")
  button.clicks.subscribe { _ =>
    val selected = suggestionList.selection.items.toList.headOption
    selected.map { s => println("Selected: " + s); channel.onNext(s) }
  }
  channel
}
  
val pages: Observable[Try[String]] = {
  val channel = PublishSubject[Try[String]](Success(""))
  selections.sanitized.subscribe { term =>
    println("Getting wikipedia page for term: " + term)
  	val f = wikipediaPage(term)
  	f.onComplete(t => channel.onNext(t))
  }  
  channel
}