/**

* Copyright 2014 Netflix, Inc.

*

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

* you may not use this file except in compliance with the License.

* You may obtain a copy of the License at

*

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

*

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

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

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

* See the License for the specific language governing permissions and

* limitations under the License.

*/

package rx.observables;

import java.util.Arrays;

import java.util.concurrent.atomic.*;

import rx.*;

import rx.Observable.OnSubscribe;

import rx.annotations.Experimental;

import rx.exceptions.CompositeException;

import rx.functions.*;

import rx.internal.operators.BackpressureUtils;

/**

* Abstract base class for the {@link OnSubscribe} interface that helps you build Observable sources one

* {@code onNext} at a time, and automatically supports unsubscription and backpressure.

* <p>

* <h1>Usage rules</h1>

* When you implement the {@code next()} method, you

* <ul>

* <li>should either

* <ul>

* <li>create the next value and signal it via {@link SubscriptionState#onNext state.onNext()},</li>

* <li>signal a terminal condition via {@link SubscriptionState#onError state.onError()}, or

* {@link SubscriptionState#onCompleted state.onCompleted()}, or</li>

* <li>signal a stop condition via {@link SubscriptionState#stop state.stop()} indicating no further values

* will be sent.</li>

* </ul>

* </li>

* <li>may

* <ul>

* <li>call {@link SubscriptionState#onNext state.onNext()} and either

* {@link SubscriptionState#onError state.onError()} or

* {@link SubscriptionState#onCompleted state.onCompleted()} together, and

* <li>block or sleep.

* </ul>

* </li>

* <li>should not

* <ul>

* <li>do nothing or do async work and not produce any event or request stopping. If neither of

* the methods are called, an {@link IllegalStateException} is forwarded to the {@code Subscriber} and

* the Observable is terminated;</li>

* <li>call the {@code state.on}<i>foo</i>() methods more than once (yields

* {@link IllegalStateException}).</li>

* </ul>

* </li>

* </ul>

*

* The {@link SubscriptionState} object features counters that may help implement a state machine:

* <ul>

* <li>A call counter, accessible via {@link SubscriptionState#calls state.calls()} tells how many times the

* {@code next()} was run (zero based).</li>

* <li>You can use a phase counter, accessible via {@link SubscriptionState#phase state.phase}, that helps track

* the current emission phase, in a {@code switch()} statement to implement the state machine. (It is named

* {@code phase} to avoid confusion with the per-subscriber state.)</li>

* <li>You can arbitrarily change the current phase with

* {@link SubscriptionState#advancePhase state.advancePhase()},

* {@link SubscriptionState#advancePhaseBy(int) state.advancedPhaseBy(int)} and

* {@link SubscriptionState#phase(int) state.phase(int)}.</li>

* </ul>

* <p>

* When you implement {@code AbstractOnSubscribe}, you may override {@link AbstractOnSubscribe#onSubscribe} to

* perform special actions (such as registering {@code Subscription}s with {@code Subscriber.add()}) and return

* additional state for each subscriber subscribing. You can access this custom state with the

* {@link SubscriptionState#state state.state()} method. If you need to do some cleanup, you can override the

* {@link #onTerminated} method.

* <p>

* For convenience, a lambda-accepting static factory method, {@link #create}, is available.

* Another convenience is {@link #toObservable} which turns an {@code AbstractOnSubscribe}

* instance into an {@code Observable} fluently.

*

* <h1>Examples</h1>

* Note: these examples use the lambda-helper factories to avoid boilerplate.

*

* <h3>Implement: just</h3>

* <pre><code>

* AbstractOnSubscribe.create(s -> {

* s.onNext(1);

* s.onCompleted();

* }).toObservable().subscribe(System.out::println);

* </code></pre>

* <h3>Implement: from Iterable</h3>

* <pre><code>

* Iterable<T> iterable = ...;

* AbstractOnSubscribe.create(s -> {

* Iterator<T> it = s.state();

* if (it.hasNext()) {

* s.onNext(it.next());

* }

* if (!it.hasNext()) {

* s.onCompleted();

* }

* }, u -> iterable.iterator()).subscribe(System.out::println);

* </code></pre>

*

* <h3>Implement source that fails a number of times before succeeding</h3>

* <pre><code>

* AtomicInteger fails = new AtomicInteger();

* int numFails = 50;

* AbstractOnSubscribe.create(s -> {

* long c = s.calls();

* switch (s.phase()) {

* case 0:

* s.onNext("Beginning");

* s.onError(new RuntimeException("Oh, failure.");

* if (c == numFails.getAndIncrement()) {

* s.advancePhase();

* }

* break;

* case 1:

* s.onNext("Beginning");

* s.advancePhase();

* case 2:

* s.onNext("Finally working");

* s.onCompleted();

* s.advancePhase();

* default:

* throw new IllegalStateException("How did we get here?");

* }

* }).subscribe(System.out::println);

* </code></pre>

* <h3>Implement: never</h3>

* <pre><code>

* AbstractOnSubscribe.create(s -> {

* s.stop();

* }).toObservable()

* .timeout(1, TimeUnit.SECONDS)

* .subscribe(System.out::println, Throwable::printStacktrace, () -> System.out.println("Done"));

* </code></pre>

*

* @param <T> the value type

* @param <S> the per-subscriber user-defined state type

* @since (if this graduates from Experimental/Beta to supported, replace this parenthetical with the release number)

* @Experimental

*/

