/**

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

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

import rx.observables.ConnectableObservable;

import rx.observables.GroupedObservable;

import rx.operators.OperationOnExceptionResumeNextViaObservable;

import rx.operators.SafeObservableSubscription;

import rx.operators.SafeObserver;

import rx.operators.OperationAll;

import rx.operators.OperationBuffer;

import rx.operators.OperationCache;

import rx.operators.OperationCombineLatest;

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

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

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

import rx.util.BufferOpening;

import rx.util.OnErrorNotImplementedException;

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>

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

//TODO use a consistent parameter naming scheme (for example: for all operators that modify a source Observable, the parameter representing that source Observable should have the same name, e.g. "source" -- currently such parameters are named any of "sequence", "that", "source", "items", or "observable")

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

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

/**

* Observable with Function to execute when subscribed to.

* <p>

* NOTE: Use {@link #create(Func1)} to create an Observable instead of this method unless you

* specifically have a need for inheritance.

*

* @param onSubscribe

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

*/

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

this.onSubscribe = onSubscribe;

}

protected Observable() {

this(null);

//TODO should this be made private to prevent it? It really serves no good purpose and only confuses things. Unit tests are incorrectly using it today

}

/**

* 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<T> observer) {

// allow the hook to intercept and/or decorate

Func1<Observer<T>, Subscription> 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.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 {

SafeObservableSubscription subscription = new SafeObservableSubscription();

subscription.wrap(onSubscribeFunction.call(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<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) {

SafeObservableSubscription subscription = new SafeObservableSubscription();

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

}

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

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

if (callbacks == null) {

throw new RuntimeException("callbacks map can not be null");

}

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

if (_onNext == null) {

throw new RuntimeException("'onNext' key must contain an implementation");

}

// lookup and memoize onNext

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

handleError(e);

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

if (onError != null) {

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

} else {

throw new OnErrorNotImplementedException(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);

}

if (o == null) {

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

}

// lookup and memoize onNext

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

handleError(e);

throw new OnErrorNotImplementedException(e);

}

@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) {

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

}

});

}

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

if (onNext == null) {

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

}

if (onError == null) {

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

}

// lookup and memoize onNext

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

handleError(e);

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

}

});

}

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

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

}

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

public Subscription subscribe(final Object onNext, final Object onError, final Object 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");

}

// lookup and memoize onNext

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

Functions.from(onComplete).call();

}

@Override

public void onError(Throwable e) {

handleError(e);

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

}

});

}

public Subscription subscribe(final Action1<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<T, R> subject) {

return 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 Func1<Observer<T>, Subscription>() {

@Override

public Subscription call(Observer<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 Func1<Observer<T>, Subscription>() {

/**

* 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 call(Observer<T> observer) {

observer.onError(exception);

return Subscriptions.empty();

}

});

}

}

/**

* Creates an Observable which produces buffers of collected values. This Observable produces connected

* non-overlapping buffers. The current buffer is emitted and replaced with a new buffer when the

* Observable produced by the specified {@link Func0} produces a {@link BufferClosing} object. The

* {@link Func0} will then be used to create a new Observable to listen for the end of the next buffer.

*

* @param source

* The source {@link Observable} which produces values.

* @param bufferClosingSelector

* The {@link Func0} which is used to produce an {@link Observable} for every buffer created.

* When this {@link Observable} produces a {@link BufferClosing} object, the associated buffer

* is emitted and replaced with a new one.

* @return

* An {@link Observable} which produces connected non-overlapping buffers, which are emitted

* when the current {@link Observable} created with the {@link Func0} argument produces a

* {@link BufferClosing} object.

*/

public static <T> Observable<List<T>> buffer(Observable<T> source, Func0<Observable<BufferClosing>> bufferClosingSelector) {

return create(OperationBuffer.buffer(source, bufferClosingSelector));

}

/**

* Creates an Observable which produces buffers of collected values. This Observable produces buffers.

* Buffers are created when the specified "bufferOpenings" Observable produces a {@link BufferOpening} object.

* Additionally the {@link Func0} argument is used to create an Observable which produces {@link BufferClosing}

* objects. When this Observable produces such an object, the associated buffer is emitted.

*

* @param source

* The source {@link Observable} which produces values.

* @param bufferOpenings

* The {@link Observable} which when it produces a {@link BufferOpening} object, will cause

* another buffer to be created.

* @param bufferClosingSelector

* The {@link Func0} which is used to produce an {@link Observable} for every buffer created.

* When this {@link Observable} produces a {@link BufferClosing} object, the associated buffer

* is emitted.

* @return

* An {@link Observable} which produces buffers which are created and emitted when the specified

* {@link Observable}s publish certain objects.

*/

