/**

* Copyright 2014 Netflix, Inc.

*

* Licensed under the Apache License, Version 2.0 (the "License"); you may not

* use this file except in compliance with the License. You may obtain a copy of

* the License at

*

* http://www.apache.org/licenses/LICENSE-2.0

*

* Unless required by applicable law or agreed to in writing, software

* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT

* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the

* License for the specific language governing permissions and limitations

* under the License.

*/

package rx;

import static rx.util.functions.Functions.*;

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.joins.Pattern2;

import rx.joins.Plan0;

import rx.observables.BlockingObservable;

import rx.observables.ConnectableObservable;

import rx.observables.GroupedObservable;

import rx.observers.SafeObserver;

import rx.operators.OperationAll;

import rx.operators.OperationAmb;

import rx.operators.OperationAny;

import rx.operators.OperationAsObservable;

import rx.operators.OperationAverage;

import rx.operators.OperationBuffer;

import rx.operators.OperationCache;

import rx.operators.OperationCombineLatest;

import rx.operators.OperationConcat;

import rx.operators.OperationDebounce;

import rx.operators.OperationDefaultIfEmpty;

import rx.operators.OperationDefer;

import rx.operators.OperationDelay;

import rx.operators.OperationDematerialize;

import rx.operators.OperationDistinct;

import rx.operators.OperationDistinctUntilChanged;

import rx.operators.OperationDoOnEach;

import rx.operators.OperationElementAt;

import rx.operators.OperationFilter;

import rx.operators.OperationFinally;

import rx.operators.OperationFlatMap;

import rx.operators.OperationGroupByUntil;

import rx.operators.OperationGroupJoin;

import rx.operators.OperationInterval;

import rx.operators.OperationJoin;

import rx.operators.OperationJoinPatterns;

import rx.operators.OperationMaterialize;

import rx.operators.OperationMergeDelayError;

import rx.operators.OperationMinMax;

import rx.operators.OperationMulticast;

import rx.operators.OperationObserveOn;

import rx.operators.OperationOnErrorResumeNextViaFunction;

import rx.operators.OperationOnErrorResumeNextViaObservable;

import rx.operators.OperationOnErrorReturn;

import rx.operators.OperationOnExceptionResumeNextViaObservable;

import rx.operators.OperationParallelMerge;

import rx.operators.OperationRepeat;

import rx.operators.OperationReplay;

import rx.operators.OperationRetry;

import rx.operators.OperationSample;

import rx.operators.OperationScan;

import rx.operators.OperationSequenceEqual;

import rx.operators.OperationSingle;

import rx.operators.OperationSkip;

import rx.operators.OperationSkipLast;

import rx.operators.OperationSkipUntil;

import rx.operators.OperationSkipWhile;

import rx.operators.OperationSubscribeOn;

import rx.operators.OperationSum;

import rx.operators.OperationSwitch;

import rx.operators.OperationSynchronize;

import rx.operators.OperationTakeLast;

import rx.operators.OperationTakeUntil;

import rx.operators.OperationTakeWhile;

import rx.operators.OperationThrottleFirst;

import rx.operators.OperationTimeInterval;

import rx.operators.OperationTimeout;

import rx.operators.OperationTimer;

import rx.operators.OperationToMap;

import rx.operators.OperationToMultimap;

import rx.operators.OperationToObservableFuture;

import rx.operators.OperationUsing;

import rx.operators.OperationWindow;

import rx.operators.OperationZip;

import rx.operators.OperatorCast;

import rx.operators.OperatorFromIterable;

import rx.operators.OperatorGroupBy;

import rx.operators.OperatorMap;

import rx.operators.OperatorMerge;

import rx.operators.OperatorParallel;

import rx.operators.OperatorTake;

import rx.operators.OperationTakeTimed;

import rx.operators.OperatorTimestamp;

import rx.operators.OperatorToObservableList;

import rx.operators.OperatorToObservableSortedList;

import rx.operators.SafeObservableSubscription;

import rx.plugins.RxJavaObservableExecutionHook;

import rx.plugins.RxJavaPlugins;

import rx.schedulers.Schedulers;

import rx.subjects.AsyncSubject;

import rx.subjects.BehaviorSubject;

import rx.subjects.PublishSubject;

import rx.subjects.ReplaySubject;

import rx.subjects.Subject;

import rx.subscriptions.Subscriptions;

import rx.util.OnErrorNotImplementedException;

import rx.util.Range;

import rx.util.TimeInterval;

import rx.util.Timestamped;

import rx.util.functions.Action0;

import rx.util.functions.Action1;

import rx.util.functions.Action2;

import rx.util.functions.Func0;

import rx.util.functions.Func1;

import rx.util.functions.Func2;

import rx.util.functions.Func3;

import rx.util.functions.Func4;

