/**

* 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 org.junit.Assert.*;

import static org.mockito.Matchers.*;

import static org.mockito.Mockito.*;

import java.util.ArrayList;

import java.util.Arrays;

import java.util.Iterator;

import java.util.List;

import java.util.Map;

import java.util.concurrent.CountDownLatch;

import java.util.concurrent.Future;

import java.util.concurrent.TimeUnit;

import java.util.concurrent.atomic.AtomicInteger;

import java.util.concurrent.atomic.AtomicReference;

import org.junit.Before;

import org.junit.Test;

import org.mockito.Mock;

import org.mockito.Mockito;

import org.mockito.MockitoAnnotations;

import rx.observables.ConnectableObservable;

import rx.observables.GroupedObservable;

import rx.operators.OperationAll;

import rx.operators.OperationCache;

import rx.operators.OperationConcat;

import rx.operators.OperationDefer;

import rx.operators.OperationDematerialize;

import rx.operators.OperationFilter;

import rx.operators.OperationFinally;

import rx.operators.OperationGroupBy;

import rx.operators.OperationMap;

import rx.operators.OperationMaterialize;

import rx.operators.OperationMerge;

import rx.operators.OperationMergeDelayError;

import rx.operators.OperationMostRecent;

import rx.operators.OperationMulticast;

import rx.operators.OperationNext;

import rx.operators.OperationObserveOn;

import rx.operators.OperationOnErrorResumeNextViaFunction;

import rx.operators.OperationOnErrorResumeNextViaObservable;

import rx.operators.OperationOnErrorReturn;

import rx.operators.OperationSample;

import rx.operators.OperationScan;

import rx.operators.OperationSkip;

import rx.operators.OperationSubscribeOn;

import rx.operators.OperationSynchronize;

import rx.operators.OperationTake;

import rx.operators.OperationTakeLast;

import rx.operators.OperationTakeUntil;

import rx.operators.OperationTakeWhile;

import rx.operators.OperationTimestamp;

import rx.operators.OperationToFuture;

import rx.operators.OperationToIterator;

import rx.operators.OperationToObservableFuture;

import rx.operators.OperationToObservableIterable;

import rx.operators.OperationToObservableList;

import rx.operators.OperationToObservableSortedList;

import rx.operators.OperationWhere;

import rx.operators.OperationZip;

import rx.plugins.RxJavaErrorHandler;

import rx.plugins.RxJavaObservableExecutionHook;

import rx.plugins.RxJavaPlugins;

import rx.subjects.PublishSubject;

import rx.subjects.ReplaySubject;

import rx.subjects.Subject;

import rx.subscriptions.BooleanSubscription;

import rx.subscriptions.Subscriptions;

import rx.util.AtomicObservableSubscription;

import rx.util.AtomicObserver;

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

import rx.util.functions.Function;

import rx.util.functions.FunctionLanguageAdaptor;

import rx.util.functions.Functions;

/**

* The Observable interface that implements the Reactive Pattern.

* <p>

* It 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> {

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

private final Func1<Observer<T>, Subscription> onSubscribe;

protected Observable() {

this(null);

}

/**

* Construct an Observable with Function to execute when subscribed to.

* <p>

* NOTE: Generally you're better off using {@link #create(Func1)} to create an Observable instead of using inheritance.

*

* @param onSubscribe

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

*/

protected Observable(Func1<Observer<T>, Subscription> onSubscribe) {

this.onSubscribe = onSubscribe;

}

