/**

* 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.Comparator;

import java.util.List;

import java.util.concurrent.Future;

import java.util.concurrent.TimeUnit;

import rx.concurrency.Schedulers;

import rx.observables.BlockingObservable;

import rx.observables.ConnectableObservable;

import rx.observables.GroupedObservable;

import rx.operators.OperationAll;

import rx.operators.OperationAny;

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

import rx.operators.OperationDistinct;

import rx.operators.OperationDistinctUntilChanged;

import rx.operators.OperationElementAt;

import rx.operators.OperationFilter;

import rx.operators.OperationFinally;

import rx.operators.OperationFirstOrDefault;

import rx.operators.OperationGroupBy;

import rx.operators.OperationInterval;

import rx.operators.OperationMap;

import rx.operators.OperationMaterialize;

import rx.operators.OperationMerge;

import rx.operators.OperationMergeDelayError;

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

import rx.operators.OperationSample;

import rx.operators.OperationScan;

import rx.operators.OperationSkip;

import rx.operators.OperationSkipLast;

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

import rx.operators.OperationToObservableFuture;

import rx.operators.OperationToObservableIterable;

import rx.operators.OperationToObservableList;

import rx.operators.OperationToObservableSortedList;

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.subjects.AsyncSubject;

import rx.subjects.PublishSubject;

import rx.subjects.ReplaySubject;

import rx.subjects.Subject;

import rx.subscriptions.Subscriptions;

import rx.util.Closing;

import rx.util.OnErrorNotImplementedException;

import rx.util.Opening;

import rx.util.Range;

import rx.util.Timestamped;

import rx.util.functions.Action0;

import rx.util.functions.Action1;

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;

/**

* 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://github.com/Netflix/RxJava/wiki/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>

*/

public class Observable<T> {

/**

* 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:

* <p>

* It stores a reference to the Observer in a collection object, such as a {@code List<T>} object.

* <p>

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

* <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:

* <p>

* It stores a reference to the Observer in a collection object, such as a {@code List<T>} object.

* <p>

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

* <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);

}

/**

* Used for protecting against errors being thrown from Observer implementations and ensuring onNext/onError/onCompleted contract compliance.

* <p>

* See https://github.com/Netflix/RxJava/issues/216 for 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)));

}

/**

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

* receive items and notifications from the Observable.

*

* @param onNext

*/

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

*/

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

return subscribeOn(scheduler).subscribe(onNext);

}

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

*/

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

*/

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

*/

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

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

}

/**

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

*/

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

return OperationMulticast.multicast(this, subject);

}

/**

* 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://github.com/Netflix/RxJava/wiki/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

* canceling the subscription

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

* function

*/

public static <T> Observable<T> create(OnSubscribeFunc<T> func) {

return new Observable<T>(func);

}

/**

* Returns an Observable that emits no data 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="http://msdn.microsoft.com/en-us/library/hh229670(v=vs.103).aspx">MSDN: Observable.Empty Method</a>

*/

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

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

}

/**

* Returns an Observable that emits no data 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.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="http://msdn.microsoft.com/en-us/library/hh229066(v=vs.103).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="http://msdn.microsoft.com/en-us/library/hh244299(v=vs.103).aspx">MSDN: Observable.Throw Method</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.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="http://msdn.microsoft.com/en-us/library/hh211711(v=vs.103).aspx">MSDN: Observable.Throw Method</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://github.com/Netflix/RxJava/wiki/images/rx-operators/from.png">

*

* <p>Implementation note: the entire iterable sequence will be immediately emitted each time an {@link Observer} subscribes. Since this occurs before the {@link Subscription} is returned,

* it in 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

*/

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

return create(OperationToObservableIterable.toObservableIterable(iterable));

}

/**

* Converts an Array into an Observable.

* <p>

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

*

* <p>Implementation note: the entire iterable sequence will be immediately emitted each time an {@link Observer} subscribes. Since this occurs before the {@link Subscription} is returned,

* it in not possible to unsubscribe from the sequence before it completes.

*

* @param items

* the source 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

*/

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

return create(OperationToObservableIterable.toObservableIterable(Arrays.asList(items)));

}

/**

* Converts a series of items into an Observable.

* <p>

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

*

* <p>Implementation note: the entire array will be immediately emitted each time an {@link Observer} subscribes. Since this occurs before the {@link Subscription} is returned,

* it in not possible to unsubscribe from the sequence before it completes.

*

* @param t1

* item

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

*/

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

* <p>

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

*

* <p>Implementation note: the entire array will be immediately emitted each time an {@link Observer} subscribes. Since this occurs before the {@link Subscription} is returned,

* it in not possible to unsubscribe from the sequence before it completes.

*

* @param t1

* item

* @param t2

* item

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

*/

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

* <p>

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

*

* <p>Implementation note: the entire array will be immediately emitted each time an {@link Observer} subscribes. Since this occurs before the {@link Subscription} is returned,

