Scala Observables fix (#787)

Should handle errors from user functions gracefully

JAVA-4304

Author: rozza

driver-scala/src/main/scala/org/mongodb/scala/internal/FilterObservable.scala

@@ -18,6 +18,8 @@ package org.mongodb.scala.internal
 
 import org.mongodb.scala.{ Observable, Observer, Subscription }
 
+import scala.util.{ Failure, Success, Try }
+
 private[scala] case class FilterObservable[T](observable: Observable[T], p: T => Boolean) extends Observable[T] {
   override def subscribe(observer: Observer[_ >: T]): Unit = {
     observable.subscribe(
@@ -43,10 +45,14 @@ private[scala] case class FilterObservable[T](observable: Observable[T], p: T =>
           }
 
           override def onNext(tResult: T): Unit = {
-            if (p(tResult)) {
-              observer.onNext(tResult)
-            } else if (!terminated) {
-              subscription.foreach(_.request(1)) // No match, request more from down stream
+            Try(p(tResult)) match {
+              case Success(result) =>
+                if (result) {
+                  observer.onNext(tResult)
+                } else if (!terminated) {
+                  subscription.foreach(_.request(1)) // No match, request more from down stream
+                }
+              case Failure(exception) => onError(exception)
             }
           }
         }

driver-scala/src/main/scala/org/mongodb/scala/internal/FlatMapObservable.scala

@@ -19,6 +19,7 @@ package org.mongodb.scala.internal
 import org.mongodb.scala._
 
 import java.util.concurrent.atomic.AtomicReference
+import scala.util.{ Failure, Success, Try }
 
 sealed trait State
 case object Init extends State
@@ -78,11 +79,19 @@ private[scala] case class FlatMapObservable[T, S](observable: Observable[T], f:
           }
 
           override def onError(throwable: Throwable): Unit = {
+            state.set(Error)
             observer.onError(throwable)
           }
 
           override def onNext(tResult: T): Unit = {
-            f(tResult).subscribe(
+            Try(f(tResult)) match {
+              case Success(result)    => onNextMappedObservable(result);
+              case Failure(exception) => onError(exception)
+            }
+          }
+
+          private def onNextMappedObservable(mappedObservable: Observable[S]): Unit = {
+            mappedObservable.subscribe(
               new Observer[S]() {
                 override def onError(throwable: Throwable): Unit = {
                   state.set(Error)

driver-scala/src/main/scala/org/mongodb/scala/internal/FoldLeftObservable.scala

@@ -17,9 +17,10 @@
 package org.mongodb.scala.internal
 
 import java.util.concurrent.atomic.AtomicBoolean
-
 import org.mongodb.scala.{ Observable, Observer, SingleObservable, Subscription }
 
+import scala.util.{ Failure, Success, Try }
+
 private[scala] case class FoldLeftObservable[T, S](observable: Observable[T], initialValue: S, accumulator: (S, T) => S)
     extends SingleObservable[S] {
 
@@ -55,7 +56,10 @@ private[scala] case class FoldLeftObservable[T, S](observable: Observable[T], in
           }
 
           override def onNext(tResult: T): Unit = {
-            currentValue = accumulator(currentValue, tResult)
+            Try(accumulator(currentValue, tResult)) match {
+              case Success(result)    => currentValue = result;
+              case Failure(exception) => onError(exception)
+            }
           }
         }
       )

driver-scala/src/main/scala/org/mongodb/scala/internal/MapObservable.scala

@@ -18,6 +18,8 @@ package org.mongodb.scala.internal
 
 import org.mongodb.scala.{ Observable, Observer, Subscription }
 
+import scala.util.{ Failure, Success, Try }
+
 private[scala] case class MapObservable[T, S](observable: Observable[T], s: T => S, f: Throwable => Throwable = t => t)
     extends Observable[S] {
   override def subscribe(observer: Observer[_ >: S]): Unit = {
@@ -30,7 +32,12 @@ private[scala] case class MapObservable[T, S](observable: Observable[T], s: T =>
 
           override def onComplete(): Unit = observer.onComplete()
 
-          override def onNext(tResult: T): Unit = observer.onNext(s(tResult))
+          override def onNext(tResult: T): Unit = {
+            Try(s(tResult)) match {
+              case Success(result)    => observer.onNext(result);
+              case Failure(exception) => observer.onError(exception)
+            }
+          }
         }
       )
     )

driver-scala/src/test/scala/org/mongodb/scala/internal/FlatMapObservableTest.scala

@@ -35,15 +35,36 @@ class FlatMapObservableTest extends BaseSpec with Futures with Eventually {
         _ => (),
         e => p.failure(e),
         () => {
-          completedCounter.incrementAndGet()
-          Thread.sleep(100)
           p.trySuccess(())
+          completedCounter.incrementAndGet()
         }
       )
     eventually(assert(completedCounter.get() == 1, s"${completedCounter.get()}"))
-    Thread.sleep(200)
-    assert(completedCounter.get() == 1, s"${completedCounter.get()}")
-    Thread.sleep(1000)
+  }
+
+  it should "call onError if the mapper fails" in {
+    val p = Promise[Unit]()
+    val errorCounter = new AtomicInteger(0)
+    Observable(1 to 100)
+      .flatMap(
+        x =>
+          if (x > 10) {
+            throw new IllegalStateException("Fail")
+          } else {
+            createObservable(x)
+          }
+      )
+      .subscribe(
+        _ => (),
+        _ => {
+          p.trySuccess()
+          errorCounter.incrementAndGet()
+        },
+        () => {
+          p.failure(new IllegalStateException("Should not complete"))
+        }
+      )
+    eventually(assert(errorCounter.get() == 1, s"${errorCounter.get()}"))
   }
 
   private def createObservable(x: Int): Observable[Int] = new Observable[Int] {

driver-scala/src/test/scala/org/mongodb/scala/internal/ObservableImplementationSpec.scala

@@ -59,7 +59,17 @@ class ObservableImplementationSpec extends BaseSpec with TableDrivenPropertyChec
     forAll(failingObservables) { (observable: Observable[Int]) =>
       {
         var thrown = false
-        observable.subscribe((res: Int) => (), (t: Throwable) => thrown = true)
+        observable.subscribe(_ => (), _ => thrown = true)
+        thrown should equal(true)
+      }
+    }
+  }
+
+  it should "be well behaved when errors are caused by passed in function and call onError" in {
+    forAll(failingFunctionsObservables) { (observable: Observable[Int]) =>
+      {
+        var thrown = false
+        observable.subscribe(_ => (), _ => thrown = true)
         thrown should equal(true)
       }
     }
@@ -208,7 +218,7 @@ class ObservableImplementationSpec extends BaseSpec with TableDrivenPropertyChec
 
   val failOn = 30
 
-  def failingObservables =
+  private def failingObservables =
     Table(
       "observable",
       TestObservable[Int](failOn = failOn),
@@ -231,6 +241,35 @@ class ObservableImplementationSpec extends BaseSpec with TableDrivenPropertyChec
       ZipObservable[Int, Int](TestObservable[Int](failOn = failOn), TestObservable[Int]()).map[Int](a => a._1)
     )
 
+  private def failingFunctionsObservables =
+    Table(
+      "observable",
+      FilterObservable[Int](TestObservable[Int](), (i: Int) => {
+        if (i > 10) {
+          throw new RuntimeException("Error")
+        }
+        i % 2 == 0
+      }),
+      FlatMapObservable[Int, Int](TestObservable[Int](), (i: Int) => {
+        if (i > 10) {
+          throw new RuntimeException("Error")
+        }
+        TestObservable[Int](1 to 2)
+      }),
+      FoldLeftObservable(TestObservable[Int](1 to 100), 0, (v: Int, i: Int) => {
+        if (i > 10) {
+          throw new RuntimeException("Error")
+        }
+        v + i
+      }),
+      MapObservable[Int, Int](TestObservable[Int](), (i: Int) => {
+        if (i > 10) {
+          throw new RuntimeException("Error")
+        }
+        i * 100
+      })
+    )
+
   private def happyObservables =
     Table(
       ("observable", "observer"),