/**

* an {@link Observer} must call an Observable's <code>subscribe</code> method in order to register itself

* to receive push-based notifications from the Observable. A typical implementation of the

* <code>subscribe</code> method does the following:

* <p>

* It stores a reference to the Observer in a collection object, such as a <code>List<T></code>

* object.

* <p>

* It returns a reference to the {@link Subscription} interface. This enables

* Observers to unsubscribe (that is, to stop receiving notifications) before the Observable has

* finished sending them and has called the Observer's {@link Observer#onCompleted()} method.

* <p>

* At any given time, a particular instance of an <code>Observable<T></code> implementation is

* responsible for accepting all subscriptions and notifying all subscribers. Unless the

* documentation for a particular <code>Observable<T></code> implementation indicates otherwise,

* Observers should make no assumptions about the <code>Observable<T></code> implementation, such

* as the order of notifications that multiple Observers will receive.

* <p>

* For more information see the <a href="https://github.com/Netflix/RxJava/wiki/Observable">RxJava Wiki</a>

*

*

* @param observer

* @return a {@link Subscription} reference that allows observers

* to stop receiving notifications before the provider has finished sending them

*/

public Subscription subscribe(Observer<T> observer) {

// allow the hook to intercept and/or decorate

Func1<Observer<T>, Subscription> onSubscribeFunction = hook.onSubscribeStart(this, onSubscribe);

// validate and proceed

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.call(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 {

AtomicObservableSubscription subscription = new AtomicObservableSubscription();

subscription.wrap(onSubscribeFunction.call(new AtomicObserver<T>(subscription, observer)));

return hook.onSubscribeReturn(this, subscription);

}

} catch (Exception e) {

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

try {

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

} catch (Exception 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</code> method in order to register itself

* to receive push-based notifications from the Observable. A typical implementation of the

* <code>subscribe</code> method does the following:

* <p>

* It stores a reference to the Observer in a collection object, such as a <code>List<T></code>

* object.

* <p>

* It returns a reference to the {@link Subscription} interface. This enables

* Observers to unsubscribe (that is, to stop receiving notifications) before the Observable has

* finished sending them and has called the Observer's {@link Observer#onCompleted()} method.

* <p>

* At any given time, a particular instance of an <code>Observable<T></code> implementation is

* responsible for accepting all subscriptions and notifying all subscribers. Unless the

* documentation for a particular <code>Observable<T></code> implementation indicates otherwise,

* Observers should make no assumptions about the <code>Observable<T></code> implementation, such

* as the order of notifications that multiple Observers will receive.

* <p>

* For more information see the <a href="https://github.com/Netflix/RxJava/wiki/Observable">RxJava Wiki</a>

*

*

* @param observer

* @param scheduler

* The {@link Scheduler} that the sequence is subscribed to on.

* @return a {@link Subscription} reference that allows observers

* to stop receiving notifications before the provider has finished sending them

*/

public Subscription subscribe(Observer<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<T> o) {

AtomicObservableSubscription subscription = new AtomicObservableSubscription();

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

}

@SuppressWarnings({ "rawtypes", "unchecked" })

public Subscription subscribe(final Map<String, Object> callbacks) {

// lookup and memoize onNext

Object _onNext = callbacks.get("onNext");

if (_onNext == null) {

throw new RuntimeException("onNext must be implemented");

}

final FuncN onNext = Functions.from(_onNext);

/**

* 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() {

@Override

public void onCompleted() {

Object onComplete = callbacks.get("onCompleted");

if (onComplete != null) {

Functions.from(onComplete).call();

}

}

@Override

public void onError(Exception e) {

handleError(e);

Object onError = callbacks.get("onError");

if (onError != null) {

Functions.from(onError).call(e);

}

}

@Override

public void onNext(Object args) {

onNext.call(args);

}

});

}

public Subscription subscribe(final Map<String, Object> callbacks, Scheduler scheduler) {

return subscribeOn(scheduler).subscribe(callbacks);

}

@SuppressWarnings({ "rawtypes", "unchecked" })

public Subscription subscribe(final Object o) {

if (o instanceof Observer) {

// in case a dynamic language is not correctly handling the overloaded methods and we receive an Observer just forward to the correct method.

return subscribe((Observer) o);

}

// lookup and memoize onNext

if (o == null) {

throw new RuntimeException("onNext must be implemented");

}

final FuncN onNext = Functions.from(o);

/**

* 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() {

@Override

public void onCompleted() {

// do nothing

}

@Override

public void onError(Exception e) {

handleError(e);

// no callback defined

}

@Override

public void onNext(Object args) {

onNext.call(args);

}

});

}

public Subscription subscribe(final Object o, Scheduler scheduler) {

return subscribeOn(scheduler).subscribe(o);

}

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

/**

* 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(Exception e) {

handleError(e);

// no callback defined

}

@Override

public void onNext(T args) {

if (onNext == null) {

throw new RuntimeException("onNext must be implemented");

}

onNext.call(args);

}

});

}

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

return subscribeOn(scheduler).subscribe(onNext);

}

@SuppressWarnings({ "rawtypes", "unchecked" })

public Subscription subscribe(final Object onNext, final Object onError) {

// lookup and memoize onNext

if (onNext == null) {

throw new RuntimeException("onNext must be implemented");

}

final FuncN onNextFunction = Functions.from(onNext);

/**

* 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() {

@Override

public void onCompleted() {

// do nothing

}

@Override

public void onError(Exception e) {

handleError(e);

if (onError != null) {

Functions.from(onError).call(e);

}

}

@Override

public void onNext(Object args) {

onNextFunction.call(args);

}

});

}

public Subscription subscribe(final Object onNext, final Object onError, Scheduler scheduler) {

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

}

public Subscription subscribe(final Action1<T> onNext, final Action1<Exception> onError) {

/**

* 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(Exception e) {

handleError(e);

if (onError != null) {

onError.call(e);

}

}

@Override

public void onNext(T args) {

if (onNext == null) {

throw new RuntimeException("onNext must be implemented");

}

onNext.call(args);

}

});

}

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

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

}

@SuppressWarnings({ "rawtypes", "unchecked" })

public Subscription subscribe(final Object onNext, final Object onError, final Object onComplete) {

// lookup and memoize onNext

if (onNext == null) {

throw new RuntimeException("onNext must be implemented");

}

final FuncN onNextFunction = Functions.from(onNext);

/**

* 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() {

@Override

public void onCompleted() {

if (onComplete != null) {

Functions.from(onComplete).call();

}

}

@Override

public void onError(Exception e) {

handleError(e);

if (onError != null) {

Functions.from(onError).call(e);

}

}

@Override

public void onNext(Object args) {

onNextFunction.call(args);

}

});

}

public Subscription subscribe(final Object onNext, final Object onError, final Object onComplete, Scheduler scheduler) {

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

}

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

/**

* 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(Exception e) {

handleError(e);

if (onError != null) {

onError.call(e);

}

}

@Override

public void onNext(T args) {

if (onNext == null) {

throw new RuntimeException("onNext must be implemented");

}

onNext.call(args);

}

});

}

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

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

}

/**

* Invokes an action for each element in the observable sequence, and blocks until the sequence is terminated.

* <p>

* NOTE: This will block even if the Observable is asynchronous.

* <p>

* This is similar to {@link #subscribe(Observer)} but blocks. Because it blocks it does not need the {@link Observer#onCompleted()} or {@link Observer#onError(Exception)} methods.

*

* @param onNext

* {@link Action1}

* @throws RuntimeException

* if error occurs

*/

public void forEach(final Action1<T> onNext) {

final CountDownLatch latch = new CountDownLatch(1);

final AtomicReference<Exception> exceptionFromOnError = new AtomicReference<Exception>();

/**

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

*/

protectivelyWrapAndSubscribe(new Observer<T>() {

@Override

public void onCompleted() {

latch.countDown();

}

@Override

public void onError(Exception e) {

/*

* If we receive an onError event we set the reference on the outer thread

* so we can git it and throw after the latch.await().

*

* We do this instead of throwing directly since this may be on a different thread and the latch is still waiting.

*/

exceptionFromOnError.set(e);

latch.countDown();

}

@Override

public void onNext(T args) {

onNext.call(args);

}

});

// block until the subscription completes and then return

try {

latch.await();

} catch (InterruptedException e) {

// set the interrupted flag again so callers can still get it

// for more information see https://github.com/Netflix/RxJava/pull/147#issuecomment-13624780

Thread.currentThread().interrupt();

// using Runtime so it is not checked

throw new RuntimeException("Interrupted while waiting for subscription to complete.", e);

}

if (exceptionFromOnError.get() != null) {

if (exceptionFromOnError.get() instanceof RuntimeException) {

throw (RuntimeException) exceptionFromOnError.get();

} else {

throw new RuntimeException(exceptionFromOnError.get());

}

}

}

/**

* Invokes an action for each element in the observable sequence, and blocks until the sequence is terminated.

* <p>

* NOTE: This will block even if the Observable is asynchronous.

* <p>

* This is similar to {@link #subscribe(Observer)} but blocks. Because it blocks it does not need the {@link Observer#onCompleted()} or {@link Observer#onError(Exception)} methods.

*

* @param o

* onNext {@link Action1 action}

* @throws RuntimeException

* if error occurs

*/

@SuppressWarnings({ "rawtypes", "unchecked" })

public void forEach(final Object o) {

if (o instanceof Action1) {

// in case a dynamic language is not correctly handling the overloaded methods and we receive an Action1 just forward to the correct method.

forEach((Action1) o);

}

// lookup and memoize onNext

if (o == null) {

throw new RuntimeException("onNext must be implemented");

}

final FuncN onNext = Functions.from(o);

forEach(new Action1() {

@Override

public void call(Object args) {

onNext.call(args);

}

});

}

/**

* Returns a connectable observable sequence that upon connection causes the source sequence to push results into the specified subject.

*

* @param subject

* the subject to push source elements into.

* @param <R>

* result type

* @return a connectable observable sequence that upon connection causes the source sequence to push results into the specified subject.

*/

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

return multicast(this, subject);

}

/**

* Returns the only element of an observable sequence and throws an exception if there is not exactly one element in the observable sequence.

*

* @return The single element in the observable sequence.

*/

public T single() {

return single(this);

}

/**

* Returns the only element of an observable sequence that matches the predicate and throws an exception if there is not exactly one element in the observable sequence.

*

* @param predicate

* A predicate function to evaluate for elements in the sequence.

* @return The single element in the observable sequence.

*/

public T single(Func1<T, Boolean> predicate) {

return single(this, predicate);

}

/**

* Returns the only element of an observable sequence that matches the predicate and throws an exception if there is not exactly one element in the observable sequence.

*

* @param predicate

* A predicate function to evaluate for elements in the sequence.

* @return The single element in the observable sequence.

*/

public T single(Object predicate) {

return single(this, predicate);

}

/**

* Returns the only element of an observable sequence, or a default value if the observable sequence is empty.

*

* @param defaultValue

* default value for a sequence.

* @return The single element in the observable sequence, or a default value if no value is found.

*/

public T singleOrDefault(T defaultValue) {

return singleOrDefault(this, defaultValue);

}

/**

* Returns the only element of an observable sequence that matches the predicate, or a default value if no value is found.

*

* @param defaultValue

* default value for a sequence.

* @param predicate

* A predicate function to evaluate for elements in the sequence.

* @return The single element in the observable sequence, or a default value if no value is found.

*/

public T singleOrDefault(T defaultValue, Func1<T, Boolean> predicate) {

return singleOrDefault(this, defaultValue, predicate);

}

/**

* Returns the only element of an observable sequence that matches the predicate, or a default value if no value is found.

*

* @param defaultValue

* default value for a sequence.

* @param predicate

* A predicate function to evaluate for elements in the sequence.

* @return The single element in the observable sequence, or a default value if no value is found.

*/

public T singleOrDefault(T defaultValue, Object predicate) {

return singleOrDefault(this, defaultValue, predicate);

}

/**

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

*

* @param e

*/

private void handleError(Exception 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 Func1<Observer<T>, Subscription>() {

@Override

public Subscription call(Observer<T> t1) {

return Subscriptions.empty();

}

});

}

}

/**

* an Observable that calls {@link Observer#onError(Exception)} when the Observer subscribes.

*

* @param <T>

* the type of object returned by the Observable

*/

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

public ThrowObservable(final Exception exception) {

super(new Func1<Observer<T>, Subscription>() {

/**

* Accepts an {@link Observer} and calls its <code>onError</code> method.

*

* @param observer

* an {@link Observer} of this Observable

* @return a reference to the subscription

*/

@Override

public Subscription call(Observer<T> observer) {

observer.onError(exception);

return Subscriptions.empty();

}

});

}

}

