add mergeDelayError and flattenDelayError

Author: samuelgruetter

language-adaptors/rxjava-scala/src/main/scala/rx/lang/scala/Observable.scala

@@ -1271,6 +1271,29 @@ class Observable[+T](val asJava: rx.Observable[_ <: T])
     Observable[U](rx.Observable.merge(thisJava, thatJava))
   }
 
+  /**
+   * This behaves like {@link #merge(Observable)} except that if any of the merged Observables
+   * notify of an error via {@link Observer#onError onError}, {@code mergeDelayError} will
+   * refrain from propagating that error notification until all of the merged Observables have
+   * finished emitting items.
+   * <p>
+   * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/mergeDelayError.png">
+   * <p>
+   * Even if multiple merged Observables send {@code onError} notifications, {@code mergeDelayError} will only invoke the {@code onError} method of its
+   * Observers once.
+   * <p>
+   * This method allows an Observer to receive all successfully emitted items from all of the
+   * source Observables without being interrupted by an error notification from one of them.
+   *
+   * @param that
+   *            an Observable to be merged
+   * @return an Observable that emits items that are the result of flattening the items emitted by
+   *         {$code this} and {$code that}
+   */
+  def mergeDelayError[U >: T](that: Observable[U]): Observable[U] = {
+    Observable[U](rx.Observable.mergeDelayError[U](this.asJava, that.asJava))
+  }
+
   /**
    * Flattens the sequence of Observables emitted by {@code this} into one Observable, without any
    * transformation.
@@ -1291,6 +1314,31 @@ class Observable[+T](val asJava: rx.Observable[_ <: T])
     Observable[U](o5)
   }
 
+  /**
+   * This behaves like {@link #flatten(<:<)} except that if any of the merged Observables
+   * notify of an error via {@link Observer#onError onError}, this method will
+   * refrain from propagating that error notification until all of the merged Observables have
+   * finished emitting items.
+   * <p>
+   * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/mergeDelayError.png">
+   * <p>
+   * Even if multiple merged Observables send {@code onError} notifications, this method will only invoke the {@code onError} method of its
+   * Observers once.
+   * <p>
+   * This method allows an Observer to receive all successfully emitted items from all of the
+   * source Observables without being interrupted by an error notification from one of them.
+   *
+   * @return an Observable that emits items that are the result of flattening the items emitted by
+   *         the Observables emitted by the this Observable
+   */
+  def flattenDelayError[U](implicit evidence: Observable[T] <:< Observable[Observable[U]]): Observable[U] = {
+    val o2: Observable[Observable[U]] = this
+    val o3: Observable[rx.Observable[_ <: U]] = o2.map(_.asJava)
+    val o4: rx.Observable[_ <: rx.Observable[_ <: U]] = o3.asJava
+    val o5 = rx.Observable.mergeDelayError[U](o4)
+    Observable[U](o5)
+  }
+
   /**
    * Debounces by dropping all values that are followed by newer values before the timeout value expires. The timer resets on each `onNext` call.
    * <p>

language-adaptors/rxjava-scala/src/test/scala/rx/lang/scala/CompletenessTest.scala

@@ -74,7 +74,8 @@ class CompletenessTest extends JUnitSuite {
       "from(Iterable[_ <: T])" -> "apply(T*)", 
       "merge(Observable[_ <: T], Observable[_ <: T])" -> "merge(Observable[U])",
       "merge(Observable[_ <: Observable[_ <: T]])" -> "flatten(<:<[Observable[T], Observable[Observable[U]]])",
-      "mergeDelayError(Observable[_ <: T], Observable[_ <: T])" -> "mergeDelayError(Observable[T])",
+      "mergeDelayError(Observable[_ <: T], Observable[_ <: T])" -> "mergeDelayError(Observable[U])",
+      "mergeDelayError(Observable[_ <: Observable[_ <: T]])" -> "flattenDelayError(<:<[Observable[T], Observable[Observable[U]]])",
       "range(Int, Int)" -> "apply(Range)",
       "sequenceEqual(Observable[_ <: T], Observable[_ <: T])" -> "[use (first zip second) map (p => p._1 == p._2)]",
       "sequenceEqual(Observable[_ <: T], Observable[_ <: T], Func2[_ >: T, _ >: T, Boolean])" -> "[use (first zip second) map (p => equality(p._1, p._2))]",
@@ -105,6 +106,9 @@ class CompletenessTest extends JUnitSuite {
   }).toMap ++ List.iterate("Observable[_ <: T]", 9)(s => s + ", Observable[_ <: T]").map(
       // merge 3-9:
       "merge(" + _ + ")" -> "[unnecessary because we can use Observable(o1, o2, ...).flatten instead]"
+  ).drop(2).toMap ++ List.iterate("Observable[_ <: T]", 9)(s => s + ", Observable[_ <: T]").map(
+      // mergeDelayError 3-9:
+      "mergeDelayError(" + _ + ")" -> "[unnecessary because we can use Observable(o1, o2, ...).flattenDelayError instead]"
   ).drop(2).toMap
 
   def removePackage(s: String) = s.replaceAll("(\\w+\\.)+(\\w+)", "$2")