src/main/scala/rx/lang/scala/Notification.scala

@@ -64,9 +64,11 @@ sealed trait Notification[+T] {
 
   def apply(observer: Observer[T]): Unit = accept(observer)
 
-  def map[U](f: T => U): Notification[U] = {
+  def map[U](f: T => U): Notification[U] = flatMap(t => Notification.OnNext(f(t)))
+
+  def flatMap[U](f: T => Notification[U]): Notification[U] = {
     this match {
-      case Notification.OnNext(value) => Notification.OnNext(f(value))
+      case Notification.OnNext(value) => f(value)
       case Notification.OnError(error) => Notification.OnError(error)
       case Notification.OnCompleted => Notification.OnCompleted
     }

src/main/scala/rx/lang/scala/package.scala

@@ -25,21 +25,22 @@ import _root_.scala.concurrent.{ExecutionContext, Future}
  */
 package object scala {
 
-  @deprecated("Use IterableToObservable", "0.26.5")
-  type ObservableExtensions[T] = IterableToObservable[T]
-
-  implicit class IterableToObservable[T](val iterable: Iterable[T]) extends AnyVal {
-    def toObservable: Observable[T] = Observable.from(iterable)
+  /** Adds a [[toObservable]] extension method to [[_root_.scala.collection.Iterable Iterable]] */
+  implicit class ObservableExtensions[T](val source: Iterable[T]) extends AnyVal {
+    def toObservable: Observable[T] = Observable.from(source)
   }
 
+  /** Adds a [[toObservable]] extension method to [[_root_.scala.util.Try Try]] */
   implicit class TryToObservable[T](val tryT: Try[T]) extends AnyVal {
     def toObservable: Observable[T] = Observable.from(tryT)
   }
 
+  /** Adds a [[toObservable]] extension method to [[_root_.scala.Option Option]] */
   implicit class OptionToObservable[T](val opt: Option[T]) extends AnyVal {
     def toObservable: Observable[T] = Observable.from(opt)
   }
 
+  /** Adds a [[toObservable]] extension method to [[_root_.scala.concurrent.Future Future]] */
   implicit class FutureToObservable[T](val future: Future[T]) extends AnyVal {
     def toObservable(implicit ec: ExecutionContext): Observable[T] = Observable.from(future)
   }

src/test/scala/rx/lang/scala/NotificationTests.scala

@@ -55,4 +55,48 @@ class NotificationTests extends JUnitSuite {
       assertEquals(13, onCompleted(x=>42, e=>4711,()=>13))
 
   }
+
+  @Test
+  def TestFlatMapNextToNext() {
+    val notification = OnNext(41).flatMap(i => Notification.OnNext(i+1))
+    assertEquals(42, notification(i=>i, _ => -1, () => -1))
+  }
+
+  @Test
+  def TestFlatMapNextToError() {
+    val oops = new Exception("Oops")
+    val notification = OnNext(()).flatMap(_ => Notification.OnError(oops))
+    assertEquals(42, notification(_ => -1, {
+      case `oops` => 42
+      case _ => -1
+    }, () => -1))
+  }
+
+  @Test
+  def TestFlatMapNextToCompletion() {
+    val notification = OnNext(()).flatMap(_ => Notification.OnCompleted)
+    assertEquals(42, notification(_ => -1, e => -1, () => 42))
+  }
+
+  @Test
+  def TestFlatMapCompleted() {
+    val notification = OnCompleted.flatMap(_ => Notification.OnNext(1))
+    assertEquals(42, notification(_ => -1, e => -1, () => 42))
+  }
+
+  @Test
+  def TestFlatMapError() {
+    val oops = new Exception("Oops")
+    val notification = OnError(oops).flatMap(_ => Notification.OnNext(1))
+    assertEquals(42, notification(_ => -1, {
+      case `oops` => 42
+      case _ => -1
+    }, () => -1))
+  }
+
+  @Test
+  def map() {
+    val notification = OnNext(41).map(_+1)
+    assertEquals(42, notification(i=>i, _ => -1, () => -1))
+  }
 }