/**

* Creates an Observable that will execute the given function when a {@link Observer} subscribes to it.

* <p>

* Write the function you pass to <code>create</code> so that it behaves as an Observable - calling the passed-in

* <code>onNext</code>, <code>onError</code>, and <code>onCompleted</code> methods appropriately.

* <p>

* A well-formed Observable must call either the {@link 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 emitted by the Observable sequence

* @param func

* a function that accepts an <code>Observer<T></code> and calls its <code>onNext</code>, <code>onError</code>, and <code>onCompleted</code> methods

* as appropriate, and returns a {@link Subscription} to allow canceling the subscription (if applicable)

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

*/

public static <T> Observable<T> create(Func1<Observer<T>, Subscription> func) {

return new Observable<T>(func);

}

/**

* Creates an Observable that will execute the given function when a {@link Observer} subscribes to it.

* <p>

* This method accept {@link Object} to allow different languages to pass in closures using {@link FunctionLanguageAdaptor}.

* <p>

* Write the function you pass to <code>create</code> so that it behaves as an Observable - calling the passed-in

* <code>onNext</code>, <code>onError</code>, and <code>onCompleted</code> methods appropriately.

* <p>

* A well-formed Observable must call either the {@link 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 emitted by the Observable sequence

* @param func

* a function that accepts an <code>Observer<T></code> and calls its <code>onNext</code>, <code>onError</code>, and <code>onCompleted</code> methods

* as appropriate, and returns a {@link Subscription} to allow canceling the subscription (if applicable)

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

*/