public static <T> Observable<List<T>> buffer(Observable<T> source, Observable<BufferOpening> bufferOpenings, Func1<BufferOpening, Observable<BufferClosing>> bufferClosingSelector) {

return create(OperationBuffer.buffer(source, bufferOpenings, bufferClosingSelector));

}

/**

* Creates an Observable which produces buffers of collected values. This Observable produces connected

* non-overlapping buffers, each containing "count" elements. When the source Observable completes or

* encounters an error, the current buffer is emitted, and the event is propagated.

*

* @param source

* The source {@link Observable} which produces values.

* @param count

* The maximum size of each buffer before it should be emitted.

* @return

* An {@link Observable} which produces connected non-overlapping buffers containing at most

* "count" produced values.

*/

public static <T> Observable<List<T>> buffer(Observable<T> source, int count) {

return create(OperationBuffer.buffer(source, count));

}

/**

* Creates an Observable which produces buffers of collected values. This Observable produces buffers every

* "skip" values, each containing "count" elements. When the source Observable completes or encounters an error,

* the current buffer is emitted and the event is propagated.

*

* @param source

* The source {@link Observable} which produces values.

* @param count

* The maximum size of each buffer before it should be emitted.

* @param skip

* How many produced values need to be skipped before starting a new buffer. Note that when "skip" and

* "count" are equals that this is the same operation as {@link Observable#buffer(Observable, int)}.

* @return

* An {@link Observable} which produces buffers every "skipped" values containing at most

* "count" produced values.

*/

public static <T> Observable<List<T>> buffer(Observable<T> source, int count, int skip) {

return create(OperationBuffer.buffer(source, count, skip));

}

/**

* Creates an Observable which produces buffers of collected values. This Observable produces connected

* non-overlapping buffers, each of a fixed duration specified by the "timespan" argument. When the source

* Observable completes or encounters an error, the current buffer is emitted and the event is propagated.

*

* @param source

* The source {@link Observable} which produces values.

* @param timespan

* The period of time each buffer is collecting values before it should be emitted, and

* replaced with a new buffer.

* @param unit

* The unit of time which applies to the "timespan" argument.

* @return

* An {@link Observable} which produces connected non-overlapping buffers with a fixed duration.

*/

public static <T> Observable<List<T>> buffer(Observable<T> source, long timespan, TimeUnit unit) {

return create(OperationBuffer.buffer(source, timespan, unit));

}

/**

* Creates an Observable which produces buffers of collected values. This Observable produces connected

* non-overlapping buffers, each of a fixed duration specified by the "timespan" argument. When the source

* Observable completes or encounters an error, the current buffer is emitted and the event is propagated.

*

* @param source

* The source {@link Observable} which produces values.

* @param timespan

* The period of time each buffer is collecting values before it should be emitted, and

* replaced with a new buffer.

* @param unit

* The unit of time which applies to the "timespan" argument.

* @param scheduler

* The {@link Scheduler} to use when determining the end and start of a buffer.

* @return

* An {@link Observable} which produces connected non-overlapping buffers with a fixed duration.

*/

public static <T> Observable<List<T>> buffer(Observable<T> source, long timespan, TimeUnit unit, Scheduler scheduler) {

return create(OperationBuffer.buffer(source, timespan, unit, scheduler));

}

/**

* Creates an Observable which produces buffers of collected values. This Observable produces connected

* non-overlapping buffers, each of a fixed duration specified by the "timespan" argument or a maximum size

* specified by the "count" argument (which ever is reached first). When the source Observable completes

* or encounters an error, the current buffer is emitted and the event is propagated.

*

* @param source

* The source {@link Observable} which produces values.

* @param timespan

* The period of time each buffer is collecting values before it should be emitted, and

* replaced with a new buffer.

* @param unit

* The unit of time which applies to the "timespan" argument.

* @param count

* The maximum size of each buffer before it should be emitted.

* @return

* An {@link Observable} which produces connected non-overlapping buffers which are emitted after

* a fixed duration or when the buffer has reached maximum capacity (which ever occurs first).

*/