import rx.util.functions.Func5;

import rx.util.functions.Func6;

import rx.util.functions.Func7;

import rx.util.functions.Func8;

import rx.util.functions.Func9;

import rx.util.functions.FuncN;

import rx.util.functions.Function;

import rx.util.functions.Functions;

/**

* The Observable class that implements the Reactive Pattern.

* <p>

* This class provides methods for subscribing to the Observable as well as delegate methods to the

* various Observers.

* <p>

* The documentation for this class makes use of marble diagrams. The following legend explains

* these diagrams:

* <p>

* <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/legend.png">

* <p>

* For more information see the

* <a href="https://github.com/Netflix/RxJava/wiki/Observable">RxJava Wiki</a>

*

* @param <T>

* the type of the items emitted by the Observable

*/

public class Observable<T> {

final OnSubscribe<T> f;

/**

* Observable with Function to execute when subscribed to.

* <p>

* <em>Note:</em> Use {@link #create(OnSubscribeFunc)} to create an Observable, instead of this

* constructor, unless you specifically have a need for inheritance.

*

* @param onSubscribe

* {@link OnSubscribeFunc} to be executed when {@link #subscribe(Observer)} is called

*/

protected Observable(OnSubscribe<T> f) {

this.f = f;

}

private final static RxJavaObservableExecutionHook hook = RxJavaPlugins.getInstance().getObservableExecutionHook();

/**

* Returns an Observable that will execute the specified function when an {@link Observer} subscribes to it.

* <p>

* <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/create.png">

* <p>

* Write the function you pass to {@code create} so that it behaves as an Observable: It should

* invoke the Observer's {@link Observer#onNext onNext}, {@link Observer#onError onError}, and {@link Observer#onCompleted onCompleted} methods appropriately.

* <p>

* A well-formed Observable must invoke either the Observer's {@code onCompleted} method

* exactly once or its {@code onError} method exactly once.

* <p>

* See <a href="http://go.microsoft.com/fwlink/?LinkID=205219">Rx Design Guidelines (PDF)</a>

* for detailed information.

*

* @param <T>

* the type of the items that this Observable emits

* @param func

* a function that accepts an {@code Observer<T>}, invokes its {@code onNext}, {@code onError}, and {@code onCompleted} methods as appropriate, and returns a {@link Subscription} that

* allows the Observer to cancel the subscription

* @return an Observable that, when an {@link Observer} subscribes to it, will execute the

* specified function

* @see <a href="https://github.com/Netflix/RxJava/wiki/Creating-Observables#create">RxJava Wiki: create()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/system.reactive.linq.observable.create.aspx">MSDN: Observable.Create</a>

*/

public final static <T> Observable<T> create(OnSubscribe<T> f) {

return new Observable<T>(f);

}

public static interface OnSubscribe<T> extends Action1<Observer<? super T>> {

}

public final static <T> Observable<T> create(final OnSubscribeFunc<T> f) {

return new Observable<T>(new OnSubscribe<T>() {

@Override

public void call(Observer<? super T> observer) {

Subscription s = f.onSubscribe(observer);

if (s != null && s != observer) {

observer.add(s);

}

}

});

}

public static interface OnSubscribeFunc<T> extends Function {

public Subscription onSubscribe(Observer<? super T> op);

}

/**

* Lift a function to the current Observable and return a new Observable that when subscribed to will pass the values of the current Observable through the function.

* <p>

* In other words, this allows chaining Observers together on an Observable for acting on the values within the Observable.

* <p> {@code

* observable.map(...).filter(...).take(5).lift(new ObserverA()).lift(new ObserverB(...)).subscribe()

* }

*

* @param bind

* @return an Observable that emits values that are the result of applying the bind function to the values of the current Observable

*/

public <R> Observable<R> lift(final Func1<Observer<? super R>, Observer<? super T>> bind) {

return new Observable<R>(new OnSubscribe<R>() {

@Override

public void call(Observer<? super R> o) {

subscribe(bind.call(o));

}

});

}

/* ******************************************************************************

* Observers Below Here

* ******************************************************************************

*/

/**

* Mirror the one Observable in an Iterable of several Observables that first emits an item.

* <p>

* <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/amb.png">

*

* @param sources

* an Iterable of Observable sources competing to react first

* @return an Observable that emits the same sequence of items as whichever of the source

* Observables first emitted an item

* @see <a href="https://github.com/Netflix/RxJava/wiki/Conditional-and-Boolean-Operators#amb">RxJava Wiki: amb()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh229115.aspx">MSDN: Observable.Amb</a>

*/

public final static <T> Observable<T> amb(Iterable<? extends Observable<? extends T>> sources) {

return create(OperationAmb.amb(sources));

}

/**

* Given two Observables, mirror the one that first emits an item.

* <p>

* <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/amb.png">

*

* @param o1

* an Observable competing to react first

* @param o2

* an Observable competing to react first

* @return an Observable that emits the same sequence of items as whichever of the source

* Observables first emitted an item

* @see <a href="https://github.com/Netflix/RxJava/wiki/Conditional-and-Boolean-Operators#amb">RxJava Wiki: amb()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh229733.aspx">MSDN: Observable.Amb</a>

*/

public final static <T> Observable<T> amb(Observable<? extends T> o1, Observable<? extends T> o2) {

return create(OperationAmb.amb(o1, o2));

}

/**

* Given three Observables, mirror the one that first emits an item.

* <p>

* <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/amb.png">

*

* @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 emits the same sequence of items as whichever of the source

* Observables first emitted an item

* @see <a href="https://github.com/Netflix/RxJava/wiki/Conditional-and-Boolean-Operators#amb">RxJava Wiki: amb()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh229733.aspx">MSDN: Observable.Amb</a>

*/

public final static <T> Observable<T> amb(Observable<? extends T> o1, Observable<? extends T> o2, Observable<? extends T> o3) {

return create(OperationAmb.amb(o1, o2, o3));

}

/**

* Given four Observables, mirror the one that first emits an item.

* <p>

* <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/amb.png">

*

* @param o1

* an Observable competing to react first

* @param o2

* an Observable competing to react first

* @param o3

* an Observable competing to react first

* @param o4

* an Observable competing to react first

* @return an Observable that emits the same sequence of items as whichever of the source

* Observables first emitted an item

* @see <a href="https://github.com/Netflix/RxJava/wiki/Conditional-and-Boolean-Operators#amb">RxJava Wiki: amb()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh229733.aspx">MSDN: Observable.Amb</a>

*/

public final static <T> Observable<T> amb(Observable<? extends T> o1, Observable<? extends T> o2, Observable<? extends T> o3, Observable<? extends T> o4) {

return create(OperationAmb.amb(o1, o2, o3, o4));

}

/**

* Given five Observables, mirror the one that first emits an item.

* <p>

* <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/amb.png">

*

* @param o1

* an Observable competing to react first

* @param o2

* an Observable competing to react first

* @param o3

* an Observable competing to react first

* @param o4

* an Observable competing to react first

* @param o5

* an Observable competing to react first

* @return an Observable that emits the same sequence of items as whichever of the source

* Observables first emitted an item

* @see <a href="https://github.com/Netflix/RxJava/wiki/Conditional-and-Boolean-Operators#amb">RxJava Wiki: amb()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh229733.aspx">MSDN: Observable.Amb</a>

*/

public final static <T> Observable<T> amb(Observable<? extends T> o1, Observable<? extends T> o2, Observable<? extends T> o3, Observable<? extends T> o4, Observable<? extends T> o5) {

return create(OperationAmb.amb(o1, o2, o3, o4, o5));

}

/**

* Given six Observables, mirror the one that first emits an item.

* <p>

* <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/amb.png">

*

* @param o1

* an Observable competing to react first

* @param o2

* an Observable competing to react first

* @param o3

* an Observable competing to react first

* @param o4

* an Observable competing to react first

* @param o5

* an Observable competing to react first

* @param o6

* an Observable competing to react first

* @return an Observable that emits the same sequence of items as whichever of the source

* Observables first emitted an item

* @see <a href="https://github.com/Netflix/RxJava/wiki/Conditional-and-Boolean-Operators#amb">RxJava Wiki: amb()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh229733.aspx">MSDN: Observable.Amb</a>

*/

public final static <T> Observable<T> amb(Observable<? extends T> o1, Observable<? extends T> o2, Observable<? extends T> o3, Observable<? extends T> o4, Observable<? extends T> o5, Observable<? extends T> o6) {

return create(OperationAmb.amb(o1, o2, o3, o4, o5, o6));

}

/**

* Given seven Observables, mirror the one that first emits an item.

* <p>

* <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/amb.png">

*

* @param o1

* an Observable competing to react first

* @param o2

* an Observable competing to react first

* @param o3

* an Observable competing to react first

* @param o4

* an Observable competing to react first

* @param o5

* an Observable competing to react first

* @param o6

* an Observable competing to react first

* @param o7

* an Observable competing to react first

* @return an Observable that emits the same sequence of items as whichever of the source

* Observables first emitted an item

* @see <a href="https://github.com/Netflix/RxJava/wiki/Conditional-and-Boolean-Operators#amb">RxJava Wiki: amb()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh229733.aspx">MSDN: Observable.Amb</a>

*/

public final static <T> Observable<T> amb(Observable<? extends T> o1, Observable<? extends T> o2, Observable<? extends T> o3, Observable<? extends T> o4, Observable<? extends T> o5, Observable<? extends T> o6, Observable<? extends T> o7) {

return create(OperationAmb.amb(o1, o2, o3, o4, o5, o6, o7));

}

/**

* Given eight Observables, mirror the one that first emits an item.

* <p>

* <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/amb.png">

*

* @param o1

* an Observable competing to react first

* @param o2

* an Observable competing to react first

* @param o3

* an Observable competing to react first

* @param o4

* an Observable competing to react first

* @param o5

* an Observable competing to react first

* @param o6

* an Observable competing to react first

* @param o7

* an Observable competing to react first

* @param o8

* an observable competing to react first

* @return an Observable that emits the same sequence of items as whichever of the source

* Observables first emitted an item

* @see <a href="https://github.com/Netflix/RxJava/wiki/Conditional-and-Boolean-Operators#amb">RxJava Wiki: amb()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh229733.aspx">MSDN: Observable.Amb</a>

*/

public final static <T> Observable<T> amb(Observable<? extends T> o1, Observable<? extends T> o2, Observable<? extends T> o3, Observable<? extends T> o4, Observable<? extends T> o5, Observable<? extends T> o6, Observable<? extends T> o7, Observable<? extends T> o8) {

return create(OperationAmb.amb(o1, o2, o3, o4, o5, o6, o7, o8));

}

/**

* Given nine Observables, mirror the one that first emits an item.

* <p>

* <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/amb.png">

*

* @param o1

* an Observable competing to react first

* @param o2

* an Observable competing to react first

* @param o3

* an Observable competing to react first

* @param o4

* an Observable competing to react first

* @param o5

* an Observable competing to react first

* @param o6

* an Observable competing to react first

* @param o7

* an Observable competing to react first

* @param o8

* an Observable competing to react first

* @param o9

* an Observable competing to react first

* @return an Observable that emits the same sequence of items as whichever of the source

* Observables first emitted an item

* @see <a href="https://github.com/Netflix/RxJava/wiki/Conditional-and-Boolean-Operators#amb">RxJava Wiki: amb()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/hh229733.aspx">MSDN: Observable.Amb</a>

*/

public final static <T> Observable<T> amb(Observable<? extends T> o1, Observable<? extends T> o2, Observable<? extends T> o3, Observable<? extends T> o4, Observable<? extends T> o5, Observable<? extends T> o6, Observable<? extends T> o7, Observable<? extends T> o8, Observable<? extends T> o9) {

return create(OperationAmb.amb(o1, o2, o3, o4, o5, o6, o7, o8, o9));

}

/**

* @deprecated use {@link #averageInteger}

*/

@Deprecated

public final static Observable<Integer> average(Observable<Integer> source) {

return OperationAverage.average(source);

}

/**

* Returns an Observable that emits the average of the Doubles emitted by the source Observable.

* <p>

* <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/average.png">

*

* @param source

* source Observable to compute the average of

* @return an Observable that emits a single item: the average of all the Doubles emitted

* by the source Observable

* @see <a href="https://github.com/Netflix/RxJava/wiki/Mathematical-and-Aggregate-Operators#averageinteger-averagelong-averagefloat-and-averagedouble">RxJava Wiki: averageDouble()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/system.reactive.linq.observable.average.aspx">MSDN: Observable.Average</a>

*/

public final static Observable<Double> averageDouble(Observable<Double> source) {

return OperationAverage.averageDoubles(source);

}

/**

* Returns an Observable that emits the average of the Floats emitted by the source Observable.

* <p>

* <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/average.png">

*

* @param source

* source Observable to compute the average of

* @return an Observable that emits a single item: the average of all the Floats emitted by

* the source Observable

* @see <a href="https://github.com/Netflix/RxJava/wiki/Mathematical-and-Aggregate-Operators#averageinteger-averagelong-averagefloat-and-averagedouble">RxJava Wiki: averageFloat()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/system.reactive.linq.observable.average.aspx">MSDN: Observable.Average</a>

*/

public final static Observable<Float> averageFloat(Observable<Float> source) {

return OperationAverage.averageFloats(source);

}

/**

* Returns an Observable that emits the average of the Integers emitted by the source

* Observable.

* <p>

* <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/average.png">

*

* @param source

* source Observable to compute the average of

* @return an Observable that emits a single item: the average of all the Integers emitted

* by the source Observable

* @throws IllegalArgumentException

* if the source Observable emits no items

* @see <a href="https://github.com/Netflix/RxJava/wiki/Mathematical-and-Aggregate-Operators#averageinteger-averagelong-averagefloat-and-averagedouble">RxJava Wiki: averageInteger()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/system.reactive.linq.observable.average.aspx">MSDN: Observable.Average</a>

*/

public final static Observable<Integer> averageInteger(Observable<Integer> source) {

return OperationAverage.average(source);

}

/**

* Returns an Observable that emits the average of the Longs emitted by the source Observable.

* <p>

* <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/average.png">

*

* @param source

* source Observable to compute the average of

* @return an Observable that emits a single item: the average of all the Longs emitted by

* the source Observable

* @see <a href="https://github.com/Netflix/RxJava/wiki/Mathematical-and-Aggregate-Operators#averageinteger-averagelong-averagefloat-and-averagedouble">RxJava Wiki: averageLong()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/system.reactive.linq.observable.average.aspx">MSDN: Observable.Average</a>

*/

public final static Observable<Long> averageLong(Observable<Long> source) {

return OperationAverage.averageLongs(source);

}

/**

* Combines two source Observables by emitting an item that aggregates the latest values of each

* of the source Observables each time an item is received from either of the source

* Observables, where this aggregation is defined by a specified function.

* <p>

* <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/combineLatest.png">

*

* @param o1

* the first source Observable

* @param o2

* the second source Observable

* @param combineFunction

* the aggregation function used to combine the items emitted by the source

* Observables

* @return an Observable that emits items that are the result of combining the items emitted by

* the source Observables by means of the given aggregation function

* @see <a href="https://github.com/Netflix/RxJava/wiki/Combining-Observables#combinelatest">RxJava Wiki: combineLatest()</a>

*/

public final static <T1, T2, R> Observable<R> combineLatest(Observable<? extends T1> o1, Observable<? extends T2> o2, Func2<? super T1, ? super T2, ? extends R> combineFunction) {

return create(OperationCombineLatest.combineLatest(o1, o2, combineFunction));

}

/**

* Combines three source Observables by emitting an item that aggregates the latest values of

* each of the source Observables each time an item is received from any of the source

* Observables, where this aggregation is defined by a specified function.

* <p>

* <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/combineLatest.png">

*

* @param o1

* the first source Observable

* @param o2

* the second source Observable

* @param o3

* the third source Observable

* @param combineFunction

* the aggregation function used to combine the items emitted by the source

* Observables

* @return an Observable that emits items that are the result of combining the items emitted by

* the source Observables by means of the given aggregation function

* @see <a href="https://github.com/Netflix/RxJava/wiki/Combining-Observables#combinelatest">RxJava Wiki: combineLatest()</a>

*/

public final static <T1, T2, T3, R> Observable<R> combineLatest(Observable<? extends T1> o1, Observable<? extends T2> o2, Observable<? extends T3> o3, Func3<? super T1, ? super T2, ? super T3, ? extends R> combineFunction) {

return create(OperationCombineLatest.combineLatest(o1, o2, o3, combineFunction));

}

/**

* Combines four source Observables by emitting an item that aggregates the latest values of

* each of the source Observables each time an item is received from any of the source

* Observables, where this aggregation is defined by a specified function.

* <p>

* <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/combineLatest.png">

*

* @param o1

* the first source Observable

* @param o2

* the second source Observable

* @param o3

* the third source Observable

* @param o4

* the fourth source Observable

* @param combineFunction

* the aggregation function used to combine the items emitted by the source

* Observables

* @return an Observable that emits items that are the result of combining the items emitted by

* the source Observables by means of the given aggregation function

* @see <a href="https://github.com/Netflix/RxJava/wiki/Combining-Observables#combinelatest">RxJava Wiki: combineLatest()</a>

*/

public final static <T1, T2, T3, T4, R> Observable<R> combineLatest(Observable<? extends T1> o1, Observable<? extends T2> o2, Observable<? extends T3> o3, Observable<? extends T4> o4,

Func4<? super T1, ? super T2, ? super T3, ? super T4, ? extends R> combineFunction) {

return create(OperationCombineLatest.combineLatest(o1, o2, o3, o4, combineFunction));

}

/**

* Combines five source Observables by emitting an item that aggregates the latest values of

* each of the source Observables each time an item is received from any of the source

* Observables, where this aggregation is defined by a specified function.

* <p>

* <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/combineLatest.png">

*

* @param o1

* the first source Observable

* @param o2

* the second source Observable

* @param o3

* the third source Observable

* @param o4

* the fourth source Observable

* @param o5

* the fifth source Observable

* @param combineFunction

* the aggregation function used to combine the items emitted by the source

* Observables

* @return an Observable that emits items that are the result of combining the items emitted by

* the source Observables by means of the given aggregation function

* @see <a href="https://github.com/Netflix/RxJava/wiki/Combining-Observables#combinelatest">RxJava Wiki: combineLatest()</a>

*/

public final static <T1, T2, T3, T4, T5, R> Observable<R> combineLatest(Observable<? extends T1> o1, Observable<? extends T2> o2, Observable<? extends T3> o3, Observable<? extends T4> o4, Observable<? extends T5> o5,

Func5<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? extends R> combineFunction) {

return create(OperationCombineLatest.combineLatest(o1, o2, o3, o4, o5, combineFunction));

}

/**

* Combines six source Observables by emitting an item that aggregates the latest values of each

* of the source Observables each time an item is received from any of the source Observables,

* where this aggregation is defined by a specified function.

* <p>

* <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/combineLatest.png">

*

* @param o1

* the first source Observable

* @param o2

* the second source Observable

* @param o3

* the third source Observable

* @param o4

* the fourth source Observable

* @param o5

* the fifth source Observable

* @param o6

* the sixth source Observable

* @param combineFunction

* the aggregation function used to combine the items emitted by the source

* Observables

* @return an Observable that emits items that are the result of combining the items emitted by

* the source Observables by means of the given aggregation function

* @see <a href="https://github.com/Netflix/RxJava/wiki/Combining-Observables#combinelatest">RxJava Wiki: combineLatest()</a>

*/

public final static <T1, T2, T3, T4, T5, T6, R> Observable<R> combineLatest(Observable<? extends T1> o1, Observable<? extends T2> o2, Observable<? extends T3> o3, Observable<? extends T4> o4, Observable<? extends T5> o5, Observable<? extends T6> o6,

Func6<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? extends R> combineFunction) {

return create(OperationCombineLatest.combineLatest(o1, o2, o3, o4, o5, o6, combineFunction));

}

/**

* Combines seven source Observables by emitting an item that aggregates the latest values of

* each of the source Observables each time an item is received from any of the source

* Observables, where this aggregation is defined by a specified function.

* <p>

* <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/combineLatest.png">

*

* @param o1

* the first source Observable

* @param o2

* the second source Observable

* @param o3

* the third source Observable

* @param o4

* the fourth source Observable

* @param o5

* the fifth source Observable

* @param o6

* the sixth source Observable

* @param o7

* the seventh source Observable

* @param combineFunction

* the aggregation function used to combine the items emitted by the source

* Observables

* @return an Observable that emits items that are the result of combining the items emitted by

* the source Observables by means of the given aggregation function

* @see <a href="https://github.com/Netflix/RxJava/wiki/Combining-Observables#combinelatest">RxJava Wiki: combineLatest()</a>

*/

public final static <T1, T2, T3, T4, T5, T6, T7, R> Observable<R> combineLatest(Observable<? extends T1> o1, Observable<? extends T2> o2, Observable<? extends T3> o3, Observable<? extends T4> o4, Observable<? extends T5> o5, Observable<? extends T6> o6, Observable<? extends T7> o7,

Func7<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? super T7, ? extends R> combineFunction) {

return create(OperationCombineLatest.combineLatest(o1, o2, o3, o4, o5, o6, o7, combineFunction));

}

/**

* Combines eight source Observables by emitting an item that aggregates the latest values of

* each of the source Observables each time an item is received from any of the source

* Observables, where this aggregation is defined by a specified function.

* <p>

* <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/combineLatest.png">

*

* @param o1

* the first source Observable

* @param o2

* the second source Observable

* @param o3

* the third source Observable

* @param o4

* the fourth source Observable

* @param o5

* the fifth source Observable

* @param o6

* the sixth source Observable

* @param o7

* the seventh source Observable

* @param o8

* the eighth source Observable

* @param combineFunction

* the aggregation function used to combine the items emitted by the source

* Observables

* @return an Observable that emits items that are the result of combining the items emitted by

* the source Observables by means of the given aggregation function

* @see <a href="https://github.com/Netflix/RxJava/wiki/Combining-Observables#combinelatest">RxJava Wiki: combineLatest()</a>

*/

public final static <T1, T2, T3, T4, T5, T6, T7, T8, R> Observable<R> combineLatest(Observable<? extends T1> o1, Observable<? extends T2> o2, Observable<? extends T3> o3, Observable<? extends T4> o4, Observable<? extends T5> o5, Observable<? extends T6> o6, Observable<? extends T7> o7, Observable<? extends T8> o8,

Func8<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? super T7, ? super T8, ? extends R> combineFunction) {

return create(OperationCombineLatest.combineLatest(o1, o2, o3, o4, o5, o6, o7, o8, combineFunction));

}

/**

* Combines nine source Observables by emitting an item that aggregates the latest values of

* each of the source Observables each time an item is received from any of the source

* Observables, where this aggregation is defined by a specified function.

* <p>

* <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/combineLatest.png">

*

* @param o1

* the first source Observable

* @param o2

* the second source Observable

* @param o3

* the third source Observable

* @param o4

* the fourth source Observable

* @param o5

* the fifth source Observable

* @param o6

* the sixth source Observable

* @param o7

* the seventh source Observable

* @param o8

* the eighth source Observable

* @param o9

* the ninth source Observable

* @param combineFunction

* the aggregation function used to combine the items emitted by the source

* Observables

* @return an Observable that emits items that are the result of combining the items emitted by

* the source Observables by means of the given aggregation function

* @see <a href="https://github.com/Netflix/RxJava/wiki/Combining-Observables#combinelatest">RxJava Wiki: combineLatest()</a>

*/

public final static <T1, T2, T3, T4, T5, T6, T7, T8, T9, R> Observable<R> combineLatest(Observable<? extends T1> o1, Observable<? extends T2> o2, Observable<? extends T3> o3, Observable<? extends T4> o4, Observable<? extends T5> o5, Observable<? extends T6> o6, Observable<? extends T7> o7, Observable<? extends T8> o8,

Observable<? extends T9> o9,

Func9<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? super T7, ? super T8, ? super T9, ? extends R> combineFunction) {

return create(OperationCombineLatest.combineLatest(o1, o2, o3, o4, o5, o6, o7, o8, o9, combineFunction));

}

/**

* Returns an Observable that emits the items emitted by each of the Observables emitted by an

* Observable, one after the other, without interleaving them.

* <p>

* <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/concat.png">

*

* @param observables

* an Observable that emits Observables

* @return an Observable that emits items all of the items emitted by the Observables emitted by {@code observables}, one after the other, without interleaving them

* @see <a href="https://github.com/Netflix/RxJava/wiki/Mathematical-and-Aggregate-Operators#concat">RxJava Wiki: concat()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/system.reactive.linq.observable.concat.aspx">MSDN: Observable.Concat</a>

*/

public final static <T> Observable<T> concat(Observable<? extends Observable<? extends T>> observables) {

return create(OperationConcat.concat(observables));

}

/**

* Returns an Observable that emits the items emitted by two Observables, one after the other,

* without interleaving them.

* <p>

* <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/concat.png">

*

* @param t1

* an Observable to be concatenated

* @param t2

* an Observable to be concatenated

* @return an Observable that emits items emitted by the two source Observables, one after the

* other, without interleaving them

* @see <a href="https://github.com/Netflix/RxJava/wiki/Mathematical-and-Aggregate-Operators#concat">RxJava Wiki: concat()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/system.reactive.linq.observable.concat.aspx">MSDN: Observable.Concat</a>

*/

@SuppressWarnings("unchecked")

// suppress because the types are checked by the method signature before using a vararg

public final static <T> Observable<T> concat(Observable<? extends T> t1, Observable<? extends T> t2) {

return create(OperationConcat.concat(t1, t2));

}

/**

* Returns an Observable that emits the items emitted by three Observables, one after the other,

* without interleaving them.

* <p>

* <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/concat.png">

*

* @param t1

* an Observable to be concatenated

* @param t2

* an Observable to be concatenated

* @param t3

* an Observable to be concatenated

* @return an Observable that emits items emitted by the three source Observables, one after the

* other, without interleaving them

* @see <a href="https://github.com/Netflix/RxJava/wiki/Mathematical-and-Aggregate-Operators#concat">RxJava Wiki: concat()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/system.reactive.linq.observable.concat.aspx">MSDN: Observable.Concat</a>

*/

@SuppressWarnings("unchecked")

// suppress because the types are checked by the method signature before using a vararg

public final static <T> Observable<T> concat(Observable<? extends T> t1, Observable<? extends T> t2, Observable<? extends T> t3) {

return create(OperationConcat.concat(t1, t2, t3));

}

/**

* Returns an Observable that emits the items emitted by four Observables, one after the other,

* without interleaving them.

* <p>

* <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/concat.png">

*

* @param t1

* an Observable to be concatenated

* @param t2

* an Observable to be concatenated

* @param t3

* an Observable to be concatenated

* @param t4

* an Observable to be concatenated

* @return an Observable that emits items emitted by the four source Observables, one after the

* other, without interleaving them

* @see <a href="https://github.com/Netflix/RxJava/wiki/Mathematical-and-Aggregate-Operators#concat">RxJava Wiki: concat()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/system.reactive.linq.observable.concat.aspx">MSDN: Observable.Concat</a>

*/

@SuppressWarnings("unchecked")

// suppress because the types are checked by the method signature before using a vararg

public final static <T> Observable<T> concat(Observable<? extends T> t1, Observable<? extends T> t2, Observable<? extends T> t3, Observable<? extends T> t4) {

return create(OperationConcat.concat(t1, t2, t3, t4));

}

/**

* Returns an Observable that emits the items emitted by five Observables, one after the other,

* without interleaving them.

* <p>

* <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/concat.png">

*

* @param t1

* an Observable to be concatenated

* @param t2

* an Observable to be concatenated

* @param t3

* an Observable to be concatenated

* @param t4

* an Observable to be concatenated

* @param t5

* an Observable to be concatenated

* @return an Observable that emits items emitted by the five source Observables, one after the

* other, without interleaving them

* @see <a href="https://github.com/Netflix/RxJava/wiki/Mathematical-and-Aggregate-Operators#concat">RxJava Wiki: concat()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/system.reactive.linq.observable.concat.aspx">MSDN: Observable.Concat</a>

*/

@SuppressWarnings("unchecked")

// suppress because the types are checked by the method signature before using a vararg

public final static <T> Observable<T> concat(Observable<? extends T> t1, Observable<? extends T> t2, Observable<? extends T> t3, Observable<? extends T> t4, Observable<? extends T> t5) {

return create(OperationConcat.concat(t1, t2, t3, t4, t5));

}

/**

* Returns an Observable that emits the items emitted by six Observables, one after the other,

* without interleaving them.

* <p>

* <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/concat.png">

*

* @param t1

* an Observable to be concatenated

* @param t2

* an Observable to be concatenated

* @param t3

* an Observable to be concatenated

* @param t4

* an Observable to be concatenated

* @param t5

* an Observable to be concatenated

* @param t6

* an Observable to be concatenated

* @return an Observable that emits items emitted by the six source Observables, one after the

* other, without interleaving them

* @see <a href="https://github.com/Netflix/RxJava/wiki/Mathematical-and-Aggregate-Operators#concat">RxJava Wiki: concat()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/system.reactive.linq.observable.concat.aspx">MSDN: Observable.Concat</a>

*/

@SuppressWarnings("unchecked")

// suppress because the types are checked by the method signature before using a vararg

public final static <T> Observable<T> concat(Observable<? extends T> t1, Observable<? extends T> t2, Observable<? extends T> t3, Observable<? extends T> t4, Observable<? extends T> t5, Observable<? extends T> t6) {

return create(OperationConcat.concat(t1, t2, t3, t4, t5, t6));

}

/**

* Returns an Observable that emits the items emitted by seven Observables, one after the other,

* without interleaving them.

* <p>

* <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/concat.png">

*

* @param t1

* an Observable to be concatenated

* @param t2

* an Observable to be concatenated

* @param t3

* an Observable to be concatenated

* @param t4

* an Observable to be concatenated

* @param t5

* an Observable to be concatenated

* @param t6

* an Observable to be concatenated

* @param t7

* an Observable to be concatenated

* @return an Observable that emits items emitted by the seven source Observables, one after the

* other, without interleaving them

* @see <a href="https://github.com/Netflix/RxJava/wiki/Mathematical-and-Aggregate-Operators#concat">RxJava Wiki: concat()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/system.reactive.linq.observable.concat.aspx">MSDN: Observable.Concat</a>

*/

@SuppressWarnings("unchecked")

// suppress because the types are checked by the method signature before using a vararg

public final static <T> Observable<T> concat(Observable<? extends T> t1, Observable<? extends T> t2, Observable<? extends T> t3, Observable<? extends T> t4, Observable<? extends T> t5, Observable<? extends T> t6, Observable<? extends T> t7) {

return create(OperationConcat.concat(t1, t2, t3, t4, t5, t6, t7));

}

/**

* Returns an Observable that emits the items emitted by eight Observables, one after the other,

* without interleaving them.

* <p>

* <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/concat.png">

*

* @param t1

* an Observable to be concatenated

* @param t2

* an Observable to be concatenated

* @param t3

* an Observable to be concatenated

* @param t4

* an Observable to be concatenated

* @param t5

* an Observable to be concatenated

* @param t6

* an Observable to be concatenated

* @param t7

* an Observable to be concatenated

* @param t8

* an Observable to be concatenated

* @return an Observable that emits items emitted by the eight source Observables, one after the

* other, without interleaving them

* @see <a href="https://github.com/Netflix/RxJava/wiki/Mathematical-and-Aggregate-Operators#concat">RxJava Wiki: concat()</a>

* @see <a href="http://msdn.microsoft.com/en-us/library/system.reactive.linq.observable.concat.aspx">MSDN: Observable.Concat</a>

*/

@SuppressWarnings("unchecked")

// suppress because the types are checked by the method signature before using a vararg

public final static <T> Observable<T> concat(Observable<? extends T> t1, Observable<? extends T> t2, Observable<? extends T> t3, Observable<? extends T> t4, Observable<? extends T> t5, Observable<? extends T> t6, Observable<? extends T> t7, Observable<? extends T> t8) {

return create(OperationConcat.concat(t1, t2, t3, t4, t5, t6, t7, t8));

}

/**

* Returns an Observable that emits the items emitted by nine Observables, one after the other,

* without interleaving them.

* <p>

* <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/concat.png">

*

* @param t1

* an Observable to be concatenated

* @param t2

* an Observable to be concatenated