/**

* Copyright 2013 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.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.OperationCast;

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.OperationGroupBy;

import rx.operators.OperationGroupByUntil;

import rx.operators.OperationGroupJoin;

import rx.operators.OperationInterval;

import rx.operators.OperationJoin;

import rx.operators.OperationJoinPatterns;

import rx.operators.OperationMap;

import rx.operators.OperationMaterialize;

import rx.operators.OperationMerge;

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.OperationParallel;

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.OperationTake;

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.OperationTimestamp;

import rx.operators.OperationToMap;

import rx.operators.OperationToMultimap;

import rx.operators.OperationToObservableFuture;

import rx.operators.OperationToObservableIterable;

import rx.operators.OperationToObservableList;

import rx.operators.OperationToObservableSortedList;

import rx.operators.OperationUsing;

import rx.operators.OperationWindow;

import rx.operators.OperationZip;

import rx.operators.SafeObservableSubscription;

import rx.operators.SafeObserver;

import rx.plugins.RxJavaErrorHandler;

import rx.plugins.RxJavaObservableExecutionHook;

import rx.plugins.RxJavaPlugins;

import rx.schedulers.Schedulers;

import rx.subjects.AsyncSubject;

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 interface that implements the Reactive Pattern.

* <p>

* This interface provides overloaded methods for subscribing as well as

* delegate methods to the various operators.

* <p>

* The documentation for this interface 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 item emitted by the Observable

*/

