recoverWith.scala

def recoverWith[T](source: Observable[T], pf: PartialFunction[Throwable, Observable[T]]) =
  Observable.create[T] { subscriber =>
    implicit val s = subscriber.scheduler
    val o = subscriber.observer

    source.unsafeSubscribe(new Observer[T] {
      def onNext(elem: T) =
        o.onNext(elem)

      def onError(ex: Throwable) = {
        if (pf.isDefinedAt(ex))
          pf(ex).unsafeSubscribe(o)
        else
          o.onError(ex)
      }

      def onComplete() =
        o.onComplete()
    })
  }