diff --git a/rxjava-core/src/main/java/rx/Observable.java b/rxjava-core/src/main/java/rx/Observable.java index b9ae791382..0cef17b1d3 100644 --- a/rxjava-core/src/main/java/rx/Observable.java +++ b/rxjava-core/src/main/java/rx/Observable.java @@ -19,13 +19,17 @@ import java.util.ArrayList; import java.util.Arrays; +import java.util.Collection; import java.util.Comparator; import java.util.List; +import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import rx.concurrency.Schedulers; +import rx.joins.Pattern2; +import rx.joins.Plan0; import rx.observables.BlockingObservable; import rx.observables.ConnectableObservable; import rx.observables.GroupedObservable; @@ -51,6 +55,8 @@ import rx.operators.OperationFirstOrDefault; import rx.operators.OperationGroupBy; import rx.operators.OperationInterval; +import rx.operators.OperationJoin; +import rx.operators.OperationJoinPatterns; import rx.operators.OperationLast; import rx.operators.OperationMap; import rx.operators.OperationMaterialize; @@ -64,6 +70,7 @@ import rx.operators.OperationOnErrorReturn; import rx.operators.OperationOnExceptionResumeNextViaObservable; import rx.operators.OperationParallel; +import rx.operators.OperationParallelMerge; import rx.operators.OperationRetry; import rx.operators.OperationSample; import rx.operators.OperationScan; @@ -82,6 +89,8 @@ import rx.operators.OperationTimeInterval; import rx.operators.OperationTimeout; import rx.operators.OperationTimestamp; +import rx.operators.OperationToMap; +import rx.operators.OperationToMultimap; import rx.operators.OperationToObservableFuture; import rx.operators.OperationToObservableIterable; import rx.operators.OperationToObservableList; @@ -129,7 +138,7 @@ * The documentation for this interface makes use of marble diagrams. The * following legend explains these diagrams: *

- * + * *

