groupByUntil -> groupBy

This collapses groupByUntil and groupBy into a single groupBy operator. The new implementation has 2 major changes:

1) It supports reactive pull backpressure.
2) Child GroupedObservables can be unsubscribed and they will be cleaned up and then new instances for the same key can be emitted, like groupByUntil, except that now instead of passing in a special durationSelector function, the child can be composed using take/takeUntil/etc to cause an unsubscribe.

If the previous non-obvious groupBy behavior is wanted, then instead of unsubscribing, it can be filtered to ignore all further data, which is what the old groupBy used to do when a child was unsubscribed.

Author: benjchristensen

src/main/java/rx/Observable.java

@@ -4708,15 +4708,6 @@ public final void forEach(final Action1<? super T> onNext, final Action1<Throwab
      * is subscribed to. For this reason, in order to avoid memory leaks, you should not simply ignore those
      * {@code GroupedObservable}s that do not concern you. Instead, you can signal to them that they may
      * discard their buffers by applying an operator like {@link #take}{@code (0)} to them.
-     * <dl>
-     *  <dt><b>Backpressure Support:</b></dt>
-     *  <dd>This operator does not support backpressure as splitting a stream effectively turns it into a "hot
-     *      observable" and blocking any one group would block the entire parent stream. If you need
-     *      backpressure on individual groups then you should use operators such as {@link #onBackpressureDrop}
-     *      or {@link #onBackpressureBuffer}.</dd>
-     *  <dt><b>Scheduler:</b></dt>
-     *  <dd>{@code groupBy} does not operate by default on a particular {@link Scheduler}.</dd>
-     * </dl>
      * 
      * @param keySelector
      *            a function that extracts the key for each item
@@ -4746,15 +4737,6 @@ public final <K, R> Observable<GroupedObservable<K, R>> groupBy(final Func1<? su
      * is subscribed to. For this reason, in order to avoid memory leaks, you should not simply ignore those
      * {@code GroupedObservable}s that do not concern you. Instead, you can signal to them that they may
      * discard their buffers by applying an operator like {@link #take}{@code (0)} to them.
-     * <dl>
-     *  <dt><b>Backpressure Support:</b></dt>
-     *  <dd>This operator does not support backpressure as splitting a stream effectively turns it into a "hot
-     *      observable" and blocking any one group would block the entire parent stream. If you need
-     *      backpressure on individual groups then you should use operators such as {@link #onBackpressureDrop}
-     *      or {@link #onBackpressureBuffer}.</dd>
-     *  <dt><b>Scheduler:</b></dt>
-     *  <dd>{@code groupBy} does not operate by default on a particular {@link Scheduler}.</dd>
-     * </dl>
      * 
      * @param keySelector
      *            a function that extracts the key for each item
@@ -4770,77 +4752,6 @@ public final <K> Observable<GroupedObservable<K, T>> groupBy(final Func1<? super
         return lift(new OperatorGroupBy<T, K, T>(keySelector));
     }
 
-    /**
-     * Groups the items emitted by an {@code Observable} according to a specified key selector function until
-     * the duration {@code Observable} expires for the key.
-     * <p>
-     * <img width="640" height="375" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/groupByUntil.png" alt="">
-     * <p>
-     * <em>Note:</em> A {@link GroupedObservable} 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
-     * {@code GroupedObservable}s that do not concern you. Instead, you can signal to them that they may
-     * discard their buffers by applying an operator like {@link #take}{@code (0)} to them.
-     * <dl>
-     *  <dt><b>Backpressure Support:</b></dt>
-     *  <dd>This operator does not support backpressure as splitting a stream effectively turns it into a "hot
-     *      observable" and blocking any one group would block the entire parent stream. If you need
-     *      backpressure on individual groups then you should use operators such as {@link #onBackpressureDrop}
-     *      or {@link #onBackpressureBuffer}.</dd>
-     *  <dt><b>Scheduler:</b></dt>
-     *  <dd>{@code groupByUntil} does not operate by default on a particular {@link Scheduler}.</dd>
-     * </dl>
-     * 
-     * @param keySelector
-     *            a function to extract the key for each item
-     * @param durationSelector
-     *            a function to signal the expiration of a group
-     * @return an {@code Observable} that emits {@link GroupedObservable}s, each of which corresponds to a key
-     *         value and each of which emits all items emitted by the source {@code Observable} during that
-     *         key's duration that share that same key value
-     * @see <a href="https://github.com/ReactiveX/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 final <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);
-    }
-
-    /**
-     * Groups the items emitted by an {@code Observable} (transformed by a selector) according to a specified
-     * key selector function until the duration Observable expires for the key.
-     * <p>
-     * <img width="640" height="375" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/groupByUntil.png" alt="">
-     * <p>
-     * <em>Note:</em> A {@link GroupedObservable} 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
-     * {@code GroupedObservable}s that do not concern you. Instead, you can signal to them that they may
-     * discard their buffers by applying an operator like {@link #take}{@code (0)} to them.
-     * <dl>
-     *  <dt><b>Backpressure Support:</b></dt>
-     *  <dd>This operator does not support backpressure as splitting a stream effectively turns it into a "hot
-     *      observable" and blocking any one group would block the entire parent stream. If you need
-     *      backpressure on individual groups then you should use operators such as {@link #onBackpressureDrop}
-     *      or {@link #onBackpressureBuffer}.</dd>
-     *  <dt><b>Scheduler:</b></dt>
-     *  <dd>{@code groupByUntil} does not operate by default on a particular {@link Scheduler}.</dd>
-     * </dl>
-     * 
-     * @param keySelector
-     *            a function to extract the key for each item
-     * @param valueSelector
-     *            a function to map each item emitted by the source {@code Observable} to an item emitted by one
-     *            of the resulting {@link GroupedObservable}s
-     * @param durationSelector
-     *            a function to signal the expiration of a group
-     * @return an {@code Observable} that emits {@link GroupedObservable}s, each of which corresponds to a key
-     *         value and each of which emits all items emitted by the source {@code Observable} during that
-     *         key's duration that share that same key value, transformed by the value selector
-     * @see <a href="https://github.com/ReactiveX/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 final <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 lift(new OperatorGroupByUntil<T, TKey, TValue, TDuration>(keySelector, valueSelector, durationSelector));
-    }
-
     /**
      * Returns an Observable that correlates two Observables when they overlap in time and groups the results.
      * <p>