public static <T> Observable<List<T>> buffer(Observable<T> source, long timespan, TimeUnit unit, int count) {

return create(OperationBuffer.buffer(source, timespan, unit, count));

}

/**

* Creates an Observable which produces buffers of collected values. This Observable produces connected

* non-overlapping buffers, each of a fixed duration specified by the "timespan" argument or a maximum size

* specified by the "count" argument (which ever is reached first). When the source Observable completes

* or encounters an error, the current buffer is emitted and the event is propagated.

*

* @param source

* The source {@link Observable} which produces values.

* @param timespan

* The period of time each buffer is collecting values before it should be emitted, and

* replaced with a new buffer.

* @param unit

* The unit of time which applies to the "timespan" argument.

* @param count

* The maximum size of each buffer before it should be emitted.

* @param scheduler

* The {@link Scheduler} to use when determining the end and start of a buffer.

* @return

* An {@link Observable} which produces connected non-overlapping buffers which are emitted after

* a fixed duration or when the buffer has reached maximum capacity (which ever occurs first).

*/

public static <T> Observable<List<T>> buffer(Observable<T> source, long timespan, TimeUnit unit, int count, Scheduler scheduler) {

return create(OperationBuffer.buffer(source, timespan, unit, count, scheduler));

}

/**

* Creates an Observable which produces buffers of collected values. This Observable starts a new buffer

* periodically, which is determined by the "timeshift" argument. Each buffer is emitted after a fixed timespan

* specified by the "timespan" argument. When the source Observable completes or encounters an error, the

* current buffer is emitted and the event is propagated.

*

* @param source

* The source {@link Observable} which produces values.

* @param timespan

* The period of time each buffer is collecting values before it should be emitted.

* @param timeshift

* The period of time after which a new buffer will be created.

* @param unit

* The unit of time which applies to the "timespan" and "timeshift" argument.

* @return

* An {@link Observable} which produces new buffers periodically, and these are emitted after

* a fixed timespan has elapsed.

*/

public static <T> Observable<List<T>> buffer(Observable<T> source, long timespan, long timeshift, TimeUnit unit) {

return create(OperationBuffer.buffer(source, timespan, timeshift, unit));

}

/**

* Creates an Observable which produces buffers of collected values. This Observable starts a new buffer

* periodically, which is determined by the "timeshift" argument. Each buffer is emitted after a fixed timespan

* specified by the "timespan" argument. When the source Observable completes or encounters an error, the

* current buffer is emitted and the event is propagated.

*

* @param source

* The source {@link Observable} which produces values.

* @param timespan

* The period of time each buffer is collecting values before it should be emitted.

* @param timeshift

* The period of time after which a new buffer will be created.

* @param unit

* The unit of time which applies to the "timespan" and "timeshift" argument.

* @param scheduler

* The {@link Scheduler} to use when determining the end and start of a buffer.

* @return

* An {@link Observable} which produces new buffers periodically, and these are emitted after

* a fixed timespan has elapsed.

*/

public static <T> Observable<List<T>> buffer(Observable<T> source, long timespan, long timeshift, TimeUnit unit, Scheduler scheduler) {

return create(OperationBuffer.buffer(source, timespan, timeshift, unit, scheduler));

}

/**

* 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(Func1<Observer<T>, Subscription> func) {

return new Observable<T>(func);

}

/**

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

* This method accepts {@link Object} to allow different languages to pass in methods using

* {@link FunctionLanguageAdaptor}.

* <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} that allows the Observer to

* cancel the subscription

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

*/

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

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

}

/**

* 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

*/

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

return new ThrowObservable<T>(exception);

}

/**

* Filters an Observable by discarding any items it emits that do not satisfy some predicate.

* <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} if they pass the filter

* @return an Observable that emits only those items emitted by the source Observable for which the

* predicate evaluates to {@code 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 items it emits that do not satisfy some predicate

* <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 an item emitted by the source Observable, and

* returns {@code true} if it passes the filter

* @return an Observable that emits only those items emitted by the source Observable for which the

* predicate function evaluates to {@code 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) {