Add groupJoin to RxScala

Author: zsxwing

language-adaptors/rxjava-scala/src/examples/scala/rx/lang/scala/examples/RxScalaDemo.scala

@@ -1275,4 +1275,24 @@ class RxScalaDemo extends JUnitSuite {
     }
     o.toBlocking.foreach(println)
   }
+
+  @Test def joinExample() {
+    val o1 = Observable.interval(500 millis).map(n => "1: " + n)
+    val o2 = Observable.interval(100 millis).map(n => "2: " + n)
+    val o = o1.join(o2,
+      (_: String) => Observable.timer(300 millis),
+      (_: String) => Observable.timer(200 millis),
+      (t1: String, t2: String) => (t1, t2))
+    o.take(10).toBlocking.foreach(println)
+  }
+
+  @Test def groupJoinExample() {
+    val o1 = Observable.interval(500 millis).map(n => "1: " + n)
+    val o2 = Observable.interval(100 millis).map(n => "2: " + n)
+    val o = o1.groupJoin(o2,
+      (_: String) => Observable.timer(300 millis),
+      (_: String) => Observable.timer(200 millis),
+      (t1: String, t2: Observable[String]) => (t1, t2.toSeq.toBlocking.single))
+    o.take(3).toBlocking.foreach(println)
+  }
 }

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

@@ -2266,6 +2266,36 @@ trait Observable[+T]
     )
   }
 
+  /**
+   * Returns an Observable that correlates two Observables when they overlap in time and groups the results.
+   *
+   * <img width="640" height="380" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/groupJoin.png">
+   *
+   * @param other the other Observable to correlate items from the source Observable with
+   * @param leftDuration a function that returns an Observable whose emissions indicate the duration of the values of
+   *                     the source Observable
+   * @param rightDuration a function that returns an Observable whose emissions indicate the duration of the values of
+   *                      the `other` Observable
+   * @param resultSelector a function that takes an item emitted by each Observable and returns the value to be emitted
+   *                       by the resulting Observable
+   * @return an Observable that emits items based on combining those items emitted by the source Observables
+   *         whose durations overlap
+   */
+  def groupJoin[S, R](other: Observable[S], leftDuration: T => Observable[Any], rightDuration: S => Observable[Any], resultSelector: (T, Observable[S]) => R): Observable[R] = {
+    val outer: rx.Observable[_ <: T] = this.asJavaObservable
+    val inner: rx.Observable[_ <: S] = other.asJavaObservable
+    val left: Func1[_ >: T, _ <: rx.Observable[_ <: Any]] = (t: T) => leftDuration(t).asJavaObservable
+    val right: Func1[_ >: S, _ <: rx.Observable[_ <: Any]] = (s: S) => rightDuration(s).asJavaObservable
+    val f: Func2[_ >: T, _ >: rx.Observable[S], _ <: R] = (t: T, o: rx.Observable[S]) => resultSelector(t, toScalaObservable[S](o))
+    toScalaObservable[R](
+      outer.asInstanceOf[rx.Observable[T]].groupJoin[S, Any, Any, R](
+        inner.asInstanceOf[rx.Observable[S]],
+        left.asInstanceOf[Func1[T, rx.Observable[Any]]],
+        right.asInstanceOf[Func1[S, rx.Observable[Any]]],
+        f)
+    )
+  }
+
   /**
    * Returns a new Observable by applying a function that you supply to each item emitted by the source
    * Observable that returns an Observable, and then emitting the items emitted by the most recently emitted

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

@@ -97,7 +97,9 @@ class CompletenessTest extends JUnitSuite {
       "groupBy(Func1[_ >: T, _ <: K], Func1[_ >: T, _ <: R])" -> "[use `groupBy` and `map`]",
       "groupByUntil(Func1[_ >: T, _ <: TKey], Func1[_ >: GroupedObservable[TKey, T], _ <: Observable[_ <: TDuration]])" -> "groupByUntil(T => K, (K, Observable[T]) => Observable[Any])",
       "groupByUntil(Func1[_ >: T, _ <: TKey], Func1[_ >: T, _ <: TValue], Func1[_ >: GroupedObservable[TKey, TValue], _ <: Observable[_ <: TDuration]])" -> "groupByUntil(T => K, T => V, (K, Observable[V]) => Observable[Any])",
+      "groupJoin(Observable[T2], Func1[_ >: T, _ <: Observable[D1]], Func1[_ >: T2, _ <: Observable[D2]], Func2[_ >: T, _ >: Observable[T2], _ <: R])" -> "groupJoin(Observable[S], T => Observable[Any], S => Observable[Any], (T, Observable[S]) => R)",
       "ignoreElements()" -> "[use `filter(_ => false)`]",
+      "join(Observable[TRight], Func1[T, Observable[TLeftDuration]], Func1[TRight, Observable[TRightDuration]], Func2[T, TRight, R])" -> "join(Observable[S], T => Observable[Any], S => Observable[Any], (T, S) => R)",
       "last(Func1[_ >: T, Boolean])" -> "[use `filter(predicate).last`]",
       "lastOrDefault(T)" -> "lastOrElse(=> U)",
       "lastOrDefault(T, Func1[_ >: T, Boolean])" -> "[use `filter(predicate).lastOrElse(default)`]",