* For more information see the * RxJava Wiki @@ -302,6 +311,33 @@ private Subscription protectivelyWrapAndSubscribe(Observer o) { return subscription.wrap(subscribe(new SafeObserver(subscription, o))); } + /** + * Subscribe and ignore all events. + * + * @return + */ + public Subscription subscribe() { + return protectivelyWrapAndSubscribe(new Observer() { + + @Override + public void onCompleted() { + // do nothing + } + + @Override + public void onError(Throwable e) { + handleError(e); + throw new OnErrorNotImplementedException(e); + } + + @Override + public void onNext(T args) { + // do nothing + } + + }); + } + /** * An {@link Observer} must call an Observable's {@code subscribe} method * in order to receive items and notifications from the Observable. @@ -477,6 +513,7 @@ public Subscription subscribe(final Action1 onNext, final Action1RxJava Wiki: Observable.publish() and Observable.multicast() */ public ConnectableObservable multicast(Subject subject) { return OperationMulticast.multicast(this, subject); @@ -546,7 +583,7 @@ public Subscription onSubscribe(Observer observer) { * Creates an Observable that will execute the given function when an * {@link Observer} subscribes to it. *

- * + * *

* Write the function you pass to create so that it behaves as * an Observable: It should invoke the Observer's @@ -567,6 +604,7 @@ public Subscription onSubscribe(Observer observer) { * allow the Observer to cancel the subscription * @return an Observable that, when an {@link Observer} subscribes to it, * will execute the given function + * @see RxJava Wiki: create() */ public static Observable create(OnSubscribeFunc func) { return new Observable(func); @@ -582,6 +620,7 @@ public static Observable create(OnSubscribeFunc func) { * @return an Observable that returns no data to the {@link Observer} and * immediately invokes the {@link Observer}'s * {@link Observer#onCompleted() onCompleted} method + * @see RxJava Wiki: empty() * @see MSDN: Observable.Empty Method */ public static Observable empty() { @@ -602,6 +641,7 @@ public static Observable empty() { * immediately invokes the {@link Observer}'s * {@link Observer#onCompleted() onCompleted} method with the * specified scheduler + * @see RxJava Wiki: empty() * @see MSDN: Observable.Empty Method (IScheduler) */ public static Observable empty(Scheduler scheduler) { @@ -620,6 +660,7 @@ public static Observable empty(Scheduler scheduler) { * @return an Observable that invokes the {@link Observer}'s * {@link Observer#onError onError} method when the Observer * subscribes to it + * @see RxJava Wiki: error() * @see MSDN: Observable.Throw Method */ public static Observable error(Throwable exception) { @@ -639,6 +680,7 @@ public static Observable error(Throwable exception) { * @return an Observable that invokes the {@link Observer}'s * {@link Observer#onError onError} method with the specified * scheduler + * @see RxJava Wiki: error() * @see MSDN: Observable.Throw Method */ public static Observable error(Throwable exception, Scheduler scheduler) { @@ -648,7 +690,7 @@ public static Observable error(Throwable exception, Scheduler scheduler) /** * Converts an {@link Iterable} sequence into an Observable. *

- * + * *

* Note: the entire iterable sequence is immediately emitted each time an * {@link Observer} subscribes. Since this occurs before the @@ -660,15 +702,34 @@ public static Observable error(Throwable exception, Scheduler scheduler) * type of items to be emitted by the resulting Observable * @return an Observable that emits each item in the source {@link Iterable} * sequence + * @see RxJava Wiki: from() */ public static Observable from(Iterable iterable) { return create(OperationToObservableIterable.toObservableIterable(iterable)); } + /** + * Converts an {@link Iterable} sequence into an Observable with the specified scheduler. + *

+ * + * + * @param iterable the source {@link Iterable} sequence + * @param scheduler the scheduler to emit the items of the iterable + * @param the type of items in the {@link Iterable} sequence and the + * type of items to be emitted by the resulting Observable + * @return an Observable that emits each item in the source {@link Iterable} + * sequence with the specified scheduler + * @see RxJava Wiki: from() + * @see MSDN: Observable.ToObservable + */ + public static Observable from(Iterable iterable, Scheduler scheduler) { + return from(iterable).observeOn(scheduler); + } + /** * Converts an Array into an Observable. *

- * + * *

* Note: the entire array is immediately emitted each time an * {@link Observer} subscribes. Since this occurs before the @@ -679,6 +740,7 @@ public static Observable from(Iterable iterable) { * @param the type of items in the Array and the type of items to be * emitted by the resulting Observable * @return an Observable that emits each item in the source Array + * @see RxJava Wiki: from() */ public static Observable from(T[] items) { return create(OperationToObservableIterable.toObservableIterable(Arrays.asList(items))); @@ -687,7 +749,7 @@ public static Observable from(T[] items) { /** * Converts an item into an Observable that emits that item. *

- * + * *

* Note: the item is immediately emitted each time an {@link Observer} * subscribes. Since this occurs before the {@link Subscription} is @@ -698,6 +760,7 @@ public static Observable from(T[] items) { * @param the type of the item, and the type of the item to be * emitted by the resulting Observable * @return an Observable that emits the item + * @see RxJava Wiki: from() */ @SuppressWarnings("unchecked") // suppress unchecked because we are using varargs inside the method @@ -708,7 +771,7 @@ public static Observable from(T t1) { /** * Converts a series of items into an Observable. *

- * + * *

* Note: the items will be immediately emitted each time an {@link Observer} * subscribes. Since this occurs before the {@link Subscription} is @@ -720,6 +783,7 @@ public static Observable from(T t1) { * @param the type of items, and the type of items to be emitted by the * resulting Observable * @return an Observable that emits each item + * @see RxJava Wiki: from() */ @SuppressWarnings("unchecked") // suppress unchecked because we are using varargs inside the method @@ -730,7 +794,7 @@ public static Observable from(T t1, T t2) { /** * Converts a series of items into an Observable. *

- * + * *

* Note: the items will be immediately emitted each time an {@link Observer} * subscribes. Since this occurs before the {@link Subscription} is @@ -743,6 +807,7 @@ public static Observable from(T t1, T t2) { * @param the type of items, and the type of items to be emitted by the * resulting Observable * @return an Observable that emits each item + * @see RxJava Wiki: from() */ @SuppressWarnings("unchecked") // suppress unchecked because we are using varargs inside the method @@ -753,7 +818,7 @@ public static Observable from(T t1, T t2, T t3) { /** * Converts a series of items into an Observable. *

- * + * *

* Note: the items will be immediately emitted each time an {@link Observer} * subscribes. Since this occurs before the {@link Subscription} is @@ -767,6 +832,7 @@ public static Observable from(T t1, T t2, T t3) { * @param the type of items, and the type of items to be emitted by the * resulting Observable * @return an Observable that emits each item + * @see RxJava Wiki: from() */ @SuppressWarnings("unchecked") // suppress unchecked because we are using varargs inside the method @@ -777,7 +843,7 @@ public static Observable from(T t1, T t2, T t3, T t4) { /** * Converts a series of items into an Observable. *

- * + * *

* Note: the items will be immediately emitted each time an {@link Observer} * subscribes. Since this occurs before the {@link Subscription} is @@ -792,6 +858,7 @@ public static Observable from(T t1, T t2, T t3, T t4) { * @param the type of items, and the type of items to be emitted by the * resulting Observable * @return an Observable that emits each item + * @see RxJava Wiki: from() */ @SuppressWarnings("unchecked") // suppress unchecked because we are using varargs inside the method @@ -802,7 +869,7 @@ public static Observable from(T t1, T t2, T t3, T t4, T t5) { /** * Converts a series of items into an Observable. *

- * + * *

* Note: the items will be immediately emitted each time an {@link Observer} * subscribes. Since this occurs before the {@link Subscription} is @@ -818,6 +885,7 @@ public static Observable from(T t1, T t2, T t3, T t4, T t5) { * @param the type of items, and the type of items to be emitted by the * resulting Observable * @return an Observable that emits each item + * @see RxJava Wiki: from() */ @SuppressWarnings("unchecked") // suppress unchecked because we are using varargs inside the method @@ -828,7 +896,7 @@ public static Observable from(T t1, T t2, T t3, T t4, T t5, T t6) { /** * Converts a series of items into an Observable. *

- * + * *

* Note: the items will be immediately emitted each time an {@link Observer} * subscribes. Since this occurs before the {@link Subscription} is @@ -845,6 +913,7 @@ public static Observable from(T t1, T t2, T t3, T t4, T t5, T t6) { * @param the type of items, and the type of items to be emitted by the * resulting Observable * @return an Observable that emits each item + * @see RxJava Wiki: from() */ @SuppressWarnings("unchecked") // suppress unchecked because we are using varargs inside the method @@ -855,7 +924,7 @@ public static Observable from(T t1, T t2, T t3, T t4, T t5, T t6, T t7) { /** * Converts a series of items into an Observable. *

- * + * *

* Note: the items will be immediately emitted each time an {@link Observer} * subscribes. Since this occurs before the {@link Subscription} is @@ -873,6 +942,7 @@ public static Observable from(T t1, T t2, T t3, T t4, T t5, T t6, T t7) { * @param the type of items, and the type of items to be emitted by the * resulting Observable * @return an Observable that emits each item + * @see RxJava Wiki: from() */ @SuppressWarnings("unchecked") // suppress unchecked because we are using varargs inside the method @@ -883,7 +953,7 @@ public static Observable from(T t1, T t2, T t3, T t4, T t5, T t6, T t7, T /** * Converts a series of items into an Observable. *

- * + * *

* Note: the items will be immediately emitted each time an {@link Observer} * subscribes. Since this occurs before the {@link Subscription} is @@ -902,6 +972,7 @@ public static Observable from(T t1, T t2, T t3, T t4, T t5, T t6, T t7, T * @param the type of items, and the type of items to be emitted by the * resulting Observable * @return an Observable that emits each item + * @see RxJava Wiki: from() */ @SuppressWarnings("unchecked") // suppress unchecked because we are using varargs inside the method @@ -912,7 +983,7 @@ public static Observable from(T t1, T t2, T t3, T t4, T t5, T t6, T t7, T /** * Converts a series of items into an Observable. *

- * + * *

* Note: the items will be immediately emitted each time an {@link Observer} * subscribes. Since this occurs before the {@link Subscription} is @@ -932,6 +1003,7 @@ public static Observable from(T t1, T t2, T t3, T t4, T t5, T t6, T t7, T * @param the type of items, and the type of items to be emitted by the * resulting Observable * @return an Observable that emits each item + * @see RxJava Wiki: from() */ @SuppressWarnings("unchecked") // suppress unchecked because we are using varargs inside the method @@ -943,7 +1015,7 @@ public static Observable from(T t1, T t2, T t3, T t4, T t5, T t6, T t7, T * Generates an Observable that emits a sequence of integers within a * specified range. *

- * + * *

* Note: the entire range is immediately emitted each time an * {@link Observer} subscribes. Since this occurs before the @@ -953,18 +1025,35 @@ public static Observable from(T t1, T t2, T t3, T t4, T t5, T t6, T t7, T * @param start the value of the first integer in the sequence * @param count the number of sequential integers to generate * @return an Observable that emits a range of sequential integers + * @see RxJava Wiki: range() * @see Observable.Range Method (Int32, Int32) */ public static Observable range(int start, int count) { return from(Range.createWithCount(start, count)); } + /** + * Generates an Observable that emits a sequence of integers within a + * specified range with the specified scheduler. + *

+ * + * @param start the value of the first integer in the sequence + * @param count the number of sequential integers to generate + * @param scheduler the scheduler to run the generator loop on + * @return an Observable that emits a range of sequential integers + * @see RxJava Wiki: range() + * @see Observable.Range Method (Int32, Int32, IScheduler) + */ + public static Observable range(int start, int count, Scheduler scheduler) { + return range(start, count).observeOn(scheduler); + } + /** * Returns an Observable that calls an Observable factory to create its * Observable for each new Observer that subscribes. That is, for each * subscriber, the actuall Observable is determined by the factory function. *

- * + * *

* The defer operator allows you to defer or delay emitting items from an * Observable until such time as an Observer subscribes to the Observable. @@ -977,6 +1066,7 @@ public static Observable range(int start, int count) { * @param the type of the items emitted by the Observable * @return an Observable whose {@link Observer}s trigger an invocation of * the given Observable factory function + * @see RxJava Wiki: defer() */ public static Observable defer(Func0> observableFactory) { return create(OperationDefer.defer(observableFactory)); @@ -1000,6 +1090,7 @@ public static Observable defer(Func0> o * {@link Observer#onNext onNext} method * @param the type of that item * @return an Observable that emits a single item and then completes + * @see RxJava Wiki: just() */ public static Observable just(T value) { List list = new ArrayList(); @@ -1020,6 +1111,7 @@ public static Observable just(T value) { * @param scheduler the scheduler to send the single element on * @return an Observable that emits a single item and then completes on a * specified scheduler + * @see RxJava Wiki: just() */ public static Observable just(T value, Scheduler scheduler) { return just(value).observeOn(scheduler); @@ -1038,6 +1130,7 @@ public static Observable just(T value, Scheduler scheduler) { * @return an Observable that emits items that are the result of flattening * the items emitted by the Observables emitted by the * {@code source} Observable + * @see RxJava Wiki: merge() * @see MSDN: Observable.Merge Method */ public static Observable merge(Observable> source) { @@ -1057,6 +1150,7 @@ public static Observable merge(ObservableRxJava Wiki: merge() * @see MSDN: Observable.Merge Method */ @SuppressWarnings("unchecked") @@ -1079,6 +1173,7 @@ public static Observable merge(Observable t1, ObservableRxJava Wiki: merge() * @see MSDN: Observable.Merge Method */ @SuppressWarnings("unchecked") @@ -1102,6 +1197,7 @@ public static Observable merge(Observable t1, ObservableRxJava Wiki: merge() * @see MSDN: Observable.Merge Method */ @SuppressWarnings("unchecked") @@ -1126,6 +1222,7 @@ public static Observable merge(Observable t1, ObservableRxJava Wiki: merge() * @see MSDN: Observable.Merge Method */ @SuppressWarnings("unchecked") @@ -1151,6 +1248,7 @@ public static Observable merge(Observable t1, ObservableRxJava Wiki: merge() * @see MSDN: Observable.Merge Method */ @SuppressWarnings("unchecked") @@ -1177,6 +1275,7 @@ public static Observable merge(Observable t1, ObservableRxJava Wiki: merge() * @see MSDN: Observable.Merge Method */ @SuppressWarnings("unchecked") @@ -1204,6 +1303,7 @@ public static Observable merge(Observable t1, ObservableRxJava Wiki: merge() * @see MSDN: Observable.Merge Method */ @SuppressWarnings("unchecked") @@ -1232,6 +1332,7 @@ public static Observable merge(Observable t1, ObservableRxJava Wiki: merge() * @see MSDN: Observable.Merge Method */ @SuppressWarnings("unchecked") @@ -1250,6 +1351,7 @@ public static Observable merge(Observable t1, ObservableRxJava Wiki: concat() * @see MSDN: Observable.Concat Method */ public static Observable concat(Observable> observables) { @@ -1267,6 +1369,7 @@ public static Observable concat(ObservableRxJava Wiki: concat() * @see MSDN: Observable.Concat Method */ @SuppressWarnings("unchecked") @@ -1288,6 +1391,7 @@ public static Observable concat(Observable t1, ObservableRxJava Wiki: concat() * @see MSDN: Observable.Concat Method */ @SuppressWarnings("unchecked") @@ -1309,6 +1413,7 @@ public static Observable concat(Observable t1, ObservableRxJava Wiki: concat() * @see MSDN: Observable.Concat Method */ @SuppressWarnings("unchecked") @@ -1331,6 +1436,7 @@ public static Observable concat(Observable t1, ObservableRxJava Wiki: concat() * @see MSDN: Observable.Concat Method */ @SuppressWarnings("unchecked") @@ -1354,6 +1460,7 @@ public static Observable concat(Observable t1, ObservableRxJava Wiki: concat() * @see MSDN: Observable.Concat Method */ @SuppressWarnings("unchecked") @@ -1378,6 +1485,7 @@ public static Observable concat(Observable t1, ObservableRxJava Wiki: concat() * @see MSDN: Observable.Concat Method */ @SuppressWarnings("unchecked") @@ -1403,6 +1511,7 @@ public static Observable concat(Observable t1, ObservableRxJava Wiki: concat() * @see MSDN: Observable.Concat Method */ @SuppressWarnings("unchecked") @@ -1429,6 +1538,7 @@ public static Observable concat(Observable t1, ObservableRxJava Wiki: concat() * @see MSDN: Observable.Concat Method */ @SuppressWarnings("unchecked") @@ -1458,6 +1568,7 @@ public static Observable concat(Observable t1, ObservableRxJava Wiki: mergeDelayError() * @see MSDN: Observable.Merge Method */ public static Observable mergeDelayError(Observable> source) { @@ -1485,6 +1596,7 @@ public static Observable mergeDelayError(ObservableRxJava Wiki: mergeDelayError() * @see MSDN: Observable.Merge Method */ @SuppressWarnings("unchecked") @@ -1515,6 +1627,7 @@ public static Observable mergeDelayError(Observable t1, Obse * @param t3 an Observable to be merged * @return an Observable that emits items that are the result of flattening * the items emitted by the {@code source} Observables + * @see RxJava Wiki: mergeDelayError() * @see MSDN: Observable.Merge Method */ @SuppressWarnings("unchecked") @@ -1547,6 +1660,7 @@ public static Observable mergeDelayError(Observable t1, Obse * @param t4 an Observable to be merged * @return an Observable that emits items that are the result of flattening * the items emitted by the {@code source} Observables + * @see RxJava Wiki: mergeDelayError() * @see MSDN: Observable.Merge Method */ @SuppressWarnings("unchecked") @@ -1579,6 +1693,7 @@ public static Observable mergeDelayError(Observable t1, Obse * @param t5 an Observable to be merged * @return an Observable that emits items that are the result of flattening * the items emitted by the {@code source} Observables + * @see RxJava Wiki: mergeDelayError() * @see MSDN: Observable.Merge Method */ @SuppressWarnings("unchecked") @@ -1612,6 +1727,7 @@ public static Observable mergeDelayError(Observable t1, Obse * @param t6 an Observable to be merged * @return an Observable that emits items that are the result of flattening * the items emitted by the {@code source} Observables + * @see RxJava Wiki: mergeDelayError() * @see MSDN: Observable.Merge Method */ @SuppressWarnings("unchecked") @@ -1646,6 +1762,7 @@ public static Observable mergeDelayError(Observable t1, Obse * @param t7 an Observable to be merged * @return an Observable that emits items that are the result of flattening * the items emitted by the {@code source} Observables + * @see RxJava Wiki: mergeDelayError() * @see MSDN: Observable.Merge Method */ @SuppressWarnings("unchecked") @@ -1681,6 +1798,7 @@ public static Observable mergeDelayError(Observable t1, Obse * @param t8 an Observable to be merged * @return an Observable that emits items that are the result of flattening * the items emitted by the {@code source} Observables + * @see RxJava Wiki: mergeDelayError() * @see MSDN: Observable.Merge Method */ @SuppressWarnings("unchecked") @@ -1717,6 +1835,7 @@ public static Observable mergeDelayError(Observable t1, Obse * @param t9 an Observable to be merged * @return an Observable that emits items that are the result of flattening * the items emitted by the {@code source} Observables + * @see RxJava Wiki: mergeDelayError() * @see MSDN: Observable.Merge Method */ @SuppressWarnings("unchecked") @@ -1729,13 +1848,14 @@ public static Observable mergeDelayError(Observable t1, Obse * Returns an Observable that never sends any items or notifications to an * {@link Observer}. *

- * + * *

* This Observable is useful primarily for testing purposes. * * @param the type of items (not) emitted by the Observable * @return an Observable that never sends any items or notifications to an * {@link Observer} + * @see RxJava Wiki: never() */ public static Observable never() { return new NeverObservable(); @@ -1746,11 +1866,12 @@ public static Observable never() { * that emits the items emitted by the most recently published of those * Observables. *

- * + * * * @param sequenceOfSequences the source Observable that emits Observables * @return an Observable that emits only the items emitted by the most * recently published Observable + * @see RxJava Wiki: switchOnNext() * @deprecated use {@link #switchOnNext} */ @Deprecated @@ -1763,11 +1884,12 @@ public static Observable switchDo(Observable - * + * * * @param sequenceOfSequences the source Observable that emits Observables * @return an Observable that emits only the items emitted by the most * recently published Observable + * @see RxJava Wiki: switchOnNext() */ public static Observable switchOnNext(Observable> sequenceOfSequences) { return create(OperationSwitch.switchDo(sequenceOfSequences)); @@ -1777,7 +1899,7 @@ public static Observable switchOnNext(Observable - * + * *

* A well-behaved Observable does not interleave its invocations of the * {@link Observer#onNext onNext}, {@link Observer#onCompleted onCompleted}, @@ -1791,6 +1913,7 @@ public static Observable switchOnNext(ObservableRxJava Wiki: synchronize() */ public Observable synchronize() { return create(OperationSynchronize.synchronize(this)); @@ -1802,7 +1925,7 @@ public Observable synchronize() { * accomplished by acquiring a mutual-exclusion lock for the object * provided as the lock parameter. *

- * + * *

* A well-behaved Observable does not interleave its invocations of the * {@link Observer#onNext onNext}, {@link Observer#onCompleted onCompleted}, @@ -1817,6 +1940,7 @@ public Observable synchronize() { * @return an Observable that is a chronologically well-behaved version of * the source Observable, and that synchronously notifies its * {@link Observer}s + * @see RxJava Wiki: synchronize() */ public Observable synchronize(Object lock) { return create(OperationSynchronize.synchronize(this, lock)); @@ -1833,12 +1957,13 @@ public static Observable synchronize(Observable source) { /** * Emits an item each time interval (containing a sequential number). *

- * + * * * @param interval interval size in time units (see below) * @param unit time units to use for the interval size * @return an Observable that emits an item each time interval - * @see MSDN: Observable.Interval + * @see RxJava Wiki: interval() + * @see MSDN: Observable.Interval */ public static Observable interval(long interval, TimeUnit unit) { return create(OperationInterval.interval(interval, unit)); @@ -1847,13 +1972,14 @@ public static Observable interval(long interval, TimeUnit unit) { /** * Emits an item each time interval (containing a sequential number). *

- * + * * * @param interval interval size in time units (see below) * @param unit time units to use for the interval size * @param scheduler the scheduler to use for scheduling the items * @return an Observable that emits an item each time interval - * @see MSDN: Observable.Interval + * @see RxJava Wiki: interval() + * @see MSDN: Observable.Interval */ public static Observable interval(long interval, TimeUnit unit, Scheduler scheduler) { return create(OperationInterval.interval(interval, unit, scheduler)); @@ -1866,7 +1992,7 @@ public static Observable interval(long interval, TimeUnit unit, Scheduler * Note: If events keep firing faster than the timeout then no data will be * emitted. *

- * + * *

* Information on debounce vs throttle: *

@@ -1881,6 +2007,7 @@ public static Observable interval(long interval, TimeUnit unit, Scheduler * @param unit the {@link TimeUnit} for the timeout * @return an {@link Observable} that filters out items that are too * quickly followed by newer items + * @see RxJava Wiki: debounce() * @see #throttleWithTimeout(long, TimeUnit) */ public Observable debounce(long timeout, TimeUnit unit) { @@ -1894,7 +2021,7 @@ public Observable debounce(long timeout, TimeUnit unit) { * Note: If events keep firing faster than the timeout then no data will be * emitted. *

- * + * *

* Information on debounce vs throttle: *

@@ -1911,6 +2038,7 @@ public Observable debounce(long timeout, TimeUnit unit) { * timers that handle the timeout for each event * @return an {@link Observable} that filters out items that are too * quickly followed by newer items + * @see RxJava Wiki: debounce() * @see #throttleWithTimeout(long, TimeUnit, Scheduler) */ public Observable debounce(long timeout, TimeUnit unit, Scheduler scheduler) { @@ -1924,7 +2052,7 @@ public Observable debounce(long timeout, TimeUnit unit, Scheduler scheduler) * Note: If events keep firing faster than the timeout then no data will be * emitted. *

- * + * *

* Information on debounce vs throttle: *

@@ -1939,6 +2067,7 @@ public Observable debounce(long timeout, TimeUnit unit, Scheduler scheduler) * @param unit the {@link TimeUnit} for the timeout * @return an {@link Observable} that filters out items that are too * quickly followed by newer items + * @see RxJava Wiki: throttleWithTimeout() * @see #debounce(long, TimeUnit) */ public Observable throttleWithTimeout(long timeout, TimeUnit unit) { @@ -1952,7 +2081,7 @@ public Observable throttleWithTimeout(long timeout, TimeUnit unit) { * Note: If events keep firing faster than the timeout then no data will be * emitted. *

- * + * *

* Information on debounce vs throttle: *

@@ -1969,6 +2098,7 @@ public Observable throttleWithTimeout(long timeout, TimeUnit unit) { * timers that handle the timeout for each event * @return an {@link Observable} that filters out items that are too * quickly followed by newer items + * @see RxJava Wiki: throttleWithTimeout() * @see #debounce(long, TimeUnit, Scheduler) */ public Observable throttleWithTimeout(long timeout, TimeUnit unit, Scheduler scheduler) { @@ -1982,12 +2112,13 @@ public Observable throttleWithTimeout(long timeout, TimeUnit unit, Scheduler * This differs from {@link #throttleLast} in that this only tracks passage * of time whereas {@link #throttleLast} ticks at scheduled intervals. *

- * + * * * @param windowDuration time to wait before sending another item after * emitting the last item * @param unit the unit of time for the specified timeout * @return an Observable that performs the throttle operation + * @see RxJava Wiki: throttleFirst() */ public Observable throttleFirst(long windowDuration, TimeUnit unit) { return create(OperationThrottleFirst.throttleFirst(this, windowDuration, unit)); @@ -2000,7 +2131,7 @@ public Observable throttleFirst(long windowDuration, TimeUnit unit) { * This differs from {@link #throttleLast} in that this only tracks passage * of time whereas {@link #throttleLast} ticks at scheduled intervals. *

- * + * * * @param skipDuration time to wait before sending another item after * emitting the last item @@ -2008,6 +2139,7 @@ public Observable throttleFirst(long windowDuration, TimeUnit unit) { * @param scheduler the {@link Scheduler} to use internally to manage the * timers that handle timeout for each event * @return an Observable that performs the throttle operation + * @see RxJava Wiki: throttleFirst() */ public Observable throttleFirst(long skipDuration, TimeUnit unit, Scheduler scheduler) { return create(OperationThrottleFirst.throttleFirst(this, skipDuration, unit, scheduler)); @@ -2021,12 +2153,13 @@ public Observable throttleFirst(long skipDuration, TimeUnit unit, Scheduler s * scheduled interval whereas {@link #throttleFirst} does not tick, it just * tracks passage of time. *

- * + * * * @param intervalDuration duration of windows within which the last item * will be emitted * @param unit the unit of time for the specified interval * @return an Observable that performs the throttle operation + * @see RxJava Wiki: throttleLast() * @see #sample(long, TimeUnit) */ public Observable throttleLast(long intervalDuration, TimeUnit unit) { @@ -2041,12 +2174,13 @@ public Observable throttleLast(long intervalDuration, TimeUnit unit) { * scheduled interval whereas {@link #throttleFirst} does not tick, it just * tracks passage of time. *

- * + * * * @param intervalDuration duration of windows within which the last item * will be emitted * @param unit the unit of time for the specified interval * @return an Observable that performs the throttle operation + * @see RxJava Wiki: throttleLast() * @see #sample(long, TimeUnit, Scheduler) */ public Observable throttleLast(long intervalDuration, TimeUnit unit, Scheduler scheduler) { @@ -2057,10 +2191,11 @@ public Observable throttleLast(long intervalDuration, TimeUnit unit, Schedule * Wraps each item emitted by a source Observable in a {@link Timestamped} * object. *

- * + * * * @return an Observable that emits timestamped items from the source * Observable + * @see RxJava Wiki: timestamp() */ public Observable> timestamp() { return create(OperationTimestamp.timestamp(this)); @@ -2069,7 +2204,7 @@ public Observable> timestamp() { /** * Converts a {@link Future} into an Observable. *

- * + * *

* You can convert any object that supports the {@link Future} interface * into an Observable that emits the return value of the {@link Future#get} @@ -2083,6 +2218,7 @@ public Observable> timestamp() { * @param the type of object that the {@link Future} returns, and also * the type of item to be emitted by the resulting Observable * @return an Observable that emits the item from the source Future + * @see RxJava Wiki: from() */ public static Observable from(Future future) { return create(OperationToObservableFuture.toObservableFuture(future)); @@ -2091,7 +2227,7 @@ public static Observable from(Future future) { /** * Converts a {@link Future} into an Observable. *

- * + * *

* You can convert any object that supports the {@link Future} interface * into an Observable that emits the return value of the {@link Future#get} @@ -2106,6 +2242,7 @@ public static Observable from(Future future) { * @param the type of object that the {@link Future} returns, and also * the type of item to be emitted by the resulting Observable * @return an Observable that emits the item from the source Future + * @see RxJava Wiki: from() */ public static Observable from(Future future, Scheduler scheduler) { return create(OperationToObservableFuture.toObservableFuture(future)).subscribeOn(scheduler); @@ -2114,7 +2251,7 @@ public static Observable from(Future future, Scheduler sched /** * Converts a {@link Future} into an Observable with timeout. *

- * + * *

* You can convert any object that supports the {@link Future} interface * into an Observable that emits the return value of the {link Future#get} @@ -2130,6 +2267,7 @@ public static Observable from(Future future, Scheduler sched * @param the type of object that the {@link Future} returns, and also * the type of item to be emitted by the resulting Observable * @return an Observable that emits the item from the source {@link Future} + * @see RxJava Wiki: from() */ public static Observable from(Future future, long timeout, TimeUnit unit) { return create(OperationToObservableFuture.toObservableFuture(future, timeout, unit)); @@ -2139,13 +2277,14 @@ public static Observable from(Future future, long timeout, T * Returns an Observable that emits Boolean values that indicate whether the * pairs of items emitted by two source Observables are equal. *

- * + * * * @param first the first Observable to compare * @param second the second Observable to compare * @param the type of items emitted by each Observable * @return an Observable that emits Booleans that indicate whether the * corresponding items emitted by the source Observables are equal + * @see RxJava Wiki: sequenceEqual() */ public static Observable sequenceEqual(Observable first, Observable second) { return sequenceEqual(first, second, new Func2() { @@ -2161,7 +2300,7 @@ public Boolean call(T first, T second) { * pairs of items emitted by two source Observables are equal based on the * results of a specified equality function. *

- * + * * * @param first the first Observable to compare * @param second the second Observable to compare @@ -2170,6 +2309,7 @@ public Boolean call(T first, T second) { * @param the type of items emitted by each Observable * @return an Observable that emits Booleans that indicate whether the * corresponding items emitted by the source Observables are equal + * @see RxJava Wiki: sequenceEqual() */ public static Observable sequenceEqual(Observable first, Observable second, Func2 equality) { return zip(first, second, equality); @@ -2200,6 +2340,7 @@ public static Observable sequenceEqual(Observable firs * each of the source Observables, results in an item that will * be emitted by the resulting Observable * @return an Observable that emits the zipped results + * @see RxJava Wiki: zip() */ public static Observable zip(Observable o1, Observable o2, Func2 zipFunction) { return create(OperationZip.zip(o1, o2, zipFunction)); @@ -2232,6 +2373,7 @@ public static Observable zip(Observable o1, Observa * each of the source Observables, results in an item * that will be emitted by the resulting Observable * @return an Observable that emits the zipped results + * @see RxJava Wiki: zip() */ public static Observable zip(Observable o1, Observable o2, Observable o3, Func3 zipFunction) { return create(OperationZip.zip(o1, o2, o3, zipFunction)); @@ -2265,6 +2407,7 @@ public static Observable zip(Observable o1, Obs * each of the source Observables, results in an item that will * be emitted by the resulting Observable * @return an Observable that emits the zipped results + * @see RxJava Wiki: zip() */ public static Observable zip(Observable o1, Observable o2, Observable o3, Observable o4, Func4 zipFunction) { return create(OperationZip.zip(o1, o2, o3, o4, zipFunction)); @@ -2299,6 +2442,7 @@ public static Observable zip(Observable o1, * each of the source Observables, results in an item * that will be emitted by the resulting Observable * @return an Observable that emits the zipped results + * @see RxJava Wiki: zip() */ public static Observable zip(Observable o1, Observable o2, Observable o3, Observable o4, Observable o5, Func5 zipFunction) { return create(OperationZip.zip(o1, o2, o3, o4, o5, zipFunction)); @@ -2332,6 +2476,7 @@ public static Observable zip(Observable * each of the source Observables, results in an item * that will be emitted by the resulting Observable * @return an Observable that emits the zipped results + * @see RxJava Wiki: zip() */ public static Observable zip(Observable o1, Observable o2, Observable o3, Observable o4, Observable o5, Observable o6, Func6 zipFunction) { @@ -2367,6 +2512,7 @@ public static Observable zip(ObservableRxJava Wiki: zip() */ public static Observable zip(Observable o1, Observable o2, Observable o3, Observable o4, Observable o5, Observable o6, Observable o7, Func7 zipFunction) { @@ -2403,6 +2549,7 @@ public static Observable zip(ObservableRxJava Wiki: zip() */ public static Observable zip(Observable o1, Observable o2, Observable o3, Observable o4, Observable o5, Observable o6, Observable o7, Observable o8, Func8 zipFunction) { @@ -2440,6 +2587,7 @@ public static Observable zip(ObservableRxJava Wiki: zip() */ public static Observable zip(Observable o1, Observable o2, Observable o3, Observable o4, Observable o5, Observable o6, Observable o7, Observable o8, Observable o9, Func9 zipFunction) { @@ -2460,6 +2608,7 @@ public static Observable zip(Observab * source observable values * @return an Observable that combines the source Observables with the * given combine function + * @see RxJava Wiki: combineLatest() */ public static Observable combineLatest(Observable o1, Observable o2, Func2 combineFunction) { return create(OperationCombineLatest.combineLatest(o1, o2, combineFunction)); @@ -2480,6 +2629,7 @@ public static Observable combineLatest(Observable o * source observable values * @return an Observable that combines the source Observables with the * given combine function + * @see RxJava Wiki: combineLatest() */ public static Observable combineLatest(Observable o1, Observable o2, Observable o3, Func3 combineFunction) { return create(OperationCombineLatest.combineLatest(o1, o2, o3, combineFunction)); @@ -2501,6 +2651,7 @@ public static Observable combineLatest(ObservableRxJava Wiki: combineLatest() */ public static Observable combineLatest(Observable o1, Observable o2, Observable o3, Observable o4, Func4 combineFunction) { @@ -2524,6 +2675,7 @@ public static Observable combineLatest(ObservableRxJava Wiki: combineLatest() */ public static Observable combineLatest(Observable o1, Observable o2, Observable o3, Observable o4, Observable o5, Func5 combineFunction) { @@ -2548,6 +2700,7 @@ public static Observable combineLatest(ObservableRxJava Wiki: combineLatest() */ public static Observable combineLatest(Observable o1, Observable o2, Observable o3, Observable o4, Observable o5, Observable o6, Func6 combineFunction) { @@ -2573,6 +2726,7 @@ public static Observable combineLatest(Observable * source observable values * @return an Observable that combines the source Observables with the * given combine function + * @see RxJava Wiki: combineLatest() */ public static Observable combineLatest(Observable o1, Observable o2, Observable o3, Observable o4, Observable o5, Observable o6, Observable o7, Func7 combineFunction) { @@ -2599,6 +2753,7 @@ public static Observable combineLatest(Observ * source observable values * @return an Observable that combines the source Observables with the * given combine function + * @see RxJava Wiki: combineLatest() */ public static Observable combineLatest(Observable o1, Observable o2, Observable o3, Observable o4, Observable o5, Observable o6, Observable o7, Observable o8, Func8 combineFunction) { @@ -2626,6 +2781,7 @@ public static Observable combineLatest(Ob * source observable values * @return an Observable that combines the source Observables with the * given combine function + * @see RxJava Wiki: combineLatest() */ public static Observable combineLatest(Observable o1, Observable o2, Observable o3, Observable o4, Observable o5, Observable o6, Observable o7, Observable o8, Observable o9, Func9 combineFunction) { @@ -2635,7 +2791,7 @@ public static Observable combineLates /** * Creates an Observable that produces buffers of collected items. *

- * + * *

* This Observable produces connected, non-overlapping buffers. The current * buffer is emitted and replaced with a new buffer when the Observable @@ -2654,6 +2810,7 @@ public static Observable combineLates * buffers, which are emitted when the current {@link Observable} * created with the {@link Func0} argument produces a * {@link rx.util.Closing} object + * @see RxJava Wiki: buffer() */ public Observable> buffer(Func0> bufferClosingSelector) { return create(OperationBuffer.buffer(this, bufferClosingSelector)); @@ -2662,7 +2819,7 @@ public Observable> buffer(Func0> /** * Creates an Observable which produces buffers of collected values. *

- * + * *

* This Observable produces buffers. Buffers are created when the specified * bufferOpenings Observable produces a {@link rx.util.Opening} @@ -2682,6 +2839,7 @@ public Observable> buffer(Func0> * @return an {@link Observable} that produces buffers that are created and * emitted when the specified {@link Observable}s publish certain * objects + * @see RxJava Wiki: buffer() */ public Observable> buffer(Observable bufferOpenings, Func1> bufferClosingSelector) { return create(OperationBuffer.buffer(this, bufferOpenings, bufferClosingSelector)); @@ -2690,7 +2848,7 @@ public Observable> buffer(Observable bufferOpenings, /** * Creates an Observable that produces buffers of collected items. *

- * + * *

* This Observable produces connected, non-overlapping buffers, each * containing count items. When the source Observable completes @@ -2700,6 +2858,7 @@ public Observable> buffer(Observable bufferOpenings, * @param count the maximum size of each buffer before it should be emitted * @return an {@link Observable} that produces connected, non-overlapping * buffers containing at most "count" items + * @see RxJava Wiki: buffer() */ public Observable> buffer(int count) { return create(OperationBuffer.buffer(this, count)); @@ -2708,7 +2867,7 @@ public Observable> buffer(int count) { /** * Creates an Observable which produces buffers of collected items. *

- * + * *

* This Observable produces buffers every skip items, each * containing count items. When the source Observable @@ -2723,6 +2882,7 @@ public Observable> buffer(int count) { * @return an {@link Observable} that produces buffers every * skip item containing at most count * items + * @see RxJava Wiki: buffer() */ public Observable> buffer(int count, int skip) { return create(OperationBuffer.buffer(this, count, skip)); @@ -2731,7 +2891,7 @@ public Observable> buffer(int count, int skip) { /** * Creates an Observable that produces buffers of collected values. *

- * + * *

* This Observable produces connected, non-overlapping buffers, each of a * fixed duration specified by the timespan argument. When the @@ -2744,6 +2904,7 @@ public Observable> buffer(int count, int skip) { * argument * @return an {@link Observable} that produces connected, non-overlapping * buffers with a fixed duration + * @see RxJava Wiki: buffer() */ public Observable> buffer(long timespan, TimeUnit unit) { return create(OperationBuffer.buffer(this, timespan, unit)); @@ -2752,7 +2913,7 @@ public Observable> buffer(long timespan, TimeUnit unit) { /** * Creates an Observable that produces buffers of collected values. *

- * + * *

* This Observable produces connected, non-overlapping buffers, each of a * fixed duration specified by the timespan argument. When the @@ -2767,6 +2928,7 @@ public Observable> buffer(long timespan, TimeUnit unit) { * and start of a buffer * @return an {@link Observable} that produces connected, non-overlapping * buffers with a fixed duration + * @see RxJava Wiki: buffer() */ public Observable> buffer(long timespan, TimeUnit unit, Scheduler scheduler) { return create(OperationBuffer.buffer(this, timespan, unit, scheduler)); @@ -2780,7 +2942,7 @@ public Observable> buffer(long timespan, TimeUnit unit, Scheduler schedu * first). When the source Observable completes or encounters an error, the * current buffer is emitted and the event is propagated. *

- * + * * * @param timespan the period of time each buffer collects values before it * should be emitted and replaced with a new buffer @@ -2790,6 +2952,7 @@ public Observable> buffer(long timespan, TimeUnit unit, Scheduler schedu * @return an {@link Observable} that produces connected, non-overlapping * buffers that are emitted after a fixed duration or when the * buffer reaches maximum capacity (whichever occurs first) + * @see RxJava Wiki: buffer() */ public Observable> buffer(long timespan, TimeUnit unit, int count) { return create(OperationBuffer.buffer(this, timespan, unit, count)); @@ -2803,7 +2966,7 @@ public Observable> buffer(long timespan, TimeUnit unit, int count) { * first). When the source Observable completes or encounters an error, the * current buffer is emitted and the event is propagated. *

- * + * * * @param timespan the period of time each buffer collects values before it * should be emitted and replaced with a new buffer @@ -2815,6 +2978,7 @@ public Observable> buffer(long timespan, TimeUnit unit, int count) { * @return an {@link Observable} that produces connected, non-overlapping * buffers that are emitted after a fixed duration or when the * buffer has reached maximum capacity (whichever occurs first) + * @see RxJava Wiki: buffer() */ public Observable> buffer(long timespan, TimeUnit unit, int count, Scheduler scheduler) { return create(OperationBuffer.buffer(this, timespan, unit, count, scheduler)); @@ -2828,7 +2992,7 @@ public Observable> buffer(long timespan, TimeUnit unit, int count, Sched * source Observable completes or encounters an error, the current buffer is * emitted and the event is propagated. *

- * + * * * @param timespan the period of time each buffer collects values before it * should be emitted @@ -2838,6 +3002,7 @@ public Observable> buffer(long timespan, TimeUnit unit, int count, Sched * and timeshift arguments * @return an {@link Observable} that produces new buffers periodically and * emits these after a fixed timespan has elapsed. + * @see RxJava Wiki: buffer() */ public Observable> buffer(long timespan, long timeshift, TimeUnit unit) { return create(OperationBuffer.buffer(this, timespan, timeshift, unit)); @@ -2851,7 +3016,7 @@ public Observable> buffer(long timespan, long timeshift, TimeUnit unit) * source Observable completes or encounters an error, the current buffer is * emitted and the event is propagated. *

- * + * * * @param timespan the period of time each buffer collects values before it * should be emitted @@ -2863,6 +3028,7 @@ public Observable> buffer(long timespan, long timeshift, TimeUnit unit) * and start of a buffer * @return an {@link Observable} that produces new buffers periodically and * emits these after a fixed timespan has elapsed + * @see RxJava Wiki: buffer() */ public Observable> buffer(long timespan, long timeshift, TimeUnit unit, Scheduler scheduler) { return create(OperationBuffer.buffer(this, timespan, timeshift, unit, scheduler)); @@ -2877,7 +3043,7 @@ public Observable> buffer(long timespan, long timeshift, TimeUnit unit, * then be used to create a new Observable to listen for the end of the next * window. *

- * + * * * @param closingSelector the {@link Func0} used to produce an * {@link Observable} for every window created. When this @@ -2887,6 +3053,7 @@ public Observable> buffer(long timespan, long timeshift, TimeUnit unit, * windows, which are emitted when the current {@link Observable} * created with the closingSelector argument emits a * {@link rx.util.Closing} object. + * @see RxJava Wiki: window() */ public Observable> window(Func0> closingSelector) { return create(OperationWindow.window(this, closingSelector)); @@ -2900,7 +3067,7 @@ public Observable> window(Func0 - * + * * * @param windowOpenings the {@link Observable} that, when it produces a * {@link rx.util.Opening} object, causes another @@ -2913,6 +3080,7 @@ public Observable> window(Func0RxJava Wiki: window() */ public Observable> window(Observable windowOpenings, Func1> closingSelector) { return create(OperationWindow.window(this, windowOpenings, closingSelector)); @@ -2925,11 +3093,12 @@ public Observable> window(Observable windowOpen * encounters an error, the current window is emitted, and the event is * propagated. *

- * + * * * @param count the maximum size of each window before it should be emitted * @return an {@link Observable} that produces connected, non-overlapping * windows containing at most count items + * @see RxJava Wiki: window() */ public Observable> window(int count) { return create(OperationWindow.window(this, count)); @@ -2942,7 +3111,7 @@ public Observable> window(int count) { * completes or encounters an error, the current window is emitted and the * event is propagated. *

- * + * * * @param count the maximum size of each window before it should be emitted * @param skip how many items need to be skipped before starting a new @@ -2950,6 +3119,7 @@ public Observable> window(int count) { * are equal this is the same operation as {@link #window(int)}. * @return an {@link Observable} that produces windows every "skipped" * items containing at most count items + * @see RxJava Wiki: window() */ public Observable> window(int count, int skip) { return create(OperationWindow.window(this, count, skip)); @@ -2962,7 +3132,7 @@ public Observable> window(int count, int skip) { * Observable completes or encounters an error, the current window is * emitted and the event is propagated. *

- * + * * * @param timespan the period of time each window collects items before it * should be emitted and replaced with a new window @@ -2970,6 +3140,7 @@ public Observable> window(int count, int skip) { * argument * @return an {@link Observable} that produces connected, non-overlapping * windows with a fixed duration + * @see RxJava Wiki: window() */ public Observable> window(long timespan, TimeUnit unit) { return create(OperationWindow.window(this, timespan, unit)); @@ -2982,7 +3153,7 @@ public Observable> window(long timespan, TimeUnit unit) { * source Observable completes or encounters an error, the current window is * emitted and the event is propagated. *

- * + * * * @param timespan the period of time each window collects items before it * should be emitted and replaced with a new window @@ -2992,6 +3163,7 @@ public Observable> window(long timespan, TimeUnit unit) { * and start of a window * @return an {@link Observable} that produces connected, non-overlapping * windows with a fixed duration + * @see RxJava Wiki: window() */ public Observable> window(long timespan, TimeUnit unit, Scheduler scheduler) { return create(OperationWindow.window(this, timespan, unit, scheduler)); @@ -3005,7 +3177,7 @@ public Observable> window(long timespan, TimeUnit unit, Scheduler * reached first). When the source Observable completes or encounters an * error, the current window is emitted and the event is propagated. *

- * + * * * @param timespan the period of time each window collects values before it * should be emitted and replaced with a new window @@ -3015,6 +3187,7 @@ public Observable> window(long timespan, TimeUnit unit, Scheduler * @return an {@link Observable} that produces connected, non-overlapping * windows that are emitted after a fixed duration or when the * window has reached maximum capacity (whichever occurs first) + * @see RxJava Wiki: window() */ public Observable> window(long timespan, TimeUnit unit, int count) { return create(OperationWindow.window(this, timespan, unit, count)); @@ -3028,7 +3201,7 @@ public Observable> window(long timespan, TimeUnit unit, int count) * first). When the source Observable completes or encounters an error, the * current window is emitted and the event is propagated. *

- * + * * * @param timespan the period of time each window collects values before it * should be emitted and replaced with a new window @@ -3040,6 +3213,7 @@ public Observable> window(long timespan, TimeUnit unit, int count) * @return an {@link Observable} that produces connected non-overlapping * windows that are emitted after a fixed duration or when the * window has reached maximum capacity (whichever occurs first). + * @see RxJava Wiki: window() */ public Observable> window(long timespan, TimeUnit unit, int count, Scheduler scheduler) { return create(OperationWindow.window(this, timespan, unit, count, scheduler)); @@ -3053,7 +3227,7 @@ public Observable> window(long timespan, TimeUnit unit, int count, * source Observable completes or encounters an error, the current window is * emitted and the event is propagated. *

- * + * * * @param timespan the period of time each window collects values before it * should be emitted @@ -3063,6 +3237,7 @@ public Observable> window(long timespan, TimeUnit unit, int count, * and timeshift arguments * @return an {@link Observable} that produces new windows periodically and * emits these after a fixed timespan has elapsed + * @see RxJava Wiki: window() */ public Observable> window(long timespan, long timeshift, TimeUnit unit) { return create(OperationWindow.window(this, timespan, timeshift, unit)); @@ -3076,7 +3251,7 @@ public Observable> window(long timespan, long timeshift, TimeUnit * source Observable completes or encounters an error, the current window is * emitted and the event is propagated. *

- * + * * * @param timespan the period of time each window collects values before it * should be emitted @@ -3088,6 +3263,7 @@ public Observable> window(long timespan, long timeshift, TimeUnit * and start of a window * @return an {@link Observable} that produces new windows periodically and * emits these after a fixed timespan has elapsed + * @see RxJava Wiki: window() */ public Observable> window(long timespan, long timeshift, TimeUnit unit, Scheduler scheduler) { return create(OperationWindow.window(this, timespan, timeshift, unit, scheduler)); @@ -3116,6 +3292,7 @@ public Observable> window(long timespan, long timeshift, TimeUnit * each of the source Observables, results in an item * that will be emitted by the resulting Observable * @return an Observable that emits the zipped results + * @see RxJava Wiki: zip() */ public static Observable zip(Observable> ws, final FuncN zipFunction) { return ws.toList().mapMany(new Func1>, Observable>() { @@ -3149,6 +3326,7 @@ public Observable call(List> wsList) { * each of the source Observables, results in an item * that will be emitted by the resulting Observable * @return an Observable that emits the zipped results + * @see RxJava Wiki: zip() */ public static Observable zip(Iterable> ws, FuncN zipFunction) { return create(OperationZip.zip(ws, zipFunction)); @@ -3164,6 +3342,7 @@ public static Observable zip(Iterable> ws, FuncN< * the filter * @return an Observable that emits only those items in the original * Observable that the filter evaluates as {@code true} + * @see RxJava Wiki: filter() */ public Observable filter(Func1 predicate) { return create(OperationFilter.filter(this, predicate)); @@ -3176,7 +3355,8 @@ public Observable filter(Func1 predicate) { * * * @return an Observable of sequentially distinct items - * @see MSDN: Observable.distinctUntilChanged + * @see RxJava Wiki: distinctUntilChanged() + * @see MSDN: Observable.distinctUntilChanged */ public Observable distinctUntilChanged() { return create(OperationDistinctUntilChanged.distinctUntilChanged(this)); @@ -3193,7 +3373,8 @@ public Observable distinctUntilChanged() { * value that is used for deciding whether an item is * sequentially distinct from another one or not * @return an Observable of sequentially distinct items - * @see MSDN: Observable.distinctUntilChanged + * @see RxJava Wiki: distinctUntilChanged() + * @see MSDN: Observable.distinctUntilChanged */ public Observable distinctUntilChanged(Func1 keySelector) { return create(OperationDistinctUntilChanged.distinctUntilChanged(this, keySelector)); @@ -3206,7 +3387,8 @@ public Observable distinctUntilChanged(Func1 keyS * * * @return an Observable of distinct items - * @see MSDN: Observable.distinct + * @see RxJava Wiki: distinct() + * @see MSDN: Observable.distinct */ public Observable distinct() { return create(OperationDistinct.distinct(this)); @@ -3222,7 +3404,8 @@ public Observable distinct() { * value that is used to decide whether an item is * distinct from another one or not * @return an Observable that emits distinct items - * @see MSDN: Observable.distinct + * @see RxJava Wiki: distinct() + * @see MSDN: Observable.distinct */ public Observable distinct(Func1 keySelector) { return create(OperationDistinct.distinct(this, keySelector)); @@ -3240,6 +3423,7 @@ public Observable distinct(Func1 keySelector) { * or equal to the number of elements in * the source sequence * @throws IndexOutOfBoundsException if index is less than 0 + * @see RxJava Wiki: elementAt() */ public Observable elementAt(int index) { return create(OperationElementAt.elementAt(this, index)); @@ -3257,6 +3441,7 @@ public Observable elementAt(int index) { * the source sequence, or the default item if the index is outside * the bounds of the source sequence * @throws IndexOutOfBoundsException if index is less than 0 + * @see RxJava Wiki: elementAtOrDefault() */ public Observable elementAtOrDefault(int index, T defaultValue) { return create(OperationElementAt.elementAtOrDefault(this, index, defaultValue)); @@ -3275,6 +3460,7 @@ public Observable elementAtOrDefault(int index, T defaultValue) { * * @param predicate the condition to test every element * @return a subscription function for creating the target Observable + * @see RxJava Wiki: exists() * @see MSDN: Observable.Any Note: the description in this page is wrong. */ public Observable exists(Func1 predicate) { @@ -3289,6 +3475,7 @@ public Observable exists(Func1 predicate) { * @param element the item to search in the sequence * @return an Observable that emits true if the item is in the * source sequence + * @see RxJava Wiki: contains() * @see MSDN: Observable.Contains */ public Observable contains(final T element) { @@ -3304,12 +3491,13 @@ public Boolean call(T t1) { * {@link Observer#onCompleted onCompleted} or * {@link Observer#onError onError}. *

- * + * * * @param action an {@link Action0} to be invoked when the source * Observable finishes * @return an Observable that emits the same items as the source Observable, * then invokes the {@link Action0} + * @see RxJava Wiki: finallyDo() * @see MSDN: Observable.Finally Method */ public Observable finallyDo(Action0 action) { @@ -3332,6 +3520,7 @@ public Observable finallyDo(Action0 action) { * transformation function to each item emitted by the source * Observable and merging the results of the Observables obtained * from this transformation. + * @see RxJava Wiki: flatMap() * @see #mapMany(Func1) */ public Observable flatMap(Func1> func) { @@ -3348,6 +3537,7 @@ public Observable flatMap(Func1RxJava Wiki: where() * @see #filter(Func1) */ public Observable where(Func1 predicate) { @@ -3363,7 +3553,8 @@ public Observable where(Func1 predicate) { * @param func a function to apply to each item emitted by the Observable * @return an Observable that emits the items from the source Observable, * transformed by the given function - * @see MSDN: Observable.Select + * @see RxJava Wiki: map() + * @see MSDN: Observable.Select */ public Observable map(Func1 func) { return create(OperationMap.map(this, func)); @@ -3380,7 +3571,8 @@ public Observable map(Func1 func) { * additional parameter. * @return an Observable that emits the items from the source Observable, * transformed by the given function - * @see MSDN: Observable.Select + * @see RxJava Wiki: mapWithIndex() + * @see MSDN: Observable.Select */ public Observable mapWithIndex(Func2 func) { return create(OperationMap.mapWithIndex(this, func)); @@ -3402,6 +3594,7 @@ public Observable mapWithIndex(Func2 fun * transformation function to each item emitted by the source * Observable and merging the results of the Observables obtained * from this transformation. + * @see RxJava Wiki: mapMany() * @see #flatMap(Func1) */ public Observable mapMany(Func1> func) { @@ -3417,6 +3610,7 @@ public Observable mapMany(Func1RxJava Wiki: materialize() * @see MSDN: Observable.materialize */ public Observable> materialize() { @@ -3427,12 +3621,13 @@ public Observable> materialize() { * Asynchronously subscribes and unsubscribes Observers on the specified * {@link Scheduler}. *

- * + * * * @param scheduler the {@link Scheduler} to perform subscription and * unsubscription actions on * @return the source Observable modified so that its subscriptions and * unsubscriptions happen on the specified {@link Scheduler} + * @see RxJava Wiki: subscribeOn() */ public Observable subscribeOn(Scheduler scheduler) { return create(OperationSubscribeOn.subscribeOn(this, scheduler)); @@ -3442,11 +3637,12 @@ public Observable subscribeOn(Scheduler scheduler) { * Asynchronously notify {@link Observer}s on the specified * {@link Scheduler}. *

- * + * * * @param scheduler the {@link Scheduler} to notify {@link Observer}s on * @return the source Observable modified so that its {@link Observer}s are * notified on the specified {@link Scheduler} + * @see RxJava Wiki: observeOn() */ public Observable observeOn(Scheduler scheduler) { return create(OperationObserveOn.observeOn(this, scheduler)); @@ -3458,13 +3654,14 @@ public Observable observeOn(Scheduler scheduler) { * objects emitted by the source Observable into the items or notifications * they represent. *

- * + * * * @return an Observable that emits the items and notifications embedded in * the {@link Notification} objects emitted by the source Observable - * @see MSDN: Observable.dematerialize * @throws Throwable if the source Observable is not of type - * {@code Observable>}. + * {@code Observable>} + * @see RxJava Wiki: dematerialize() + * @see MSDN: Observable.dematerialize */ @SuppressWarnings("unchecked") public Observable dematerialize() { @@ -3498,6 +3695,7 @@ public Observable dematerialize() { * take over if the source Observable encounters an * error * @return the original Observable, with appropriately modified behavior + * @see RxJava Wiki: onErrorResumeNext() */ public Observable onErrorResumeNext(final Func1> resumeFunction) { return create(OperationOnErrorResumeNextViaFunction.onErrorResumeNextViaFunction(this, resumeFunction)); @@ -3530,6 +3728,7 @@ public Observable onErrorResumeNext(final Func1RxJava Wiki: onErrorResumeNext() */ public Observable onErrorResumeNext(final Observable resumeSequence) { return create(OperationOnErrorResumeNextViaObservable.onErrorResumeNextViaObservable(this, resumeSequence)); @@ -3567,6 +3766,7 @@ public Observable onErrorResumeNext(final Observable resumeSeque * take over if the source Observable encounters an * error * @return the original Observable, with appropriately modified behavior + * @see RxJava Wiki: onExceptionResumeNextViaObservable() */ public Observable onExceptionResumeNext(final Observable resumeSequence) { return create(OperationOnExceptionResumeNextViaObservable.onExceptionResumeNextViaObservable(this, resumeSequence)); @@ -3577,7 +3777,7 @@ public Observable onExceptionResumeNext(final Observable resumeS * rather than invoking {@link Observer#onError onError} if it encounters an * error. *

- * + * *

* By default, when an Observable encounters an error that prevents it from * emitting the expected item to its {@link Observer}, the Observable @@ -3597,6 +3797,7 @@ public Observable onExceptionResumeNext(final Observable resumeS * Observable will emit if the source Observable * encounters an error * @return the original Observable with appropriately modified behavior + * @see RxJava Wiki: onErrorReturn() */ public Observable onErrorReturn(Func1 resumeFunction) { return create(OperationOnErrorReturn.onErrorReturn(this, resumeFunction)); @@ -3623,7 +3824,8 @@ public Observable onErrorReturn(Func1 resumeFunction) * @return an Observable that emits a single item that is the result of * accumulating the output from the source Observable * @throws IllegalArgumentException if the Observable sequence is empty - * @see MSDN: Observable.Aggregate + * @see RxJava Wiki: reduce() + * @see MSDN: Observable.Aggregate * @see Wikipedia: Fold (higher-order function) */ public Observable reduce(Func2 accumulator) { @@ -3643,7 +3845,8 @@ public Observable reduce(Func2 accumulator) { * * @return an Observable that emits the number of counted elements of the * source Observable as its single item - * @see MSDN: Observable.Count + * @see RxJava Wiki: count() + * @see MSDN: Observable.Count */ public Observable count() { return reduce(0, new Func2() { @@ -3663,7 +3866,8 @@ public Integer call(Integer t1, T t2) { * @param source source Observable to compute the sum of * @return an Observable that emits the sum of all the items of the * source Observable as its single item - * @see MSDN: Observable.Sum + * @see RxJava Wiki: sum() + * @see MSDN: Observable.Sum */ public static Observable sum(Observable source) { return OperationSum.sum(source); @@ -3678,7 +3882,8 @@ public static Observable sum(Observable source) { * @param source source Observable to compute the sum of * @return an Observable that emits the sum of all the items of the * source Observable as its single item - * @see MSDN: Observable.Sum + * @see RxJava Wiki: sumLongs() + * @see MSDN: Observable.Sum */ public static Observable sumLongs(Observable source) { return OperationSum.sumLongs(source); @@ -3693,7 +3898,8 @@ public static Observable sumLongs(Observable source) { * @param source source Observable to compute the sum of * @return an Observable that emits the sum of all the items of the * source Observable as its single item - * @see MSDN: Observable.Sum + * @see RxJava Wiki: sumFloats() + * @see MSDN: Observable.Sum */ public static Observable sumFloats(Observable source) { return OperationSum.sumFloats(source); @@ -3708,7 +3914,8 @@ public static Observable sumFloats(Observable source) { * @param source source Observable to compute the sum of * @return an Observable that emits the sum of all the items of the * source Observable as its single item - * @see MSDN: Observable.Sum + * @see RxJava Wiki: sumDoubles() + * @see MSDN: Observable.Sum */ public static Observable sumDoubles(Observable source) { return OperationSum.sumDoubles(source); @@ -3724,7 +3931,8 @@ public static Observable sumDoubles(Observable source) { * @return an Observable that emits the average of all the items emitted by * the source Observable as its single item * @throws IllegalArgumentException if the Observable sequence is empty - * @see MSDN: Observable.Average + * @see RxJava Wiki: average() + * @see MSDN: Observable.Average */ public static Observable average(Observable source) { return OperationAverage.average(source); @@ -3739,7 +3947,8 @@ public static Observable average(Observable source) { * @param source source observable to compute the average of * @return an Observable that emits the average of all the items emitted by * the source Observable as its single item - * @see MSDN: Observable.Average + * @see RxJava Wiki: averageLongs() + * @see MSDN: Observable.Average */ public static Observable averageLongs(Observable source) { return OperationAverage.averageLongs(source); @@ -3754,7 +3963,8 @@ public static Observable averageLongs(Observable source) { * @param source source observable to compute the average of * @return an Observable that emits the average of all the items emitted by * the source Observable as its single item - * @see MSDN: Observable.Average + * @see RxJava Wiki: averageFloats() + * @see MSDN: Observable.Average */ public static Observable averageFloats(Observable source) { return OperationAverage.averageFloats(source); @@ -3769,7 +3979,8 @@ public static Observable averageFloats(Observable source) { * @param source source observable to compute the average of * @return an Observable that emits the average of all the items emitted by * the source Observable as its single item - * @see MSDN: Observable.Average + * @see RxJava Wiki: averageDoubles() + * @see MSDN: Observable.Average */ public static Observable averageDoubles(Observable source) { return OperationAverage.averageDoubles(source); @@ -3786,7 +3997,7 @@ public static Observable averageDoubles(Observable source) { * @throws IllegalArgumentException if the source is empty * @see MSDN: Observable.Min */ - public static > Observable min(Observable source) { + public static > Observable min(Observable source) { return OperationMinMax.min(source); } @@ -3801,9 +4012,10 @@ public static > Observable min(Observable source) * @return an Observable that emits the minimum value according to the * specified comparator * @throws IllegalArgumentException if the source is empty + * @see RxJava Wiki: min() * @see MSDN: Observable.Min */ - public Observable min(Comparator comparator) { + public Observable min(Comparator comparator) { return OperationMinMax.min(this, comparator); } @@ -3817,9 +4029,10 @@ public Observable min(Comparator comparator) { * @param selector the key selector function * @return an Observable that emits a List of the items with the minimum key * value + * @see RxJava Wiki: minBy() * @see MSDN: Observable.MinBy */ - public > Observable> minBy(Func1 selector) { + public > Observable> minBy(Func1 selector) { return OperationMinMax.minBy(this, selector); } @@ -3834,9 +4047,10 @@ public > Observable> minBy(Func1 selector) * @param comparator the comparator used to compare key values * @return an Observable that emits a List of the elements with the minimum * key value according to the specified comparator + * @see RxJava Wiki: minBy() * @see MSDN: Observable.MinBy */ - public Observable> minBy(Func1 selector, Comparator comparator) { + public Observable> minBy(Func1 selector, Comparator comparator) { return OperationMinMax.minBy(this, selector, comparator); } @@ -3849,9 +4063,10 @@ public Observable> minBy(Func1 selector, Comparator compara * @param source an Observable to determine the maximum item of * @return an Observable that emits the maximum element * @throws IllegalArgumentException if the source is empty + * @see RxJava Wiki: max() * @see MSDN: Observable.Max */ - public static > Observable max(Observable source) { + public static > Observable max(Observable source) { return OperationMinMax.max(source); } @@ -3866,9 +4081,10 @@ public static > Observable max(Observable source) * @return an Observable that emits the maximum item according to the * specified comparator * @throws IllegalArgumentException if the source is empty + * @see RxJava Wiki: max() * @see MSDN: Observable.Max */ - public Observable max(Comparator comparator) { + public Observable max(Comparator comparator) { return OperationMinMax.max(this, comparator); } @@ -3881,9 +4097,10 @@ public Observable max(Comparator comparator) { * @param selector the key selector function * @return an Observable that emits a List of the items with the maximum key * value + * @see RxJava Wiki: maxBy() * @see MSDN: Observable.MaxBy */ - public > Observable> maxBy(Func1 selector) { + public > Observable> maxBy(Func1 selector) { return OperationMinMax.maxBy(this, selector); } @@ -3898,9 +4115,10 @@ public > Observable> maxBy(Func1 selector) * @param comparator the comparator used to compare key values * @return an Observable that emits a List of the elements with the maximum * key value according to the specified comparator + * @see RxJava Wiki: maxBy() * @see MSDN: Observable.MaxBy */ - public Observable> maxBy(Func1 selector, Comparator comparator) { + public Observable> maxBy(Func1 selector, Comparator comparator) { return OperationMinMax.maxBy(this, selector, comparator); } @@ -3913,6 +4131,7 @@ public Observable> maxBy(Func1 selector, Comparator compara * * @return a {@link ConnectableObservable} that upon connection causes the * source Observable to emit items to its {@link Observer}s + * @see RxJava Wiki: replay() */ public ConnectableObservable replay() { return OperationMulticast.multicast(this, ReplaySubject. create()); @@ -3935,6 +4154,7 @@ public ConnectableObservable replay() { * * @param retryCount number of retry attempts before failing * @return an Observable with retry logic + * @see RxJava Wiki: retry() */ public Observable retry(int retryCount) { return create(OperationRetry.retry(this, retryCount)); @@ -3957,6 +4177,7 @@ public Observable retry(int retryCount) { * output would be [1, 2, 1, 2, 3, 4, 5, onCompleted]. * * @return an Observable with retry logic + * @see RxJava Wiki: retry() */ public Observable retry() { return create(OperationRetry.retry(this)); @@ -3967,7 +4188,7 @@ public Observable retry() { * auto-subscribes to the source Observable rather than returning a * {@link ConnectableObservable}. *

- * + * *

* This is useful when you want an Observable to cache responses and you * can't control the subscribe/unsubscribe behavior of all the @@ -3980,6 +4201,7 @@ public Observable retry() { * * @return an Observable that, when first subscribed to, caches all of its * notifications for the benefit of subsequent subscribers. + * @see RxJava Wiki: cache() */ public Observable cache() { return create(OperationCache.cache(this)); @@ -3997,6 +4219,7 @@ public Observable cache() { * {@code Observable} * @return an Observable with the output of the {@link Func1} executed on a * {@link Scheduler} + * @see RxJava Wiki: parallel() */ public Observable parallel(Func1, Observable> f) { return OperationParallel.parallel(this, f); @@ -4014,22 +4237,76 @@ public Observable parallel(Func1, Observable> f) { * @param s a {@link Scheduler} to perform the work on * @return an Observable with the output of the {@link Func1} executed on a * {@link Scheduler} + * @see RxJava Wiki: parallel() */ public Observable parallel(final Func1, Observable> f, final Scheduler s) { return OperationParallel.parallel(this, f, s); } + + /** + * Merges an Observable<Observable<T>> to + * Observable<Observable<T>> with the number of + * inner Observables defined by parallelObservables. + *

+ * For example, if the original + * Observable<Observable<T>> has 100 Observables to + * be emitted and parallelObservables is 8, the 100 will be + * grouped onto 8 output Observables. + *

+ * This is a mechanism for efficiently processing n number of + * Observables on a smaller m number of resources (typically CPU + * cores). + *

+ * + * + * @param parallelObservables the number of Observables to merge into + * @return an Observable of Observables constrained to number defined by + * parallelObservables + * @see RxJava Wiki: parallelMerge() + */ + public static Observable> parallelMerge(Observable> source, int parallelObservables) { + return OperationParallelMerge.parallelMerge(source, parallelObservables); + } + + /** + * Merges an Observable<Observable<T>> to + * Observable<Observable<T>> with the number of + * inner Observables defined by parallelObservables and runs + * each Observable on the defined Scheduler. + *

+ * For example, if the original + * Observable<Observable<T>> has 100 Observables to + * be emitted and parallelObservables is 8, the 100 will be + * grouped onto 8 output Observables. + *

+ * This is a mechanism for efficiently processing n number of + * Observables on a smaller m number of resources (typically CPU + * cores). + *

+ * + * + * @param parallelObservables the number of Observables to merge into + * @return an Observable of Observables constrained to number defined by + * parallelObservables. + * @see RxJava Wiki: parallelMerge() + */ + public static Observable> parallelMerge(Observable> source, int parallelObservables, Scheduler scheduler) { + return OperationParallelMerge.parallelMerge(source, parallelObservables, scheduler); + } + /** * Returns a {@link ConnectableObservable}, which waits until its * {@link ConnectableObservable#connect connect} method is called before it * begins emitting items to those {@link Observer}s that have subscribed to * it. *

- * + * * * @return a {@link ConnectableObservable} that upon connection causes the * source Observable to emit items to its {@link Observer}s + * @see RxJava Wiki: publish() */ public ConnectableObservable publish() { return OperationMulticast.multicast(this, PublishSubject. create()); @@ -4039,9 +4316,10 @@ public ConnectableObservable publish() { * Returns a {@link ConnectableObservable} that shares a single subscription * that contains the last notification only. *

- * + * * * @return a {@link ConnectableObservable} + * @see RxJava Wiki: publishLast() */ public ConnectableObservable publishLast() { return OperationMulticast.multicast(this, AsyncSubject. create()); @@ -4051,7 +4329,8 @@ public ConnectableObservable publishLast() { * Synonymous with reduce(). *

* - * + * + * @see RxJava Wiki: aggregate() * @see #reduce(Func2) */ public Observable aggregate(Func2 accumulator) { @@ -4080,6 +4359,7 @@ public Observable aggregate(Func2 accumulator) { * @return an Observable that emits a single item that is the result of * accumulating the output from the items emitted by the source * Observable + * @see RxJava Wiki: reduce() * @see MSDN: Observable.Aggregate * @see Wikipedia: Fold (higher-order function) */ @@ -4092,6 +4372,7 @@ public Observable reduce(R initialValue, Func2 accumulat *

* * + * @see RxJava Wiki: aggregate() * @see #reduce(Object, Func2) */ public Observable aggregate(R initialValue, Func2 accumulator) { @@ -4119,7 +4400,8 @@ public Observable aggregate(R initialValue, Func2 accumu * accumulator call * @return an Observable that emits the results of each call to the * accumulator function - * @see MSDN: Observable.Scan + * @see RxJava Wiki: scan() + * @see MSDN: Observable.Scan */ public Observable scan(Func2 accumulator) { return create(OperationScan.scan(this, accumulator)); @@ -4129,12 +4411,13 @@ public Observable scan(Func2 accumulator) { * Returns an Observable that emits the results of sampling the items * emitted by the source Observable at a specified time interval. *

- * + * * * @param period the sampling rate * @param unit the {@link TimeUnit} in which period is defined * @return an Observable that emits the results of sampling the items * emitted by the source Observable at the specified time interval + * @see RxJava Wiki: sample() */ public Observable sample(long period, TimeUnit unit) { return create(OperationSample.sample(this, period, unit)); @@ -4144,13 +4427,14 @@ public Observable sample(long period, TimeUnit unit) { * Returns an Observable that emits the results of sampling the items * emitted by the source Observable at a specified time interval. *

- * + * * * @param period the sampling rate * @param unit the {@link TimeUnit} in which period is defined * @param scheduler the {@link Scheduler} to use when sampling * @return an Observable that emits the results of sampling the items * emitted by the source Observable at the specified time interval + * @see RxJava Wiki: sample() */ public Observable sample(long period, TimeUnit unit, Scheduler scheduler) { return create(OperationSample.sample(this, period, unit, scheduler)); @@ -4178,7 +4462,8 @@ public Observable sample(long period, TimeUnit unit, Scheduler scheduler) { * accumulator call * @return an Observable that emits the results of each call to the * accumulator function - * @see MSDN: Observable.Scan + * @see RxJava Wiki: scan() + * @see MSDN: Observable.Scan */ public Observable scan(R initialValue, Func2 accumulator) { return create(OperationScan.scan(this, initialValue, accumulator)); @@ -4188,12 +4473,13 @@ public Observable scan(R initialValue, Func2 accumulator * Returns an Observable that emits a Boolean that indicates whether all of * the items emitted by the source Observable satisfy a condition. *

- * + * * * @param predicate a function that evaluates an item and returns a Boolean * @return an Observable that emits true if all items emitted * by the source Observable satisfy the predicate; otherwise, * false + * @see RxJava Wiki: all() */ public Observable all(Func1 predicate) { return create(OperationAll.all(this, predicate)); @@ -4213,6 +4499,7 @@ public Observable all(Func1 predicate) { * @return an Observable that is identical to the source Observable except * that it does not emit the first num items that the * source emits + * @see RxJava Wiki: skip() */ public Observable skip(int num) { return create(OperationSkip.skip(this, num)); @@ -4226,8 +4513,9 @@ public Observable skip(int num) { * * @return an Observable that emits only the very first item from the * source, or none if the source Observable completes without - * emitting a single item. - * @see MSDN: Observable.First + * emitting a single item + * @see RxJava Wiki: first() + * @see MSDN: Observable.First */ public Observable first() { return take(1); @@ -4242,8 +4530,9 @@ public Observable first() { * @param predicate the condition any source emitted item has to satisfy * @return an Observable that emits only the very first item satisfying the * given condition from the source, or none if the source Observable - * completes without emitting a single matching item. - * @see MSDN: Observable.First + * completes without emitting a single matching item + * @see RxJava Wiki: first() + * @see MSDN: Observable.First */ public Observable first(Func1 predicate) { return skipWhile(not(predicate)).take(1); @@ -4260,7 +4549,8 @@ public Observable first(Func1 predicate) { * @return an Observable that emits only the very first item from the * source, or a default value if the source Observable completes * without emitting a single item - * @see MSDN: Observable.FirstOrDefault + * @see RxJava Wiki: firstOrDefault() + * @see MSDN: Observable.FirstOrDefault */ public Observable firstOrDefault(T defaultValue) { return create(OperationFirstOrDefault.firstOrDefault(this, defaultValue)); @@ -4278,7 +4568,8 @@ public Observable firstOrDefault(T defaultValue) { * doesn't emit anything that satisfies the given condition * @return an Observable that emits only the very first item from the source * that satisfies the given condition, or a default value otherwise - * @see MSDN: Observable.FirstOrDefault + * @see RxJava Wiki: firstOrDefault() + * @see MSDN: Observable.FirstOrDefault */ public Observable firstOrDefault(Func1 predicate, T defaultValue) { return create(OperationFirstOrDefault.firstOrDefault(this, predicate, defaultValue)); @@ -4288,12 +4579,12 @@ public Observable firstOrDefault(Func1 predicate, T defau * Returns the elements of the specified sequence or the specified default * value in a singleton sequence if the sequence is empty. *

- * + * * * @param defaultValue the value to return if the sequence is empty * @return an Observable that emits the specified default value if the * source is empty; otherwise, the items emitted by the source - * + * @see RxJava Wiki: defaultIfEmpty() * @see MSDN: Observable.DefaultIfEmpty */ public Observable defaultIfEmpty(T defaultValue) { @@ -4316,6 +4607,7 @@ public Observable defaultIfEmpty(T defaultValue) { * from the source Observable, or all of the items from the source * Observable if that Observable emits fewer than num * items + * @see RxJava Wiki: take() */ public Observable take(final int num) { return create(OperationTake.take(this, num)); @@ -4325,13 +4617,14 @@ public Observable take(final int num) { * Returns an Observable that emits items emitted by the source Observable * so long as a specified condition is true. *

- * + * * * @param predicate a function that evaluates an item emitted by the source * Observable and returns a Boolean * @return an Observable that emits the items from the source Observable so * long as each item satisfies the condition defined by * predicate + * @see RxJava Wiki: takeWhile() */ public Observable takeWhile(final Func1 predicate) { return create(OperationTakeWhile.takeWhile(this, predicate)); @@ -4342,7 +4635,7 @@ public Observable takeWhile(final Func1 predicate) { * so long as a given predicate remains true, where the predicate can * operate on both the item and its index relative to the complete sequence. *

- * + * * * @param predicate a function to test each item emitted by the source * Observable for a condition; the second parameter of the @@ -4350,6 +4643,7 @@ public Observable takeWhile(final Func1 predicate) { * @return an Observable that emits items from the source Observable so long * as the predicate continues to return true for each * item, then completes + * @see RxJava Wiki: takeWhileWithIndex() */ public Observable takeWhileWithIndex(final Func2 predicate) { return create(OperationTakeWhile.takeWhileWithIndex(this, predicate)); @@ -4363,8 +4657,9 @@ public Observable takeWhileWithIndex(final Func2MSDN: Observable.First + * emitting a single item + * @see RxJava Wiki: first() + * @see MSDN: Observable.First * @see #first() */ public Observable takeFirst() { @@ -4380,8 +4675,9 @@ public Observable takeFirst() { * @param predicate the condition any source emitted item has to satisfy * @return an Observable that emits only the very first item satisfying the * given condition from the source, or none if the source Observable - * completes without emitting a single matching item. - * @see MSDN: Observable.First + * completes without emitting a single matching item + * @see RxJava Wiki: first() + * @see MSDN: Observable.First * @see #first(Func1) */ public Observable takeFirst(Func1 predicate) { @@ -4392,12 +4688,13 @@ public Observable takeFirst(Func1 predicate) { * Returns an Observable that emits only the last count items * emitted by the source Observable. *

- * + * * * @param count the number of items to emit from the end of the sequence * emitted by the source Observable * @return an Observable that emits only the last count items * emitted by the source Observable + * @see RxJava Wiki: takeLast() */ public Observable takeLast(final int count) { return create(OperationTakeLast.takeLast(this, count)); @@ -4407,7 +4704,7 @@ public Observable takeLast(final int count) { * Returns an Observable that emits the items from the source Observable * only until the other Observable emits an item. *

- * + * * * @param other the Observable whose first emitted item will cause * takeUntil to stop emitting items from the @@ -4415,6 +4712,7 @@ public Observable takeLast(final int count) { * @param the type of items emitted by other * @return an Observable that emits the items of the source Observable until * such time as other emits its first item + * @see RxJava Wiki: takeUntil() */ public Observable takeUntil(Observable other) { return OperationTakeUntil.takeUntil(this, other); @@ -4433,7 +4731,8 @@ public Observable takeUntil(Observable other) { * as a second parameter. * @return an Observable that emits all items from the source Observable as * soon as the condition becomes false - * @see MSDN: Observable.SkipWhile + * @see RxJava Wiki: skipWhileWithIndex() + * @see MSDN: Observable.SkipWhile */ public Observable skipWhileWithIndex(Func2 predicate) { return create(OperationSkipWhile.skipWhileWithIndex(this, predicate)); @@ -4450,7 +4749,8 @@ public Observable skipWhileWithIndex(Func2 predi * Observable for a condition * @return an Observable that emits all items from the source Observable as * soon as the condition becomes false - * @see MSDN: Observable.SkipWhile + * @see RxJava Wiki: skipWhile() + * @see MSDN: Observable.SkipWhile */ public Observable skipWhile(Func1 predicate) { return create(OperationSkipWhile.skipWhile(this, predicate)); @@ -4465,15 +4765,14 @@ public Observable skipWhile(Func1 predicate) { * from the front of the queue and produced on the result sequence. This * causes elements to be delayed. *

- * + * * * @param count number of elements to bypass at the end of the source * sequence * @return an Observable sequence emitting the source sequence items * except for the bypassed ones at the end - * * @throws IndexOutOfBoundsException if count is less than zero - * + * @see RxJava Wiki: skipLast() * @see MSDN: Observable.SkipLast */ public Observable skipLast(int count) { @@ -4500,6 +4799,7 @@ public Observable skipLast(int count) { * * @return an Observable that emits a single item: a List containing all of * the items emitted by the source Observable. + * @see RxJava Wiki: toList() */ public Observable> toList() { return create(OperationToObservableList.toObservableList(this)); @@ -4518,6 +4818,7 @@ public Observable> toList() { * all other items emitted by the Observable * @return an Observable that emits the items from the source Observable in * sorted order + * @see RxJava Wiki: toSortedList() */ public Observable> toSortedList() { return create(OperationToObservableSortedList.toSortedList(this)); @@ -4534,6 +4835,7 @@ public Observable> toSortedList() { * indicates their sort order * @return an Observable that emits the items from the source Observable in * sorted order + * @see RxJava Wiki: toSortedList() */ public Observable> toSortedList(Func2 sortFunction) { return create(OperationToObservableSortedList.toSortedList(this, sortFunction)); @@ -4548,11 +4850,45 @@ public Observable> toSortedList(Func2 sor * @param values Iterable of the items you want the modified Observable to * emit first * @return an Observable that exhibits the modified behavior + * @see RxJava Wiki: startWith() */ public Observable startWith(Iterable values) { return concat(Observable. from(values), this); } + /** + * Emit a specified set of items with the specified scheduler before + * beginning to emit items from the source Observable. + *

+ * + * + * @param values iterable of the items you want the modified Observable to + * emit first + * @param scheduler the scheduler to emit the prepended values on + * @return an Observable that exhibits the modified behavior + * @see RxJava Wiki: startWith() + * @see MSDN: Observable.StartWith + */ + public Observable startWith(Iterable values, Scheduler scheduler) { + return concat(from(values, scheduler), this); + } + + /** + * Emit a specified array of items with the specified scheduler before + * beginning to emit items from the source Observable. + *

+ * + * + * @param values the items you want the modified Observable to emit first + * @param scheduler the scheduler to emit the prepended values on + * @return an Observable that exhibits the modified behavior + * @see RxJava Wiki: startWith() + * @see MSDN: Observable.StartWith + */ + public Observable startWith(T[] values, Scheduler scheduler) { + return startWith(Arrays.asList(values), scheduler); + } + /** * Emit a specified item before beginning to emit items from the source * Observable. @@ -4561,6 +4897,7 @@ public Observable startWith(Iterable values) { * * @param t1 item to emit * @return an Observable that exhibits the modified behavior + * @see RxJava Wiki: startWith() */ public Observable startWith(T t1) { return concat(Observable. from(t1), this); @@ -4575,6 +4912,7 @@ public Observable startWith(T t1) { * @param t1 first item to emit * @param t2 second item to emit * @return an Observable that exhibits the modified behavior + * @see RxJava Wiki: startWith() */ public Observable startWith(T t1, T t2) { return concat(Observable. from(t1, t2), this); @@ -4590,6 +4928,7 @@ public Observable startWith(T t1, T t2) { * @param t2 second item to emit * @param t3 third item to emit * @return an Observable that exhibits the modified behavior + * @see RxJava Wiki: startWith() */ public Observable startWith(T t1, T t2, T t3) { return concat(Observable. from(t1, t2, t3), this); @@ -4606,6 +4945,7 @@ public Observable startWith(T t1, T t2, T t3) { * @param t3 third item to emit * @param t4 fourth item to emit * @return an Observable that exhibits the modified behavior + * @see RxJava Wiki: startWith() */ public Observable startWith(T t1, T t2, T t3, T t4) { return concat(Observable. from(t1, t2, t3, t4), this); @@ -4623,6 +4963,7 @@ public Observable startWith(T t1, T t2, T t3, T t4) { * @param t4 fourth item to emit * @param t5 fifth item to emit * @return an Observable that exhibits the modified behavior + * @see RxJava Wiki: startWith() */ public Observable startWith(T t1, T t2, T t3, T t4, T t5) { return concat(Observable. from(t1, t2, t3, t4, t5), this); @@ -4641,6 +4982,7 @@ public Observable startWith(T t1, T t2, T t3, T t4, T t5) { * @param t5 fifth item to emit * @param t6 sixth item to emit * @return an Observable that exhibits the modified behavior + * @see RxJava Wiki: startWith() */ public Observable startWith(T t1, T t2, T t3, T t4, T t5, T t6) { return concat(Observable. from(t1, t2, t3, t4, t5, t6), this); @@ -4660,6 +5002,7 @@ public Observable startWith(T t1, T t2, T t3, T t4, T t5, T t6) { * @param t6 sixth item to emit * @param t7 seventh item to emit * @return an Observable that exhibits the modified behavior + * @see RxJava Wiki: startWith() */ public Observable startWith(T t1, T t2, T t3, T t4, T t5, T t6, T t7) { return concat(Observable. from(t1, t2, t3, t4, t5, t6, t7), this); @@ -4680,6 +5023,7 @@ public Observable startWith(T t1, T t2, T t3, T t4, T t5, T t6, T t7) { * @param t7 seventh item to emit * @param t8 eighth item to emit * @return an Observable that exhibits the modified behavior + * @see RxJava Wiki: startWith() */ public Observable startWith(T t1, T t2, T t3, T t4, T t5, T t6, T t7, T t8) { return concat(Observable. from(t1, t2, t3, t4, t5, t6, t7, t8), this); @@ -4701,6 +5045,7 @@ public Observable startWith(T t1, T t2, T t3, T t4, T t5, T t6, T t7, T t8) { * @param t8 eighth item to emit * @param t9 ninth item to emit * @return an Observable that exhibits the modified behavior + * @see RxJava Wiki: startWith() */ public Observable startWith(T t1, T t2, T t3, T t4, T t5, T t6, T t7, T t8, T t9) { return concat(Observable. from(t1, t2, t3, t4, t5, t6, t7, t8, t9), this); @@ -4711,7 +5056,7 @@ public Observable startWith(T t1, T t2, T t3, T t4, T t5, T t6, T t7, T t8, T * criterion, and emits these grouped items as {@link GroupedObservable}s, * one GroupedObservable per group. *

- * + * * * @param keySelector a function that extracts the key from an item * @param elementSelector a function to map a source item to an item in a @@ -4723,6 +5068,7 @@ public Observable startWith(T t1, T t2, T t3, T t4, T t5, T t6, T t7, T t8, T * which corresponds to a unique key value and emits items * representing items from the source Observable that share that key * value + * @see RxJava Wiki: groupBy */ public Observable> groupBy(final Func1 keySelector, final Func1 elementSelector) { return create(OperationGroupBy.groupBy(this, keySelector, elementSelector)); @@ -4733,7 +5079,7 @@ public Observable> groupBy(final Func1 - * + * * * @param keySelector a function that extracts the key for each item * @param the key type @@ -4741,6 +5087,7 @@ public Observable> groupBy(final Func1RxJava Wiki: groupBy */ public Observable> groupBy(final Func1 keySelector) { return create(OperationGroupBy.groupBy(this, keySelector)); @@ -4753,9 +5100,10 @@ public Observable> groupBy(final Func1any operator but renamed in * RxJava to better match Java naming idioms. *

- * + * * * @return an Observable that emits a Boolean + * @see RxJava Wiki: isEmpty() * @see MSDN: Observable.Any */ public Observable isEmpty() { @@ -4767,9 +5115,10 @@ public Observable isEmpty() { * source or an IllegalArgumentException if the source * {@link Observable} is empty. *

- * + * * * @return + * @see RxJava Wiki: last() */ public Observable last() { return create(OperationLast.last(this)); @@ -4780,7 +5129,7 @@ public Observable last() { * with blocking operators). * * @return - * @see Blocking Observable Operators + * @see RxJava Wiki: Blocking Observable Operators */ public BlockingObservable toBlockingObservable() { return BlockingObservable.from(this); @@ -4789,11 +5138,12 @@ public BlockingObservable toBlockingObservable() { /** * Converts the items emitted by an Observable to the specified type. *

- * + * * * @param klass the target class type which the items will be converted to * @return an Observable that emits each item from the source Observable * converted to the specified type + * @see RxJava Wiki: cast() * @see MSDN: Observable.Cast */ public Observable cast(final Class klass) { @@ -4803,12 +5153,13 @@ public Observable cast(final Class klass) { /** * Filters the items emitted by an Observable based on the specified type. *

- * + * * * @param klass the class type to filter the items emitted by the source * Observable * @return an Observable that emits items from the source Observable of * type klass. + * @see RxJava Wiki: ofType() * @see MSDN: Observable.OfType */ public Observable ofType(final Class klass) { @@ -4823,11 +5174,11 @@ public Boolean call(T t) { * Ignores all items emitted by an Observable and only calls * onCompleted or onError. *

- * + * * * @return an empty Observable that only calls onCompleted or * onError - * + * @see RxJava Wiki: ignoreElements() * @see MSDN: Observable.IgnoreElements */ public Observable ignoreElements() { @@ -4840,13 +5191,14 @@ public Observable ignoreElements() { * isn't received within the specified timeout duration starting from its * predecessor, a TimeoutException is propagated to the observer. *

- * + * * * @param timeout maximum duration between values before a timeout occurs * @param timeUnit the unit of time which applies to the * timeout argument. * @return the source Observable with a TimeoutException in * case of a timeout + * @see RxJava Wiki: timeout() * @see MSDN: Observable.Timeout */ public Observable timeout(long timeout, TimeUnit timeUnit) { @@ -4860,7 +5212,7 @@ public Observable timeout(long timeout, TimeUnit timeUnit) { * predecessor, the other observable sequence is used to produce future * messages from that point on. *

- * + * * * @param timeout maximum duration between values before a timeout occurs * @param timeUnit the unit of time which applies to the @@ -4868,6 +5220,7 @@ public Observable timeout(long timeout, TimeUnit timeUnit) { * @param other sequence to return in case of a timeout * @return the source sequence switching to the other sequence in case of a * timeout + * @see RxJava Wiki: timeout() * @see MSDN: Observable.Timeout */ public Observable timeout(long timeout, TimeUnit timeUnit, Observable other) { @@ -4880,7 +5233,7 @@ public Observable timeout(long timeout, TimeUnit timeUnit, Observable - * + * * * @param timeout maximum duration between values before a timeout occurs * @param timeUnit the unit of time which applies to the @@ -4888,6 +5241,7 @@ public Observable timeout(long timeout, TimeUnit timeUnit, ObservableTimeoutException in case * of a timeout + * @see RxJava Wiki: timeout() * @see MSDN: Observable.Timeout */ public Observable timeout(long timeout, TimeUnit timeUnit, Scheduler scheduler) { @@ -4901,7 +5255,7 @@ public Observable timeout(long timeout, TimeUnit timeUnit, Scheduler schedule * predecessor, the other observable sequence is used to produce future * messages from that point on. *

- * + * * * @param timeout maximum duration between values before a timeout occurs * @param timeUnit the unit of time which applies to the @@ -4910,6 +5264,7 @@ public Observable timeout(long timeout, TimeUnit timeUnit, Scheduler schedule * @param scheduler Scheduler to run the timeout timers on * @return the source sequence switching to the other sequence in case of a * timeout + * @see RxJava Wiki: timeout() * @see MSDN: Observable.Timeout */ public Observable timeout(long timeout, TimeUnit timeUnit, Observable other, Scheduler scheduler) { @@ -4920,9 +5275,10 @@ public Observable timeout(long timeout, TimeUnit timeUnit, Observable - * + * * * @return an Observable that emits time interval information items + * @see RxJava Wiki: timeInterval() * @see MSDN: Observable.TimeInterval */ public Observable> timeInterval() { @@ -4933,10 +5289,11 @@ public Observable> timeInterval() { * Records the time interval between consecutive items emitted by an * Observable, using the specified Scheduler to compute time intervals. *

- * + * * * @param scheduler Scheduler used to compute time intervals * @return an Observable that emits time interval information items + * @see RxJava Wiki: timeInterval() * @see MSDN: Observable.TimeInterval */ public Observable> timeInterval(Scheduler scheduler) { @@ -4946,13 +5303,14 @@ public Observable> timeInterval(Scheduler scheduler) { /** * Constructs an Observable that depends on a resource object. *

- * + * * * @param resourceFactory the factory function to obtain a resource object * that depends on the Observable * @param observableFactory the factory function to obtain an Observable * @return the Observable whose lifetime controls the lifetime of the * dependent resource object + * @see RxJava Wiki: using() * @see MSDN: Observable.Using */ public static Observable using(Func0 resourceFactory, Func1> observableFactory) { @@ -4962,12 +5320,13 @@ public static Observable using(Func0 - * + * * * @param o1 an Observable competing to react first * @param o2 an Observable competing to react first * @return an Observable that reflects whichever of the given Observables * reacted first + * @see RxJava Wiki: amb() * @see MSDN: Observable.Amb */ public static Observable amb(Observable o1, Observable o2) { @@ -4977,13 +5336,14 @@ public static Observable amb(Observable o1, Observable - * + * * * @param o1 an Observable competing to react first * @param o2 an Observable competing to react first * @param o3 an Observable competing to react first * @return an Observable that reflects whichever of the given Observables * reacted first + * @see RxJava Wiki: amb() * @see MSDN: Observable.Amb */ public static Observable amb(Observable o1, Observable o2, Observable o3) { @@ -4993,7 +5353,7 @@ public static Observable amb(Observable o1, Observable - * + * * * @param o1 an Observable competing to react first * @param o2 an Observable competing to react first @@ -5001,6 +5361,7 @@ public static Observable amb(Observable o1, ObservableRxJava Wiki: amb() * @see MSDN: Observable.Amb */ public static Observable amb(Observable o1, Observable o2, Observable o3, Observable o4) { @@ -5010,7 +5371,7 @@ public static Observable amb(Observable o1, Observable - * + * * * @param o1 an Observable competing to react first * @param o2 an Observable competing to react first @@ -5019,6 +5380,7 @@ public static Observable amb(Observable o1, ObservableRxJava Wiki: amb() * @see MSDN: Observable.Amb */ public static Observable amb(Observable o1, Observable o2, Observable o3, Observable o4, Observable o5) { @@ -5028,7 +5390,7 @@ public static Observable amb(Observable o1, Observable - * + * * * @param o1 an Observable competing to react first * @param o2 an Observable competing to react first @@ -5038,6 +5400,7 @@ public static Observable amb(Observable o1, ObservableRxJava Wiki: amb() * @see MSDN: Observable.Amb */ public static Observable amb(Observable o1, Observable o2, Observable o3, Observable o4, Observable o5, Observable o6) { @@ -5047,7 +5410,7 @@ public static Observable amb(Observable o1, Observable - * + * * * @param o1 an Observable competing to react first * @param o2 an Observable competing to react first @@ -5058,6 +5421,7 @@ public static Observable amb(Observable o1, ObservableRxJava Wiki: amb() * @see MSDN: Observable.Amb */ public static Observable amb(Observable o1, Observable o2, Observable o3, Observable o4, Observable o5, Observable o6, Observable o7) { @@ -5067,7 +5431,7 @@ public static Observable amb(Observable o1, Observable - * + * * * @param o1 an Observable competing to react first * @param o2 an Observable competing to react first @@ -5079,6 +5443,7 @@ public static Observable amb(Observable o1, ObservableRxJava Wiki: amb() * @see MSDN: Observable.Amb */ public static Observable amb(Observable o1, Observable o2, Observable o3, Observable o4, Observable o5, Observable o6, Observable o7, Observable o8) { @@ -5088,7 +5453,7 @@ public static Observable amb(Observable o1, Observable - * + * * * @param o1 an Observable competing to react first * @param o2 an Observable competing to react first @@ -5101,6 +5466,7 @@ public static Observable amb(Observable o1, ObservableRxJava Wiki: amb() * @see MSDN: Observable.Amb */ public static Observable amb(Observable o1, Observable o2, Observable o3, Observable o4, Observable o5, Observable o6, Observable o7, Observable o8, Observable o9) { @@ -5110,26 +5476,27 @@ public static Observable amb(Observable o1, Observable - * + * * * @param sources Observable sources competing to react first * @return an Observable that reflects whichever of the given Observables * reacted first + * @see RxJava Wiki: amb() * @see MSDN: Observable.Amb */ public static Observable amb(Iterable> sources) { return create(OperationAmb.amb(sources)); } - /** * Invokes an action for each item emitted by the Observable. *

- * + * * * @param observer the action to invoke for each item emitted in the source * sequence * @return the source sequence with the side-effecting behavior applied + * @see RxJava Wiki: doOnEach() * @see MSDN: Observable.Do */ public Observable doOnEach(Observer observer) { @@ -5139,11 +5506,12 @@ public Observable doOnEach(Observer observer) { /** * Invokes an action for each item emitted by an Observable. *

- * + * * * @param onNext the action to invoke for each item in the source * sequence * @return the source sequence with the side-effecting behavior applied + * @see RxJava Wiki: doOnEach() * @see MSDN: Observable.Do */ public Observable doOnEach(final Action1 onNext) { @@ -5168,10 +5536,11 @@ public void onNext(T args) { /** * Invokes an action if onError is called from the Observable. *

- * + * * * @param onError the action to invoke if onError is invoked * @return the source sequence with the side-effecting behavior applied + * @see RxJava Wiki: doOnError() * @see MSDN: Observable.Do */ public Observable doOnError(final Action1 onError) { @@ -5197,11 +5566,12 @@ public void onNext(T args) { } * Invokes an action when onCompleted is called by the * Observable. *

- * + * * * @param onCompleted the action to invoke when onCompleted is * called * @return the source sequence with the side-effecting behavior applied + * @see RxJava Wiki: doOnCompleted() * @see MSDN: Observable.Do */ public Observable doOnCompleted(final Action0 onCompleted) { @@ -5230,6 +5600,7 @@ public void onNext(T args) { } * @param onError the action to invoke when the source Observable calls * onError * @return the source sequence with the side-effecting behavior applied + * @see RxJava Wiki: doOnEach() * @see MSDN: Observable.Do */ public Observable doOnEach(final Action1 onNext, final Action1 onError) { @@ -5253,7 +5624,6 @@ public void onNext(T args) { return create(OperationDoOnEach.doOnEach(this, observer)); } - /** * Invokes an action for each item emitted by an Observable. * @@ -5263,6 +5633,7 @@ public void onNext(T args) { * @param onCompleted the action to invoke when the source Observable calls * onCompleted * @return the source sequence with the side-effecting behavior applied + * @see RxJava Wiki: doOnEach() * @see MSDN: Observable.Do */ public Observable doOnEach(final Action1 onNext, final Action1 onError, final Action0 onCompleted) { @@ -5326,4 +5697,426 @@ private boolean isInternalImplementation(Object o) { } } + /** + * Creates a pattern that matches when both Observable sequences have an + * available item. + *

+ * + * + * @param right Observable sequence to match with the left sequence + * @return Pattern object that matches when both Observable sequences have + * an available item + * @throws NullPointerException if right is null + * @see RxJava Wiki: and() + * @see MSDN: Observable.And + */ + public Pattern2 and(Observable right) { + return OperationJoinPatterns.and(this, right); + } + + /** + * Matches when the Observable sequence has an available item and + * projects the item by invoking the selector function. + *

+ * + * + * @param selector Selector that will be invoked for elements in the source + * sequence + * @return Plan that produces the projected results, to be fed (with other + * plans) to the When operator + * @throws NullPointerException if selector is null + * @see RxJava Wiki: then() + * @see MSDN: Observable.Then + */ + public Plan0 then(Func1 selector) { + return OperationJoinPatterns.then(this, selector); + } + + /** + * Joins together the results from several patterns. + *

+ * + * + * @param plans a series of plans created by use of the Then operator on + * patterns + * @return an Observable sequence with the results from matching several + * patterns + * @throws NullPointerException if plans is null + * @see RxJava Wiki: when() + * @see MSDN: Observable.When + */ + public static Observable when(Plan0... plans) { + return create(OperationJoinPatterns.when(plans)); + } + + /** + * Joins together the results from several patterns. + *

+ * + * + * @param plans a series of plans created by use of the Then operator on + * patterns + * @return an Observable sequence with the results from matching several + * patterns + * @throws NullPointerException if plans is null + * @see RxJava Wiki: when() + * @see MSDN: Observable.When + */ + public static Observable when(Iterable> plans) { + if (plans == null) { + throw new NullPointerException("plans"); + } + return create(OperationJoinPatterns.when(plans)); + } + + /** + * Joins the results from a pattern. + *

+ * + * + * @param p1 the plan to join + * @return an Observable sequence with the results from matching a pattern + * @see RxJava Wiki: when() + * @see MSDN: Observable.When + */ + @SuppressWarnings("unchecked") + public static Observable when(Plan0 p1) { + return create(OperationJoinPatterns.when(p1)); + } + + /** + * Joins together the results from several patterns. + *

+ * + * + * @param p1 a plan + * @param p2 a plan + * @return an Observable sequence with the results from matching several + * patterns + * @see RxJava Wiki: when() + * @see MSDN: Observable.When + */ + @SuppressWarnings("unchecked") + public static Observable when(Plan0 p1, Plan0 p2) { + return create(OperationJoinPatterns.when(p1, p2)); + } + + /** + * Joins together the results from several patterns. + *

+ * + * + * @param p1 a plan + * @param p2 a plan + * @param p3 a plan + * @return an Observable sequence with the results from matching several + * patterns + * @see RxJava Wiki: when() + * @see MSDN: Observable.When + */ + @SuppressWarnings("unchecked") + public static Observable when(Plan0 p1, Plan0 p2, Plan0 p3) { + return create(OperationJoinPatterns.when(p1, p2, p3)); + } + + /** + * Joins together the results from several patterns. + *

+ * + * + * @param p1 a plan + * @param p2 a plan + * @param p3 a plan + * @param p4 a plan + * @return an Observable sequence with the results from matching several + * patterns + * @see RxJava Wiki: when() + * @see MSDN: Observable.When + */ + @SuppressWarnings("unchecked") + public static Observable when(Plan0 p1, Plan0 p2, Plan0 p3, Plan0 p4) { + return create(OperationJoinPatterns.when(p1, p2, p3, p4)); + } + + /** + * Joins together the results from several patterns. + *

+ * + * + * @param p1 a plan + * @param p2 a plan + * @param p3 a plan + * @param p4 a plan + * @param p5 a plan + * @return an Observable sequence with the results from matching several + * patterns + * @see RxJava Wiki: when() + * @see MSDN: Observable.When + */ + @SuppressWarnings("unchecked") + public static Observable when(Plan0 p1, Plan0 p2, Plan0 p3, Plan0 p4, Plan0 p5) { + return create(OperationJoinPatterns.when(p1, p2, p3, p4, p5)); + } + + /** + * Joins together the results from several patterns. + *

+ * + * + * @param p1 a plan + * @param p2 a plan + * @param p3 a plan + * @param p4 a plan + * @param p5 a plan + * @param p6 a plan + * @return an Observable sequence with the results from matching several + * patterns + * @see RxJava Wiki: when() + * @see MSDN: Observable.When + */ + @SuppressWarnings("unchecked") + public static Observable when(Plan0 p1, Plan0 p2, Plan0 p3, Plan0 p4, Plan0 p5, Plan0 p6) { + return create(OperationJoinPatterns.when(p1, p2, p3, p4, p5, p6)); + } + + /** + * Joins together the results from several patterns. + *

+ * + * + * @param p1 a plan + * @param p2 a plan + * @param p3 a plan + * @param p4 a plan + * @param p5 a plan + * @param p6 a plan + * @param p7 a plan + * @return an Observable sequence with the results from matching several + * patterns + * @see RxJava Wiki: when() + * @see MSDN: Observable.When + */ + @SuppressWarnings("unchecked") + public static Observable when(Plan0 p1, Plan0 p2, Plan0 p3, Plan0 p4, Plan0 p5, Plan0 p6, Plan0 p7) { + return create(OperationJoinPatterns.when(p1, p2, p3, p4, p5, p6, p7)); + } + + /** + * Joins together the results from several patterns. + *

+ * + * + * @param p1 a plan + * @param p2 a plan + * @param p3 a plan + * @param p4 a plan + * @param p5 a plan + * @param p6 a plan + * @param p7 a plan + * @param p8 a plan + * @return an Observable sequence with the results from matching several + * patterns + * @see RxJava Wiki: when() + * @see MSDN: Observable.When + */ + @SuppressWarnings("unchecked") + public static Observable when(Plan0 p1, Plan0 p2, Plan0 p3, Plan0 p4, Plan0 p5, Plan0 p6, Plan0 p7, Plan0 p8) { + return create(OperationJoinPatterns.when(p1, p2, p3, p4, p5, p6, p7, p8)); + } + + /** + * Joins together the results from several patterns. + *

+ * + * + * @param p1 a plan + * @param p2 a plan + * @param p3 a plan + * @param p4 a plan + * @param p5 a plan + * @param p6 a plan + * @param p7 a plan + * @param p8 a plan + * @param p9 a plan + * @return an Observable sequence with the results from matching several + * patterns + * @see RxJava Wiki: when() + * @see MSDN: Observable.When + */ + @SuppressWarnings("unchecked") + public static Observable when(Plan0 p1, Plan0 p2, Plan0 p3, Plan0 p4, Plan0 p5, Plan0 p6, Plan0 p7, Plan0 p8, Plan0 p9) { + return create(OperationJoinPatterns.when(p1, p2, p3, p4, p5, p6, p7, p8, p9)); + } + + /** + * Correlates the elements of two sequences based on overlapping durations. + *

+ * + * + * @param right the right observable sequence to join elements for + * @param leftDurationSelector a function to select the duration of each + * element of this observable sequence, used to + * determine overlap + * @param rightDurationSelector a function to select the duration of each + * element of the right observable sequence, + * used to determine overlap + * @param resultSelector a function invoked to compute a result element + * for any two overlapping elements of the left and + * right observable sequences + * @return an observable sequence that contains result elements computed + * from source elements that have an overlapping duration + * @see RxJava Wiki: join() + * @see MSDN: Observable.Join + */ + public Observable join(Observable right, Func1> leftDurationSelector, + Func1> rightDurationSelector, + Func2 resultSelector) { + return create(new OperationJoin(this, right, leftDurationSelector, rightDurationSelector, resultSelector)); + } + + /** + * Return an Observable that emits a single HashMap containing all items + * emitted by the source Observable, mapped by the keys returned by the + * {@code keySelector} function. + *

+ * + * + * If a source item maps to the same key, the HashMap will contain the + * latest of those items. + * + * @param keySelector the function that extracts the key from the source + * items to be used as keys in the HashMap + * @return an Observable that emits a single HashMap containing the mapped + * values of the source Observable + * @see RxJava Wiki: toMap() + * @see MSDN: Observable.ToDictionary + */ + public Observable> toMap(Func1 keySelector) { + return create(OperationToMap.toMap(this, keySelector)); + } + + /** + * Return an Observable that emits a single HashMap containing elements with + * key and value extracted from the values emitted by the source Observable. + *

+ * + *

+ * If a source item maps to the same key, the HashMap will contain the + * latest of those items. + * + * @param keySelector the function that extracts the key from the source + * items to be used as key in the HashMap + * @param valueSelector the function that extracts the value from the source + * items to be used as value in the HashMap + * @return an Observable that emits a single HashMap containing the mapped + * values of the source Observable + * @see RxJava Wiki: toMap() + * @see MSDN: Observable.ToDictionary + */ + public Observable> toMap(Func1 keySelector, Func1 valueSelector) { + return create(OperationToMap.toMap(this, keySelector, valueSelector)); + } + + /** + * Return an Observable that emits a single Map, returned by the + * mapFactory function, containing key and value extracted from + * the values emitted by the source Observable. + *

+ * + * + * @param keySelector the function that extracts the key from the source + * items to be used as key in the Map + * @param valueSelector the function that extracts the value from the source + * items to be used as value in the Map + * @param mapFactory the function that returns an Map instance to be used + * @return an Observable that emits a single Map containing the mapped + * values of the source Observable + * @see RxJava Wiki: toMap() + */ + public Observable> toMap(Func1 keySelector, Func1 valueSelector, Func0> mapFactory) { + return create(OperationToMap.toMap(this, keySelector, valueSelector, mapFactory)); + } + + /** + * Return an Observable that emits a single HashMap containing an ArrayList + * of elements, emitted by the source Observable and keyed by the + * keySelector function. + *

+ * + * + * @param keySelector the function that extracts the key from the source + * items to be used as key in the HashMap + * @return an Observable that emits a single HashMap containing an ArrayList + * of elements mapped from the source Observable + * @see RxJava Wiki: toMultiMap() + * @see MSDN: Observable.ToLookup + */ + public Observable>> toMultimap(Func1 keySelector) { + return create(OperationToMultimap.toMultimap(this, keySelector)); + } + + /** + * Return an Observable that emits a single HashMap containing an ArrayList + * of values, extracted by the valueSelector function, emitted + * by the source Observable and keyed by the keySelector + * function. + *

+ * + * + * @param keySelector the function that extracts the key from the source + * items to be used as key in the HashMap + * @param valueSelector the function that extracts the value from the source + * items to be used as value in the Map + * @return an Observable that emits a single HashMap containing an ArrayList + * of elements mapped from the source Observable + * @see RxJava Wiki: toMultiMap() + * @see MSDN: Observable.ToLookup + */ + public Observable>> toMultimap(Func1 keySelector, Func1 valueSelector) { + return create(OperationToMultimap.toMultimap(this, keySelector, valueSelector)); + } + + /** + * Return an Observable that emits a single Map, returned by the + * mapFactory function, containing an ArrayList of values, + * extracted by the valueSelector function, emitted by the + * source Observable and keyed by the keySelector function. + *

+ * + * + * @param keySelector the function that extracts the key from the source + * items to be used as key in the Map + * @param valueSelector the function that extracts the value from the source + * items to be used as value in the Map + * @param mapFactory the function that returns an Map instance to be used + * @return an Observable that emits a single Map containing the list of + * mapped values of the source observable. + * @see RxJava Wiki: toMultiMap() + */ + public Observable>> toMultimap(Func1 keySelector, Func1 valueSelector, Func0>> mapFactory) { + return create(OperationToMultimap.toMultimap(this, keySelector, valueSelector, mapFactory)); + } + + /** + * Return an Observable that emits a single Map, returned by the + * mapFactory function, containing a custom collection of + * values, extracted by the valueSelector function, emitted by + * the source Observable and keyed by the keySelector function. + *

+ * + * + * @param keySelector the function that extracts the key from the source + * items to be used as key in the Map + * @param valueSelector the function that extracts the value from the source + * items to be used as value in the Map + * @param mapFactory the function that returns an Map instance to be used + * @param collectionFactory the function that returns a Collection instance + * for a particular key to be used in the Map + * @return an Observable that emits a single Map containing the collection + * of mapped values of the source Observable. + * @see RxJava Wiki: toMultiMap() + */ + public Observable>> toMultimap(Func1 keySelector, Func1 valueSelector, Func0>> mapFactory, Func1> collectionFactory) { + return create(OperationToMultimap.toMultimap(this, keySelector, valueSelector, mapFactory, collectionFactory)); + } }