* it in not possible to unsubscribe from the sequence before it completes.

*

* @param t1

* item

* @param t2

* item

* @param t3

* item

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

*/

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

* <p>

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

*

* <p>Implementation note: the entire array will be immediately emitted each time an {@link Observer} subscribes. Since this occurs before the {@link Subscription} is returned,

* it in not possible to unsubscribe from the sequence before it completes.

*

* @param t1

* item

* @param t2

* item

* @param t3

* item

* @param t4

* item

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

*/

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

* <p>

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

*

* <p>Implementation note: the entire array will be immediately emitted each time an {@link Observer} subscribes. Since this occurs before the {@link Subscription} is returned,

* it in not possible to unsubscribe from the sequence before it completes.

*

* @param t1

* item

* @param t2

* item

* @param t3

* item

* @param t4

* item

* @param t5

* item

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

*/

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

* <p>

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

*

* <p>Implementation note: the entire array will be immediately emitted each time an {@link Observer} subscribes. Since this occurs before the {@link Subscription} is returned,

* it in not possible to unsubscribe from the sequence before it completes.

*

* @param t1

* item

* @param t2

* item

* @param t3

* item

* @param t4

* item

* @param t5

* item

* @param t6

* item

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

*/

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

* <p>

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

*

* <p>Implementation note: the entire array will be immediately emitted each time an {@link Observer} subscribes. Since this occurs before the {@link Subscription} is returned,

* it in not possible to unsubscribe from the sequence before it completes.

*

* @param t1

* item

* @param t2

* item

* @param t3

* item

* @param t4

* item

* @param t5

* item

* @param t6

* item

* @param t7

* item

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

*/

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

* <p>

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

*

* <p>Implementation note: the entire array will be immediately emitted each time an {@link Observer} subscribes. Since this occurs before the {@link Subscription} is returned,

* it in not possible to unsubscribe from the sequence before it completes.

*

* @param t1

* item

* @param t2

* item

* @param t3

* item

* @param t4

* item

* @param t5

* item

* @param t6

* item

* @param t7

* item

* @param t8

* item

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

*/

@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, T t8) {

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

}

/**

* Converts a series of items into an Observable.

* <p>

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

*

* <p>Implementation note: the entire array will be immediately emitted each time an {@link Observer} subscribes. Since this occurs before the {@link Subscription} is returned,

* it in not possible to unsubscribe from the sequence before it completes.

*

* @param t1

* item

* @param t2

* item

* @param t3

* item

* @param t4

* item

* @param t5

* item

* @param t6

* item

* @param t7

* item

* @param t8

* item

* @param t9

* item

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

*/

@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, T t8, T t9) {

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

}

/**

* Converts a series of items into an Observable.

* <p>

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

*

* <p>Implementation note: the entire array will be immediately emitted each time an {@link Observer} subscribes. Since this occurs before the {@link Subscription} is returned,

* it in not possible to unsubscribe from the sequence before it completes.

*

* @param t1

* item

* @param t2

* item

* @param t3

* item

* @param t4

* item

* @param t5

* item

* @param t6

* item

* @param t7

* item

* @param t8

* item

* @param t10

* item

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

*/

@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, T t8, T t9, T t10) {

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

}

/**

* Generates an Observable that emits a sequence of integers within a specified range.

* <p>

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

*

* <p>Implementation note: the entire range will be immediately emitted each time an {@link Observer} subscribes. Since this occurs before the {@link Subscription} is returned,

* it in not possible to unsubscribe from the sequence before it completes.

*

* @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 <a href="http://msdn.microsoft.com/en-us/library/hh229460(v=vs.103).aspx">Observable.Range Method (Int32, Int32)</a>

*/

public static Observable<Integer> range(int start, int count) {

return from(Range.createWithCount(start, count));

}

/**

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

* <p>

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

* <p>

* The defer operator allows you to defer or delay emitting items from an Observable until such

* time as an Observer subscribes to the Observable. This allows an {@link Observer} to easily

* obtain updates or a refreshed version of the sequence.

*

* @param observableFactory

* the Observable factory function to invoke for each {@link Observer} that

* subscribes to the resulting Observable

* @param <T>

* 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

*/

public static <T> Observable<T> defer(Func0<? extends Observable<? extends T>> observableFactory) {

return create(OperationDefer.defer(observableFactory));

}

/**

* Returns an Observable that emits a single item and then completes.

* <p>

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

* <p>

* To convert any object into an Observable that emits that object, pass that object into the

* <code>just</code> method.

* <p>

* This is similar to the {@link #from(java.lang.Object[])} method, except that

* <code>from()</code> will convert an {@link Iterable} object into an Observable that emits

* each of the items in the Iterable, one at a time, while the <code>just()</code> method

* converts an Iterable into an Observable that emits the entire Iterable as a single item.

*

* @param value

* the item to pass to the {@link Observer}'s {@link Observer#onNext onNext} method

* @param <T>