modified groupByUntil include the key of the group in the closings function.

Author: Christopher Grimm

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

@@ -1332,20 +1332,18 @@ trait Observable[+T]
    * @param f
    *            a function that extracts the key from an item
    * @param closings
-   *            the function that accepts a created grouping and emits an observable which upon emitting a Closing, 
-   *            closes the group.
+   *            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`
+   * @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: Observable[T]=>Observable[Closing]): Observable[(K, Observable[T])] = {
-    val closing = (o: rx.observables.GroupedObservable[K,T]) => closings(toScalaObservable[T](o)).asJavaObservable
-    val fclosing = new Func1[rx.observables.GroupedObservable[K, T], rx.Observable[_ <: Closing]] {
-      def call(o: rx.observables.GroupedObservable[K, T]) = closing(o)
-    }.asInstanceOf[Func1[_ >: rx.observables.GroupedObservable[K, _ <: T], _ <: rx.Observable[Closing]]]
+  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))

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

@@ -7552,7 +7552,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);
     }
 
@@ -7573,7 +7573,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() {
         }
 
     }
-}
+}