Merge pull request #732 from chrisgrimm/master

Ported groupByUntil function to scala-adapter

Author: benjchristensen

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

@@ -242,6 +242,17 @@ class RxScalaDemo extends JUnitSuite {
     waitFor(firstMedalOfEachCountry)
   }
 
+  @Test def groupByUntilExample() {
+    val numbers = Observable.interval(250 millis) take 14
+    val grouped = numbers.groupByUntil[Long, Long](
+      {case x => x % 2},
+      {case (key, obs) => obs filter {case x => x == 7}}
+    )
+    val sequenced = (grouped map {case (key, obs) => obs.toSeq}).flatten
+    sequenced subscribe {x => println(s"Emitted group: $x")}
+  }
+
+
   @Test def olympicsExample() {
     val medals = Olympics.mountainBikeMedals.publish
     medals.subscribe(println(_))
@@ -449,4 +460,4 @@ class RxScalaDemo extends JUnitSuite {
     obs.toBlockingObservable.toIterable.last
   }
 
-}
+}

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

@@ -1325,6 +1325,30 @@ trait Observable[+T]
     toScalaObservable[(K, Observable[T])](o1.map[(K, Observable[T])](func))
   }
 
+  /**
+   * Groups the items emitted by this Observable according to a specified discriminator function and terminates these groups
+   * according to a function.
+   *
+   * @param f
+   *            a function that extracts the key from an item
+   * @param closings
+   *            the function that accepts the key of a given group and an observable representing that group, and returns
+   *            an observable that emits a single Closing when the group should be closed.
+   * @tparam K 
+   *            the type of the keys returned by the discriminator function.
+   * @tparam Closing
+   *            the type of the element emitted from the closings observable.
+   * @return an Observable that emits `(key, observable)` pairs, where `observable`
+   *         contains all items for which `f` returned `key` before `closings` emits a value.
+   */
+  def groupByUntil[K, Closing](f: T => K, closings: (K, Observable[T])=>Observable[Closing]): Observable[(K, Observable[T])] = {
+    val fclosing: Func1[_ >: rx.observables.GroupedObservable[K, _ <: T], _ <: rx.Observable[_ <: Closing]] =
+      (jGrObs: rx.observables.GroupedObservable[K, _ <: T]) => closings(jGrObs.getKey, toScalaObservable[T](jGrObs)).asJavaObservable
+    val o1 = asJavaObservable.groupByUntil[K, Closing](f, fclosing) : rx.Observable[_ <: rx.observables.GroupedObservable[K, _ <: T]]
+    val func = (o: rx.observables.GroupedObservable[K, _ <: T]) => (o.getKey, toScalaObservable[T](o))
+    toScalaObservable[(K, Observable[T])](o1.map[(K, Observable[T])](func))
+  }
+
   /**
    * Given an Observable that emits Observables, creates a single Observable that
    * emits the items emitted by the most recently published of those Observables.

rxjava-core/src/main/java/rx/Observable.java

@@ -8008,7 +8008,7 @@ public <U> Observable<T> skipUntil(Observable<U> other) {
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Transforming-Observables#groupby-and-groupbyuntil">RxJava Wiki: groupByUntil()</a>
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh211932.aspx">MSDN: Observable.GroupByUntil</a>
      */
-    public <TKey, TDuration> Observable<GroupedObservable<TKey, T>> groupByUntil(Func1<? super T, ? extends TKey> keySelector, Func1<? super GroupedObservable<TKey, T>, ? extends Observable<TDuration>> durationSelector) {
+    public <TKey, TDuration> Observable<GroupedObservable<TKey, T>> groupByUntil(Func1<? super T, ? extends TKey> keySelector, Func1<? super GroupedObservable<TKey, T>, ? extends Observable<? extends TDuration>> durationSelector) {
         return groupByUntil(keySelector, Functions.<T>identity(), durationSelector);
     }
 
@@ -8029,7 +8029,7 @@ public <TKey, TDuration> Observable<GroupedObservable<TKey, T>> groupByUntil(Fun
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Transforming-Observables#groupby-and-groupbyuntil">RxJava Wiki: groupByUntil()</a>
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229433.aspx">MSDN: Observable.GroupByUntil</a>
      */
-    public <TKey, TValue, TDuration> Observable<GroupedObservable<TKey, TValue>> groupByUntil(Func1<? super T, ? extends TKey> keySelector, Func1<? super T, ? extends TValue> valueSelector, Func1<? super GroupedObservable<TKey, TValue>, ? extends Observable<TDuration>> durationSelector) {
+    public <TKey, TValue, TDuration> Observable<GroupedObservable<TKey, TValue>> groupByUntil(Func1<? super T, ? extends TKey> keySelector, Func1<? super T, ? extends TValue> valueSelector, Func1<? super GroupedObservable<TKey, TValue>, ? extends Observable<? extends TDuration>> durationSelector) {
         return create(new OperationGroupByUntil<T, TKey, TValue, TDuration>(this, keySelector, valueSelector, durationSelector));
     }
 

rxjava-core/src/main/java/rx/operators/OperationGroupByUntil.java

@@ -42,11 +42,11 @@ public class OperationGroupByUntil<TSource, TKey, TResult, TDuration> implements
     final Observable<TSource> source;
     final Func1<? super TSource, ? extends TKey> keySelector;
     final Func1<? super TSource, ? extends TResult> valueSelector;
-    final Func1<? super GroupedObservable<TKey, TResult>, ? extends Observable<TDuration>> durationSelector;
+    final Func1<? super GroupedObservable<TKey, TResult>, ? extends Observable<? extends TDuration>> durationSelector;
     public OperationGroupByUntil(Observable<TSource> source,
             Func1<? super TSource, ? extends TKey> keySelector,
             Func1<? super TSource, ? extends TResult> valueSelector,
-            Func1<? super GroupedObservable<TKey, TResult>, ? extends Observable<TDuration>> durationSelector) {
+            Func1<? super GroupedObservable<TKey, TResult>, ? extends Observable<? extends TDuration>> durationSelector) {
         this.source = source;
         this.keySelector = keySelector;
         this.valueSelector = valueSelector;
@@ -107,7 +107,7 @@ public void onNext(TSource args) {
             }
 
             if (newGroup) {
-                Observable<TDuration> duration;
+                Observable<? extends TDuration> duration;
                 try {
                     duration = durationSelector.call(g);
                 } catch (Throwable t) {
@@ -234,4 +234,4 @@ public void onCompleted() {
         }
 
     }
-}
+}