Add groupByUntil variant to RxScala

Author: zsxwing

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

@@ -294,6 +294,13 @@ class RxScalaDemo extends JUnitSuite {
     sequenced.subscribe(x => println(s"Emitted group: $x"))
   }
 
+  @Test def groupByUntilExample2() {
+    val numbers = Observable.interval(250 millis).take(14)
+    val grouped = numbers.groupByUntil[Long, Long](x => x % 2, x => x * 10, {case (key, obs) => Observable.interval(2 seconds)})
+    val sequenced = (grouped.map({ case (key, obs) => obs.toSeq })).flatten
+    sequenced.toBlocking.foreach(x => println(s"Emitted group: $x"))
+  }
+
   @Test def combineLatestExample() {
     val firstCounter = Observable.interval(250 millis)
     val secondCounter = Observable.interval(550 millis)

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

@@ -2193,6 +2193,35 @@ trait Observable[+T]
     toScalaObservable[(K, Observable[T])](o1.map[(K, Observable[T])](func))
   }
 
+  /**
+   * Groups the items emitted by an [[Observable]] (transformed by a selector) according to a specified key selector function
+   * until the duration Observable expires for the key.
+   *
+   * <img width="640" height="375" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/groupByUntil.png">
+   *
+   * <em>Note:</em> The `Observable` in the pair `(K, Observable[V])` will cache the items it is to emit until such time as it
+   * is subscribed to. For this reason, in order to avoid memory leaks, you should not simply ignore those `Observable` that
+   * do not concern you. Instead, you can signal to them that they may discard their buffers by applying an operator like `take(0)` to them.
+   *
+   * @param keySelector a function to extract the key for each item
+   * @param valueSelector a function to map each item emitted by the source [[Observable]] to an item emitted by one
+   *                      of the resulting `Observable[V]`s
+   * @param closings a function to signal the expiration of a group
+   * @return an [[Observable]] that emits pairs of key and `Observable[V]`, each of which corresponds to a key
+   *         value and each of which emits all items emitted by the source [[Observable]] during that
+   *         key's duration that share that same key value, transformed by the value selector
+   */
+  def groupByUntil[K, V](keySelector: T => K, valueSelector: T => V, closings: (K, Observable[V]) => Observable[Any]): Observable[(K, Observable[V])] = {
+    val jKeySelector: Func1[_ >: T, _ <: K] = keySelector
+    val jValueSelector: Func1[_ >: T, _ <: V] = valueSelector
+    val jDurationSelector = new Func1[rx.observables.GroupedObservable[_ <: K, _ <: V], rx.Observable[_ <: Any]] {
+      override def call(jgo: rx.observables.GroupedObservable[_ <: K, _ <: V]): rx.Observable[_ <: Any] = closings(jgo.getKey, toScalaObservable[V](jgo))
+    }
+    val f = (o: rx.observables.GroupedObservable[K, _ <: V]) => (o.getKey, toScalaObservable[V](o))
+    val jo = asJavaObservable.groupByUntil[K, V, Any](jKeySelector, jValueSelector, jDurationSelector).map[(K, Observable[V])](f)
+    toScalaObservable[(K, Observable[V])](jo)
+  }
+
   /**
    * Correlates the items emitted by two Observables based on overlapping durations.
    * <p>

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

@@ -96,6 +96,7 @@ class CompletenessTest extends JUnitSuite {
       "forEach(Action1[_ >: T], Action1[Throwable], Action0)" -> "foreach(T => Unit, Throwable => Unit, () => Unit)",
       "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])",
       "ignoreElements()" -> "[use `filter(_ => false)`]",
       "last(Func1[_ >: T, Boolean])" -> "[use `filter(predicate).last`]",
       "lastOrDefault(T)" -> "lastOrElse(=> U)",