public static <T> Observable<T> create(final Object func) {

@SuppressWarnings("rawtypes")

final FuncN _f = Functions.from(func);

return create(new Func1<Observer<T>, Subscription>() {

@Override

public Subscription call(Observer<T> t1) {

return (Subscription) _f.call(t1);

}

});

}

/**

* Returns an Observable that returns no data to the {@link Observer} and immediately invokes its <code>onCompleted</code> method.

* <p>

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

*

* @param <T>

* the type of item emitted by the Observable

* @return an Observable that returns no data to the {@link Observer} and immediately invokes the {@link Observer}'s <code>onCompleted</code> method

*/

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

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

}

/**

* Returns an Observable that calls <code>onError</code> when an {@link Observer} subscribes to it.

* <p>

*

* @param exception

* the error to throw

* @param <T>

* the type of object returned by the Observable

* @return an Observable object that calls <code>onError</code> when an {@link Observer} subscribes

*/

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

return new ThrowObservable<T>(exception);

}

/**

* Filters an Observable by discarding any of its emissions that do not meet some test.

* <p>

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

*

* @param that

* the Observable to filter

* @param predicate

* a function that evaluates the items emitted by the source Observable, returning <code>true</code> if they pass the filter

* @return an Observable that emits only those items in the original Observable that the filter evaluates as true

*/