@Experimental

public abstract class AbstractOnSubscribe<T, S> implements OnSubscribe<T> {

/**

* Called when a Subscriber subscribes and lets the implementor create a per-subscriber custom state.

* <p>

* Override this method to have custom state per-subscriber. The default implementation returns

* {@code null}.

*

* @param subscriber the subscriber who is subscribing

* @return the custom state

*/

protected S onSubscribe(Subscriber<? super T> subscriber) {

return null;

}

/**

* Called after the terminal emission or when the downstream unsubscribes.

* <p>

* This is called only once and no {@code onNext} call will run concurrently with it. The default

* implementation does nothing.

*

* @param state the user-provided state

*/

protected void onTerminated(S state) {

}

/**

* Override this method to create an emission state-machine.

*

* @param state the per-subscriber subscription state

*/

protected abstract void next(SubscriptionState<T, S> state);

@Override

public final void call(final Subscriber<? super T> subscriber) {

final S custom = onSubscribe(subscriber);

final SubscriptionState<T, S> state = new SubscriptionState<T, S>(this, subscriber, custom);

subscriber.add(new SubscriptionCompleter<T, S>(state));

subscriber.setProducer(new SubscriptionProducer<T, S>(state));

}

/**

* Convenience method to create an Observable from this implemented instance.

*

* @return the created observable

*/

public final Observable<T> toObservable() {

return Observable.create(this);

}

/** Function that returns null. */

private static final Func1<Object, Object> NULL_FUNC1 = new Func1<Object, Object>() {

@Override

public Object call(Object t1) {

return null;

}

};

/**

* Creates an {@code AbstractOnSubscribe} instance which calls the provided {@code next} action.

* <p>

* This is a convenience method to help create {@code AbstractOnSubscribe} instances with the help of

* lambdas.

*

* @param <T> the value type

* @param <S> the per-subscriber user-defined state type

* @param next the next action to call

* @return an {@code AbstractOnSubscribe} instance

*/

public static <T, S> AbstractOnSubscribe<T, S> create(Action1<SubscriptionState<T, S>> next) {

@SuppressWarnings("unchecked")

Func1<? super Subscriber<? super T>, ? extends S> nullFunc =

(Func1<? super Subscriber<? super T>, ? extends S>)NULL_FUNC1;

return create(next, nullFunc, Actions.empty());

}

/**

* Creates an {@code AbstractOnSubscribe} instance which creates a custom state with the {@code onSubscribe}

* function and calls the provided {@code next} action.

* <p>

* This is a convenience method to help create {@code AbstractOnSubscribe} instances with the help of

* lambdas.

*

* @param <T> the value type

* @param <S> the per-subscriber user-defined state type

* @param next the next action to call

* @param onSubscribe the function that returns a per-subscriber state to be used by {@code next}

* @return an {@code AbstractOnSubscribe} instance

*/

public static <T, S> AbstractOnSubscribe<T, S> create(Action1<SubscriptionState<T, S>> next,

Func1<? super Subscriber<? super T>, ? extends S> onSubscribe) {

return create(next, onSubscribe, Actions.empty());

}

/**

* Creates an {@code AbstractOnSubscribe} instance which creates a custom state with the {@code onSubscribe}

* function, calls the provided {@code next} action and calls the {@code onTerminated} action to release the

* state when its no longer needed.

* <p>

* This is a convenience method to help create {@code AbstractOnSubscribe} instances with the help of

* lambdas.

*

* @param <T> the value type

* @param <S> the per-subscriber user-defined state type

* @param next the next action to call

* @param onSubscribe the function that returns a per-subscriber state to be used by {@code next}

* @param onTerminated the action to call to release the state created by the {@code onSubscribe} function

* @return an {@code AbstractOnSubscribe} instance

*/

public static <T, S> AbstractOnSubscribe<T, S> create(Action1<SubscriptionState<T, S>> next,

Func1<? super Subscriber<? super T>, ? extends S> onSubscribe, Action1<? super S> onTerminated) {

return new LambdaOnSubscribe<T, S>(next, onSubscribe, onTerminated);

}

/**

* An implementation that forwards the three main methods ({@code next}, {@code onSubscribe}, and

* {@code onTermianted}) to functional callbacks.

*

* @param <T> the value type

* @param <S> the per-subscriber user-defined state type

*/

private static final class LambdaOnSubscribe<T, S> extends AbstractOnSubscribe<T, S> {

final Action1<SubscriptionState<T, S>> next;

final Func1<? super Subscriber<? super T>, ? extends S> onSubscribe;

final Action1<? super S> onTerminated;

private LambdaOnSubscribe(Action1<SubscriptionState<T, S>> next,

Func1<? super Subscriber<? super T>, ? extends S> onSubscribe, Action1<? super S> onTerminated) {

this.next = next;

this.onSubscribe = onSubscribe;

this.onTerminated = onTerminated;

}

@Override

protected S onSubscribe(Subscriber<? super T> subscriber) {

return onSubscribe.call(subscriber);

}

@Override

protected void onTerminated(S state) {

onTerminated.call(state);

}

@Override

protected void next(SubscriptionState<T, S> state) {

next.call(state);

}

}

/**

* Manages unsubscription of the state.

*

* @param <T> the value type

* @param <S> the per-subscriber user-defined state type

*/

private static final class SubscriptionCompleter<T, S> extends AtomicBoolean implements Subscription {

private static final long serialVersionUID = 7993888274897325004L;

private final SubscriptionState<T, S> state;

private SubscriptionCompleter(SubscriptionState<T, S> state) {

this.state = state;

}

@Override

public boolean isUnsubscribed() {

return get();

}

@Override

public void unsubscribe() {

if (compareAndSet(false, true)) {

state.free();

}

}

}

/**

* Contains the producer loop that reacts to downstream requests of work.

*

* @param <T> the value type

* @param <S> the per-subscriber user-defined state type

*/

private static final class SubscriptionProducer<T, S> implements Producer {

final SubscriptionState<T, S> state;

private SubscriptionProducer(SubscriptionState<T, S> state) {

this.state = state;

}

@Override

public void request(long n) {

if (n > 0 && BackpressureUtils.getAndAddRequest(state.requestCount, n) == 0) {

if (n == Long.MAX_VALUE) {

// fast-path

for (; !state.subscriber.isUnsubscribed(); ) {

if (!doNext()) {

break;

}

}

} else

if (!state.subscriber.isUnsubscribed()) {

do {

if (!doNext()) {

break;

}

} while (state.requestCount.decrementAndGet() > 0 && !state.subscriber.isUnsubscribed());

}

}

}

/**

* Executes the user-overridden next() method and performs state bookkeeping and

* verification.

*

* @return true if the outer loop may continue

*/

protected boolean doNext() {

if (state.use()) {

try {

int p = state.phase();

state.parent.next(state);

if (!state.verify()) {

throw new IllegalStateException("No event produced or stop called @ Phase: " + p + " -> " + state.phase() + ", Calls: " + state.calls());

}

if (state.accept() || state.stopRequested()) {

state.terminate();

return false;

}

state.calls++;

} catch (Throwable t) {

state.terminate();

state.subscriber.onError(t);

return false;

} finally {

state.free();

}

return true;

}

return false;

}

}

/**

* Represents a per-subscription state for the {@code AbstractOnSubscribe} operation. It supports phasing

* and counts the number of times a value was requested by the downstream.

*

* @param <T> the value type

* @param <S> the per-subscriber user-defined state type

* @since (if this graduates from Experimental/Beta to supported, replace this parenthetical with the release number)

* @Experimental

*/

public static final class SubscriptionState<T, S> {

private final AbstractOnSubscribe<T, S> parent;

private final Subscriber<? super T> subscriber;

private final S state;

private final AtomicLong requestCount;

private final AtomicInteger inUse;

private int phase;

private long calls;

private T theValue;

private boolean hasOnNext;

private boolean hasCompleted;

private boolean stopRequested;

private Throwable theException;

private SubscriptionState(AbstractOnSubscribe<T, S> parent, Subscriber<? super T> subscriber, S state) {

this.parent = parent;

this.subscriber = subscriber;

this.state = state;

this.requestCount = new AtomicLong();

this.inUse = new AtomicInteger(1);

}

/**

* @return the per-subscriber specific user-defined state created via

* {@link AbstractOnSubscribe#onSubscribe}

*/

public S state() {

return state;

}

/**

* @return the current phase value

*/

public int phase() {

return phase;

}

/**

* Sets a new phase value.

*

* @param newPhase

*/

public void phase(int newPhase) {

phase = newPhase;

}

/**

* Advance the current phase by 1.

*/

public void advancePhase() {

advancePhaseBy(1);

}

/**

* Advance the current phase by the given amount (can be negative).

*

* @param amount the amount to advance the phase

*/

public void advancePhaseBy(int amount) {

phase += amount;

}

/**

* @return the number of times {@link AbstractOnSubscribe#next} was called so far, starting at 0 for the

* very first call

*/

public long calls() {

return calls;

}

/**

* Call this method to offer the next {@code onNext} value for the subscriber.

*

* @param value the value to {@code onNext}

* @throws IllegalStateException if there is a value already offered but not taken or a terminal state

* is reached

*/

public void onNext(T value) {

if (hasOnNext) {

throw new IllegalStateException("onNext not consumed yet!");

} else

if (hasCompleted) {

throw new IllegalStateException("Already terminated", theException);

}

theValue = value;

hasOnNext = true;

}

/**

* Call this method to send an {@code onError} to the subscriber and terminate all further activities.

* If there is a pending {@code onNext}, that value is emitted to the subscriber followed by this

* exception.

*

* @param e the exception to deliver to the client

* @throws IllegalStateException if the terminal state has been reached already

*/

public void onError(Throwable e) {

if (e == null) {

throw new NullPointerException("e != null required");

}

if (hasCompleted) {

throw new IllegalStateException("Already terminated", theException);

}

theException = e;

hasCompleted = true;

}

/**

* Call this method to send an {@code onCompleted} to the subscriber and terminate all further

* activities. If there is a pending {@code onNext}, that value is emitted to the subscriber followed by

* this exception.

*

* @throws IllegalStateException if the terminal state has been reached already

*/

public void onCompleted() {

if (hasCompleted) {

throw new IllegalStateException("Already terminated", theException);

}

hasCompleted = true;

}

/**

* Signals that there won't be any further events.

*/

public void stop() {

stopRequested = true;

}

/**

* Emits the {@code onNext} and/or the terminal value to the actual subscriber.

*

* @return {@code true} if the event was a terminal event

*/

protected boolean accept() {

if (hasOnNext) {

T value = theValue;

theValue = null;

hasOnNext = false;

try {

subscriber.onNext(value);

} catch (Throwable t) {

hasCompleted = true;

Throwable e = theException;

theException = null;

if (e == null) {

subscriber.onError(t);

} else {

subscriber.onError(new CompositeException(Arrays.asList(t, e)));

}

return true;

}

}

if (hasCompleted) {

Throwable e = theException;

theException = null;

if (e != null) {

subscriber.onError(e);

} else {

subscriber.onCompleted();

}

return true;

}

return false;

}

/**

* Verify if the {@code next()} generated an event or requested a stop.

*

* @return true if either event was generated or stop was requested

*/

protected boolean verify() {

return hasOnNext || hasCompleted || stopRequested;

}

/** @return true if the {@code next()} requested a stop */

protected boolean stopRequested() {

return stopRequested;

}

/**

* Request the state to be used by {@code onNext} or returns {@code false} if the downstream has

* unsubscribed.

*

* @return {@code true} if the state can be used exclusively

* @throws IllegalStateEception

* @warn "throws" section incomplete

*/

protected boolean use() {

int i = inUse.get();

if (i == 0) {

return false;

} else

if (i == 1 && inUse.compareAndSet(1, 2)) {

return true;

}

throw new IllegalStateException("This is not reentrant nor threadsafe!");

}

/**

* Release the state if there are no more interest in it and it is not in use.

*/

protected void free() {

int i = inUse.get();

if (i <= 0) {

return;

} else

if (inUse.decrementAndGet() == 0) {

parent.onTerminated(state);

}

}

/**

* Terminates the state immediately and calls {@link AbstractOnSubscribe#onTerminated} with the custom

* state.

*/

protected void terminate() {

for (;;) {

int i = inUse.get();

if (i <= 0) {

return;

}

if (inUse.compareAndSet(i, 0)) {

parent.onTerminated(state);

break;

}

}

}

}

}