public class Observable<T> {

private final static ConcurrentHashMap<Class, Boolean> internalClassMap = new ConcurrentHashMap<Class, Boolean>();

/**

* Executed when 'subscribe' is invoked.

*/

private final OnSubscribeFunc<T> onSubscribe;

/**

* Function interface for work to be performed when an {@link Observable}

* is subscribed to via {@link Observable#subscribe(Observer)}

*

* @param <T>

*/

public static interface OnSubscribeFunc<T> extends Function {

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

}

/**

* Observable with Function to execute when subscribed to.

* <p>

* NOTE: 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(OnSubscribeFunc<T> onSubscribe) {

this.onSubscribe = onSubscribe;

}

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

/**

* An {@link Observer} must call an Observable's {@code subscribe} method in

* order to receive items and notifications from the Observable.

* <p>

* A typical implementation of {@code subscribe} does the following:

* <ol>

* <li>It stores a reference to the Observer in a collection object, such as

* a {@code List<T>} object.</li>

* <li>It returns a reference to the {@link Subscription} interface. This

* enables Observers to unsubscribe, that is, to stop receiving items

* and notifications before the Observable stops sending them, which

* also invokes the Observer's {@link Observer#onCompleted onCompleted}

* method.</li>

* </ol><p>

* An <code>Observable&lt;T&gt;</code> instance is responsible for accepting

* all subscriptions and notifying all Observers. Unless the documentation

* for a particular <code>Observable&lt;T&gt;</code> implementation

* indicates otherwise, Observers should make no assumptions about the order

* in which multiple Observers will receive their notifications.

* <p>

* For more information see the

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

*

* @param observer the Observer

* @return a {@link Subscription} reference with which the {@link Observer}

* can stop receiving items before the Observable has finished

* sending them

* @throws IllegalArgumentException if the {@link Observer} provided as the

* argument to {@code subscribe()} is

* {@code null}

*/

public Subscription subscribe(Observer<? super T> observer) {

// allow the hook to intercept and/or decorate

OnSubscribeFunc<T> onSubscribeFunction = hook.onSubscribeStart(this, onSubscribe);

// validate and proceed

if (observer == null) {

throw new IllegalArgumentException("observer can not be null");

}

if (onSubscribeFunction == null) {

throw new IllegalStateException("onSubscribe function can not be null.");

// the subscribe function can also be overridden but generally that's not the appropriate approach so I won't mention that in the exception

}

try {

/**

* See https://github.com/Netflix/RxJava/issues/216 for discussion on "Guideline 6.4: Protect calls to user code from within an operator"

*/

if (isInternalImplementation(observer)) {

Subscription s = onSubscribeFunction.onSubscribe(observer);

if (s == null) {

// this generally shouldn't be the case on a 'trusted' onSubscribe but in case it happens

// we want to gracefully handle it the same as AtomicObservableSubscription does

return hook.onSubscribeReturn(this, Subscriptions.empty());

} else {

return hook.onSubscribeReturn(this, s);

}

} else {

SafeObservableSubscription subscription = new SafeObservableSubscription();

subscription.wrap(onSubscribeFunction.onSubscribe(new SafeObserver<T>(subscription, observer)));

return hook.onSubscribeReturn(this, subscription);

}

} catch (OnErrorNotImplementedException e) {

// special handling when onError is not implemented ... we just rethrow

throw e;

} catch (Throwable e) {

// if an unhandled error occurs executing the onSubscribe we will propagate it

try {

observer.onError(hook.onSubscribeError(this, e));

} catch (OnErrorNotImplementedException e2) {

// special handling when onError is not implemented ... we just rethrow

throw e2;

} catch (Throwable e2) {

// if this happens it means the onError itself failed (perhaps an invalid function implementation)

// so we are unable to propagate the error correctly and will just throw

RuntimeException r = new RuntimeException("Error occurred attempting to subscribe [" + e.getMessage() + "] and then again while trying to pass to onError.", e2);

hook.onSubscribeError(this, r);

throw r;

}

return Subscriptions.empty();

}

}

/**

* An {@link Observer} must call an Observable's {@code subscribe} method in

* order to receive items and notifications from the Observable.

* <p>

* A typical implementation of {@code subscribe} does the following:

* <ol>

* <li>It stores a reference to the Observer in a collection object, such as

* a {@code List<T>} object.</li>

* <li>It returns a reference to the {@link Subscription} interface. This

* enables Observers to unsubscribe, that is, to stop receiving items

* and notifications before the Observable stops sending them, which

* also invokes the Observer's {@link Observer#onCompleted onCompleted}

* method.</li>

* </ol><p>

* An {@code Observable<T>} instance is responsible for accepting all

* subscriptions and notifying all Observers. Unless the documentation for a

* particular {@code Observable<T>} implementation indicates otherwise,

* Observers should make no assumptions about the order in which multiple

* Observers will receive their notifications.

* <p>

* For more information see the

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

*

* @param observer the Observer

* @param scheduler the {@link Scheduler} on which Observers subscribe to

* the Observable

* @return a {@link Subscription} reference with which Observers can stop

* receiving items and notifications before the Observable has

* finished sending them

* @throws IllegalArgumentException if an argument to {@code subscribe()}

* is {@code null}

*/

public Subscription subscribe(Observer<? super T> observer, Scheduler scheduler) {

return subscribeOn(scheduler).subscribe(observer);

}

/**

* Protects against errors being thrown from Observer implementations and

* ensures onNext/onError/onCompleted contract compliance.

* <p>

* See https://github.com/Netflix/RxJava/issues/216 for a discussion on

* "Guideline 6.4: Protect calls to user code from within an operator"

*/

private Subscription protectivelyWrapAndSubscribe(Observer<? super T> o) {

SafeObservableSubscription subscription = new SafeObservableSubscription();

return subscription.wrap(subscribe(new SafeObserver<T>(subscription, o)));

}

/**

* Subscribe and ignore all events.

*

* @return

*/

public Subscription subscribe() {

return protectivelyWrapAndSubscribe(new Observer<T>() {

@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.

*

* @param onNext

* @return

* @see <a href="https://github.com/Netflix/RxJava/wiki/Observable#onnext-oncompleted-and-onerror">RxJava Wiki: onNext, onCompleted, and onError</a>

*/

public Subscription subscribe(final Action1<? super T> onNext) {

if (onNext == null) {

throw new IllegalArgumentException("onNext can not be null");

}

/**

* Wrapping since raw functions provided by the user are being invoked.

*

* See https://github.com/Netflix/RxJava/issues/216 for discussion on "Guideline 6.4: Protect calls to user code from within an operator"

*/

return protectivelyWrapAndSubscribe(new Observer<T>() {

@Override

public void onCompleted() {

// do nothing

}

@Override

public void onError(Throwable e) {

handleError(e);

throw new OnErrorNotImplementedException(e);

}

@Override

public void onNext(T args) {

onNext.call(args);

}

});

}

/**

* An {@link Observer} must call an Observable's {@code subscribe} method in

* order to receive items and notifications from the Observable.

*

* @param onNext

* @param scheduler

* @return

* @see <a href="https://github.com/Netflix/RxJava/wiki/Observable#onnext-oncompleted-and-onerror">RxJava Wiki: onNext, onCompleted, and onError</a>

*/

public Subscription subscribe(final Action1<? super T> onNext, Scheduler scheduler) {

return subscribeOn(scheduler).subscribe(onNext);

}

/**

* An {@link Observer} must call an Observable's {@code subscribe} method in

* order to receive items and notifications from the Observable.

*

* @param onNext

* @param onError

* @return

* @see <a href="https://github.com/Netflix/RxJava/wiki/Observable#onnext-oncompleted-and-onerror">RxJava Wiki: onNext, onCompleted, and onError</a>

*/

public Subscription subscribe(final Action1<? super T> onNext, final Action1<Throwable> onError) {

if (onNext == null) {

throw new IllegalArgumentException("onNext can not be null");

}

if (onError == null) {

throw new IllegalArgumentException("onError can not be null");

}

/**

* Wrapping since raw functions provided by the user are being invoked.

*

* See https://github.com/Netflix/RxJava/issues/216 for discussion on

* "Guideline 6.4: Protect calls to user code from within an operator"

*/

return protectivelyWrapAndSubscribe(new Observer<T>() {

@Override

public void onCompleted() {

// do nothing

}

@Override

public void onError(Throwable e) {

handleError(e);

onError.call(e);

}

@Override

public void onNext(T args) {

onNext.call(args);

}

});

}

/**

* An {@link Observer} must call an Observable's {@code subscribe} method in

* order to receive items and notifications from the Observable.

*

* @param onNext

* @param onError

* @param scheduler

* @return

* @see <a href="https://github.com/Netflix/RxJava/wiki/Observable#onnext-oncompleted-and-onerror">RxJava Wiki: onNext, onCompleted, and onError</a>

*/

public Subscription subscribe(final Action1<? super T> onNext, final Action1<Throwable> onError, Scheduler scheduler) {

return subscribeOn(scheduler).subscribe(onNext, onError);

}

/**

* An {@link Observer} must call an Observable's {@code subscribe} method in

* order to receive items and notifications from the Observable.

*

* @param onNext

* @param onError

* @param onComplete

* @return

* @see <a href="https://github.com/Netflix/RxJava/wiki/Observable#onnext-oncompleted-and-onerror">RxJava Wiki: onNext, onCompleted, and onError</a>

*/

public Subscription subscribe(final Action1<? super T> onNext, final Action1<Throwable> onError, final Action0 onComplete) {

if (onNext == null) {

throw new IllegalArgumentException("onNext can not be null");

}

if (onError == null) {

throw new IllegalArgumentException("onError can not be null");

}

if (onComplete == null) {

throw new IllegalArgumentException("onComplete can not be null");

}

/**

* Wrapping since raw functions provided by the user are being invoked.

*

* See https://github.com/Netflix/RxJava/issues/216 for discussion on "Guideline 6.4: Protect calls to user code from within an operator"

*/

return protectivelyWrapAndSubscribe(new Observer<T>() {

@Override

public void onCompleted() {

onComplete.call();

}

@Override

public void onError(Throwable e) {

handleError(e);

onError.call(e);

}

@Override

public void onNext(T args) {

onNext.call(args);

}

});

}

/**

* An {@link Observer} must call an Observable's {@code subscribe} method in

* order to receive items and notifications from the Observable.

*

* @param onNext

* @param onError

* @param onComplete

* @param scheduler

* @return

* @see <a href="https://github.com/Netflix/RxJava/wiki/Observable#onnext-oncompleted-and-onerror">RxJava Wiki: onNext, onCompleted, and onError</a>

*/

public Subscription subscribe(final Action1<? super T> onNext, final Action1<Throwable> onError, final Action0 onComplete, Scheduler scheduler) {

return subscribeOn(scheduler).subscribe(onNext, onError, onComplete);

}

/**

* Hides the identity of this observable.

* @return an Observable hiding the identity of this Observable.

*/

public Observable<T> asObservable() {

return create(new OperationAsObservable<T>(this));

}

/**

* Returns a {@link ConnectableObservable} that upon connection causes the

* source Observable to push results into the specified subject.

*

* @param subject the {@link Subject} for the {@link ConnectableObservable}

* to push source items into

* @param <R> result type

* @return a {@link ConnectableObservable} that upon connection causes the

* source Observable to push results into the specified

* {@link Subject}

* @see <a href="https://github.com/Netflix/RxJava/wiki/Connectable-Observable-Operators#observablepublish-and-observablemulticast">RxJava Wiki: Observable.publish() and Observable.multicast()</a>

*/

public <R> ConnectableObservable<R> multicast(Subject<? super T, ? extends R> subject) {

return OperationMulticast.multicast(this, subject);

}

/**

* Returns an observable sequence that contains the elements of a sequence

* produced by multicasting the source sequence within a selector function.

*

* @param subjectFactory the subject factory

* @param selector the selector function which can use the multicasted

* source sequence subject to the policies enforced by the

* created subject

* @return the Observable sequence that contains the elements of a sequence

* produced by multicasting the source sequence within a selector

* function

* @see <a href="https://github.com/Netflix/RxJava/wiki/Connectable-Observable-Operators#observablepublish-and-observablemulticast">RxJava: Observable.publish() and Observable.multicast()</a>

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

*/

public <TIntermediate, TResult> Observable<TResult> multicast(

final Func0<? extends Subject<? super T, ? extends TIntermediate>> subjectFactory,

final Func1<? super Observable<TIntermediate>, ? extends Observable<TResult>> selector) {

return OperationMulticast.multicast(this, subjectFactory, selector);

}

/**

* Allow the {@link RxJavaErrorHandler} to receive the exception from

* onError.

*

* @param e

*/

private void handleError(Throwable e) {

// onError should be rare so we'll only fetch when needed

RxJavaPlugins.getInstance().getErrorHandler().handleError(e);

}

/**

* An Observable that never sends any information to an {@link Observer}.

*

* This Observable is useful primarily for testing purposes.

*

* @param <T> the type of item emitted by the Observable

*/

private static class NeverObservable<T> extends Observable<T> {

public NeverObservable() {

super(new OnSubscribeFunc<T>() {

@Override

public Subscription onSubscribe(Observer<? super T> t1) {

return Subscriptions.empty();

}

});

}

}

/**

* An Observable that invokes {@link Observer#onError onError} when the

* {@link Observer} subscribes to it.

*

* @param <T> the type of item emitted by the Observable

*/

private static class ThrowObservable<T> extends Observable<T> {

public ThrowObservable(final Throwable exception) {

super(new OnSubscribeFunc<T>() {

/**

* Accepts an {@link Observer} and calls its

* {@link Observer#onError onError} method.

*

* @param observer an {@link Observer} of this Observable

* @return a reference to the subscription

*/

@Override

public Subscription onSubscribe(Observer<? super T> observer) {

observer.onError(exception);

return Subscriptions.empty();

}

});

}

}

/**

* Creates an Observable that will execute the given 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</code> 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</code> method exactly once or its <code>onError</code>

* 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} to

* allow the Observer to cancel the subscription

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

* will execute the given 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 static <T> Observable<T> create(OnSubscribeFunc<T> func) {

return new Observable<T>(func);

}

/**

* Returns an Observable that emits no items to the {@link Observer} and

* immediately invokes its {@link Observer#onCompleted onCompleted} method.

* <p>

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

*

* @param <T> the type of the items (ostensibly) emitted by the Observable

* @return an Observable that returns no data to the {@link Observer} and

* immediately invokes the {@link Observer}'s

* {@link Observer#onCompleted() onCompleted} method

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

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

*/

public static <T> Observable<T> empty() {

return from(new ArrayList<T>());

}

/**

* Returns an Observable that emits no items to the {@link Observer} and

* immediately invokes its {@link Observer#onCompleted onCompleted} method

* with the specified scheduler.

* <p>

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

*

* @param scheduler the scheduler to call the

{@link Observer#onCompleted onCompleted} method

* @param <T> the type of the items (ostensibly) emitted by the Observable

* @return an Observable that returns no data to the {@link Observer} and

* immediately invokes the {@link Observer}'s

* {@link Observer#onCompleted() onCompleted} method with the

* specified scheduler

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

* @see <a href="http://msdn.microsoft.com/en-us/library/hh229066.aspx">MSDN: Observable.Empty Method (IScheduler)</a>

*/

public static <T> Observable<T> empty(Scheduler scheduler) {

return Observable.<T> empty().subscribeOn(scheduler);

}

/**

* Returns an Observable that invokes an {@link Observer}'s

* {@link Observer#onError onError} method when the Observer subscribes to

* it.

* <p>

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

*

* @param exception the particular error to report

* @param <T> the type of the items (ostensibly) emitted by the Observable

* @return an Observable that invokes the {@link Observer}'s

* {@link Observer#onError onError} method when the Observer

* subscribes to it

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

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

*/

public static <T> Observable<T> error(Throwable exception) {

return new ThrowObservable<T>(exception);

}

/**

* Returns an Observable that invokes an {@link Observer}'s

* {@link Observer#onError onError} method with the specified scheduler.

* <p>

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

*

* @param exception the particular error to report

* @param scheduler the scheduler to call the

* {@link Observer#onError onError} method

* @param <T> the type of the items (ostensibly) emitted by the Observable

* @return an Observable that invokes the {@link Observer}'s

* {@link Observer#onError onError} method with the specified

* scheduler

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

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

*/

public static <T> Observable<T> error(Throwable exception, Scheduler scheduler) {

return Observable.<T> error(exception).subscribeOn(scheduler);

}

/**

* Converts an {@link Iterable} sequence into an Observable.

* <p>

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

* <p>

* Note: the entire iterable sequence is immediately emitted each time an

* {@link Observer} subscribes. Since this occurs before the

* {@link Subscription} is returned, it is not possible to unsubscribe from

* the sequence before it completes.

*

* @param iterable the source {@link Iterable} sequence

* @param <T> 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

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

*/

public static <T> Observable<T> from(Iterable<? extends T> iterable) {

return from(iterable, Schedulers.currentThread());

}

/**

* Converts an {@link Iterable} sequence into an Observable with the

* specified scheduler.

* <p>

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

*

* @param iterable the source {@link Iterable} sequence

* @param scheduler the scheduler to emit the items of the iterable

* @param <T> 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 <a href="https://github.com/Netflix/RxJava/wiki/Creating-Observables#from">RxJava Wiki: from()</a>

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

*/

public static <T> Observable<T> from(Iterable<? extends T> iterable, Scheduler scheduler) {

return create(OperationToObservableIterable.toObservableIterable(iterable, scheduler));

}

/**

* Converts an Array into an Observable that emits the items in the Array.

* <p>

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

* <p>

* Note: the entire array is immediately emitted each time an

* {@link Observer} subscribes. Since this occurs before the

* {@link Subscription} is returned, it is not possible to unsubscribe from

* the sequence before it completes.

*

* @param items the source array

* @param <T> 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 <a href="https://github.com/Netflix/RxJava/wiki/Creating-Observables#from">RxJava Wiki: from()</a>

*/

public static <T> Observable<T> from(T[] items) {

return from(Arrays.asList(items));

}

/**

* Converts an Array into an Observable.

* <p>

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

* <p>

* Note: the entire array is immediately emitted each time an

* {@link Observer} subscribes. Since this occurs before the

* {@link Subscription} is returned, it is not possible to unsubscribe from

* the sequence before it completes.

*

* @param items the source array

* @param scheduler the scheduler to emit the items of the array

* @param <T> 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 <a href="https://github.com/Netflix/RxJava/wiki/Creating-Observables#from">RxJava Wiki: from()</a>

*/

public static <T> Observable<T> from(T[] items, Scheduler scheduler) {

return from(Arrays.asList(items), scheduler);

}

/**

* Converts an item into an Observable that emits that item.

* <p>

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

* <p>

* Note: the item is immediately emitted each time an {@link Observer}

* subscribes. Since this occurs before the {@link Subscription} is

* returned, it is not possible to unsubscribe from the sequence before it

* completes.

*

* @param t1 the item

* @param <T> 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 <a href="https://github.com/Netflix/RxJava/wiki/Creating-Observables#from">RxJava Wiki: from()</a>

*/

@SuppressWarnings("unchecked")

// suppress unchecked because we are using varargs inside the method

public static <T> Observable<T> from(T t1) {

return from(Arrays.asList(t1));

}

/**

* Converts a series of items into an Observable that emits those items.

* <p>

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

* <p>

* Note: the items will be immediately emitted each time an {@link Observer}

* subscribes. Since this occurs before the {@link Subscription} is

* returned, it is not possible to unsubscribe from the sequence before it

* completes.

*

* @param t1 first item

* @param t2 second item

* @param <T> the type of items, and the type of items to be emitted by the

* resulting Observable

* @return an Observable that emits each item

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

*/

@SuppressWarnings("unchecked")

// suppress unchecked because we are using varargs inside the method

public static <T> Observable<T> from(T t1, T t2) {

return from(Arrays.asList(t1, t2));

}

/**

* Converts a series of items into an Observable that emits those items.

* <p>

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

* <p>

* Note: the items will be immediately emitted each time an {@link Observer}

* subscribes. Since this occurs before the {@link Subscription} is

* returned, it is not possible to unsubscribe from the sequence before it

* completes.

*

* @param t1 first item

* @param t2 second item

* @param t3 third item

* @param <T> the type of items, and the type of items to be emitted by the

* resulting Observable

* @return an Observable that emits each item

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

*/

@SuppressWarnings("unchecked")

// suppress unchecked because we are using varargs inside the method

public static <T> Observable<T> from(T t1, T t2, T t3) {

return from(Arrays.asList(t1, t2, t3));

}

/**

* Converts a series of items into an Observable that emits those items.

* <p>

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

* <p>

* Note: the items will be immediately emitted each time an {@link Observer}

* subscribes. Since this occurs before the {@link Subscription} is

* returned, it is not possible to unsubscribe from the sequence before it

* completes.

*

* @param t1 first item

* @param t2 second item

* @param t3 third item

* @param t4 fourth item

* @param <T> the type of items, and the type of items to be emitted by the

* resulting Observable

* @return an Observable that emits each item

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

*/

@SuppressWarnings("unchecked")

// suppress unchecked because we are using varargs inside the method

public static <T> Observable<T> from(T t1, T t2, T t3, T t4) {

return from(Arrays.asList(t1, t2, t3, t4));

}

/**

* Converts a series of items into an Observable that emits those items.

* <p>

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

* <p>

* Note: the items will be immediately emitted each time an {@link Observer}

* subscribes. Since this occurs before the {@link Subscription} is

* returned, it is not possible to unsubscribe from the sequence before it

* completes.

*

* @param t1 first item

* @param t2 second item

* @param t3 third item

* @param t4 fourth item

* @param t5 fifth item

* @param <T> the type of items, and the type of items to be emitted by the

* resulting Observable

* @return an Observable that emits each item

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

*/

@SuppressWarnings("unchecked")

// suppress unchecked because we are using varargs inside the method

public static <T> Observable<T> from(T t1, T t2, T t3, T t4, T t5) {

return from(Arrays.asList(t1, t2, t3, t4, t5));

}

/**

* Converts a series of items into an Observable that emits those items.

* <p>

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

* <p>

* Note: the items will be immediately emitted each time an {@link Observer}

* subscribes. Since this occurs before the {@link Subscription} is

* returned, it is not possible to unsubscribe from the sequence before it

* completes.

*

* @param t1 first item

* @param t2 second item

* @param t3 third item

* @param t4 fourth item

* @param t5 fifth item

* @param t6 sixth item

* @param <T> the type of items, and the type of items to be emitted by the

* resulting Observable

* @return an Observable that emits each item

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

*/

@SuppressWarnings("unchecked")

// suppress unchecked because we are using varargs inside the method

public static <T> Observable<T> from(T t1, T t2, T t3, T t4, T t5, T t6) {

return from(Arrays.asList(t1, t2, t3, t4, t5, t6));

}

/**

* Converts a series of items into an Observable that emits those items.

* <p>

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

* <p>

* Note: the items will be immediately emitted each time an {@link Observer}

* subscribes. Since this occurs before the {@link Subscription} is

* returned, it is not possible to unsubscribe from the sequence before it

* completes.

*

* @param t1 first item

* @param t2 second item

* @param t3 third item

* @param t4 fourth item

* @param t5 fifth item

* @param t6 sixth item

* @param t7 seventh item

* @param <T> the type of items, and the type of items to be emitted by the

* resulting Observable

* @return an Observable that emits each item

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

*/

@SuppressWarnings("unchecked")

// suppress unchecked because we are using varargs inside the method

public static <T> Observable<T> from(T t1, T t2, T t3, T t4, T t5, T t6, T t7) {

return from(Arrays.asList(t1, t2, t3, t4, t5, t6, t7));

}

/**

* Converts a series of items into an Observable that emits those items.

* <p>

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

* <p>

* Note: the items will be immediately emitted each time an {@link Observer}

* subscribes. Since this occurs before the {@link Subscription} is

* returned, it is not possible to unsubscribe from the sequence before it

* completes.

*

* @param t1 first item

* @param t2 second item