public static <T> Observable<T> filter(Observable<T> that, Func1<T, Boolean> predicate) {

return create(OperationFilter.filter(that, predicate));

}

/**

* Filters an Observable by discarding any of its emissions that do not meet some test.

* <p>

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

*

* @param that

* the Observable to filter

* @param function

* a function that evaluates the items emitted by the source Observable, returning <code>true</code> if they pass the filter

* @return an Observable that emits only those items in the original Observable that the filter evaluates as true

*/

public static <T> Observable<T> filter(Observable<T> that, final Object function) {

@SuppressWarnings("rawtypes")

final FuncN _f = Functions.from(function);

return filter(that, new Func1<T, Boolean>() {

@Override

public Boolean call(T t1) {

return (Boolean) _f.call(t1);

}

});

}

/**

* Filters an Observable by discarding any of its emissions that do not meet some test.

* <p>

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

*

* @param that

* the Observable to filter

* @param predicate

* a function that evaluates the items emitted by the source Observable, returning <code>true</code> if they pass the filter

* @return an Observable that emits only those items in the original Observable that the filter evaluates as true

*/

public static <T> Observable<T> where(Observable<T> that, Func1<T, Boolean> predicate) {

return create(OperationWhere.where(that, predicate));

}

/**

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

*

* @param iterable

* the source {@link Iterable} sequence

* @param <T>

* the type of items in the {@link Iterable} sequence and the type emitted by the resulting Observable

* @return an Observable that emits each item in the source {@link Iterable} sequence

* @see #toObservable(Iterable)

*/

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

return toObservable(iterable);

}

/**

* Converts an Array to an Observable sequence.

*

* @param items

* the source Array

* @param <T>

* the type of items in the Array, and the type of items emitted by the resulting Observable

* @return an Observable that emits each item in the source Array

* @see #toObservable(Object...)

*/

public static <T> Observable<T> from(T... items) {

return toObservable(items);

}

/**

* Generates an observable sequence of integral numbers within a specified range.

*

* @param start

* The value of the first integer in the sequence

* @param count

* The number of sequential integers to generate.

*

* @return An observable sequence that contains a range of sequential integral numbers.

*

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

}

/**

* Asynchronously subscribes and unsubscribes observers on the specified scheduler.

*

* @param source

* the source observable.

* @param scheduler

* the scheduler to perform subscription and unsubscription actions on.

* @param <T>

* the type of observable.

* @return the source sequence whose subscriptions and unsubscriptions happen on the specified scheduler.

*/

public static <T> Observable<T> subscribeOn(Observable<T> source, Scheduler scheduler) {

return create(OperationSubscribeOn.subscribeOn(source, scheduler));

}

/**

* Asynchronously notify observers on the specified scheduler.

*

* @param source

* the source observable.

* @param scheduler

* the scheduler to notify observers on.

* @param <T>

* the type of observable.

* @return the source sequence whose observations happen on the specified scheduler.

*/

public static <T> Observable<T> observeOn(Observable<T> source, Scheduler scheduler) {

return create(OperationObserveOn.observeOn(source, scheduler));

}

/**

* Returns an observable sequence that invokes the observable factory whenever a new observer subscribes.

* The Defer operator allows you to defer or delay the creation of the sequence until the time when an observer

* subscribes to the sequence. This is useful to allow an observer to easily obtain an updates or refreshed version

* of the sequence.

*

* @param observableFactory

* the observable factory function to invoke for each observer that subscribes to the resulting sequence.

* @param <T>

* the type of the observable.

* @return the observable sequence whose observers trigger an invocation of the given observable factory function.

*/

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

return create(OperationDefer.defer(observableFactory));

}

/**

* Returns an observable sequence that invokes the observable factory whenever a new observer subscribes.

* The Defer operator allows you to defer or delay the creation of the sequence until the time when an observer

* subscribes to the sequence. This is useful to allow an observer to easily obtain an updates or refreshed version

* of the sequence.

*

* @param observableFactory

* the observable factory function to invoke for each observer that subscribes to the resulting sequence.

* @param <T>

* the type of the observable.

* @return the observable sequence whose observers trigger an invocation of the given observable factory function.

*/