BehaviorProcessor.java
/*
* Copyright (c) 2016-present, RxJava Contributors.
*
* 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 io.reactivex.rxjava3.processors;
import java.util.Objects;
import java.util.concurrent.atomic.*;
import java.util.concurrent.locks.*;
import org.reactivestreams.*;
import io.reactivex.rxjava3.annotations.*;
import io.reactivex.rxjava3.exceptions.MissingBackpressureException;
import io.reactivex.rxjava3.internal.subscriptions.SubscriptionHelper;
import io.reactivex.rxjava3.internal.util.*;
import io.reactivex.rxjava3.internal.util.AppendOnlyLinkedArrayList.NonThrowingPredicate;
import io.reactivex.rxjava3.plugins.RxJavaPlugins;
/**
* Processor that emits the most recent item it has observed and all subsequent observed items to each subscribed
* {@link Subscriber}.
* <p>
* <img width="640" height="460" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/S.BehaviorProcessor.png" alt="">
* <p>
* This processor does not have a public constructor by design; a new empty instance of this
* {@code BehaviorProcessor} can be created via the {@link #create()} method and
* a new non-empty instance can be created via {@link #createDefault(Object)} (named as such to avoid
* overload resolution conflict with {@code Flowable.create} that creates a Flowable, not a {@code BehaviorProcessor}).
* <p>
* In accordance with the Reactive Streams specification (<a href="https://github.com/reactive-streams/reactive-streams-jvm#2.13">Rule 2.13</a>)
* {@code null}s are not allowed as default initial values in {@link #createDefault(Object)} or as parameters to {@link #onNext(Object)} and
* {@link #onError(Throwable)}.
* <p>
* When this {@code BehaviorProcessor} is terminated via {@link #onError(Throwable)} or {@link #onComplete()}, the
* last observed item (if any) is cleared and late {@link org.reactivestreams.Subscriber}s only receive
* the respective terminal event.
* <p>
* The {@code BehaviorProcessor} does not support clearing its cached value (to appear empty again), however, the
* effect can be achieved by using a special item and making sure {@code Subscriber}s subscribe through a
* filter whose predicate filters out this special item:
* <pre><code>
* BehaviorProcessor<Integer> processor = BehaviorProcessor.create();
*
* final Integer EMPTY = Integer.MIN_VALUE;
*
* Flowable<Integer> flowable = processor.filter(v -> v != EMPTY);
*
* TestSubscriber<Integer> ts1 = flowable.test();
*
* processor.onNext(1);
* // this will "clear" the cache
* processor.onNext(EMPTY);
*
* TestSubscriber<Integer> ts2 = flowable.test();
*
* processor.onNext(2);
* processor.onComplete();
*
* // ts1 received both non-empty items
* ts1.assertResult(1, 2);
*
* // ts2 received only 2 even though the current item was EMPTY
* // when it got subscribed
* ts2.assertResult(2);
*
* // Subscribers coming after the processor was terminated receive
* // no items and only the onComplete event in this case.
* flowable.test().assertResult();
* </code></pre>
* <p>
* Even though {@code BehaviorProcessor} implements the {@code Subscriber} interface, calling
* {@code onSubscribe} is not required (<a href="https://github.com/reactive-streams/reactive-streams-jvm#2.12">Rule 2.12</a>)
* if the processor is used as a standalone source. However, calling {@code onSubscribe}
* after the {@code BehaviorProcessor} reached its terminal state will result in the
* given {@code Subscription} being cancelled immediately.
* <p>
* Calling {@link #onNext(Object)}, {@link #offer(Object)}, {@link #onError(Throwable)} and {@link #onComplete()}
* is required to be serialized (called from the same thread or called non-overlappingly from different threads
* through external means of serialization). The {@link #toSerialized()} method available to all {@code FlowableProcessor}s
* provides such serialization and also protects against reentrance (i.e., when a downstream {@code Subscriber}
* consuming this processor also wants to call {@link #onNext(Object)} on this processor recursively).
* Note that serializing over {@link #offer(Object)} is not supported through {@code toSerialized()} because it is a method
* available on the {@code PublishProcessor} and {@code BehaviorProcessor} classes only.
* <p>
* This {@code BehaviorProcessor} supports the standard state-peeking methods {@link #hasComplete()}, {@link #hasThrowable()},
* {@link #getThrowable()} and {@link #hasSubscribers()} as well as means to read the latest observed value
* in a non-blocking and thread-safe manner via {@link #hasValue()} or {@link #getValue()}.
* <p>
* Note that this processor signals {@code MissingBackpressureException} if a particular {@code Subscriber} is not
* ready to receive {@code onNext} events. To avoid this exception being signaled, use {@link #offer(Object)} to only
* try to emit an item when all {@code Subscriber}s have requested item(s).
* <dl>
* <dt><b>Backpressure:</b></dt>
* <dd>The {@code BehaviorProcessor} does not coordinate requests of its downstream {@code Subscriber}s and
* expects each individual {@code Subscriber} is ready to receive {@code onNext} items when {@link #onNext(Object)}
* is called. If a {@code Subscriber} is not ready, a {@code MissingBackpressureException} is signalled to it.
* To avoid overflowing the current {@code Subscriber}s, the conditional {@link #offer(Object)} method is available
* that returns true if any of the {@code Subscriber}s is not ready to receive {@code onNext} events. If
* there are no {@code Subscriber}s to the processor, {@code offer()} always succeeds.
* If the {@code BehaviorProcessor} is (optionally) subscribed to another {@code Publisher}, this upstream
* {@code Publisher} is consumed in an unbounded fashion (requesting {@link Long#MAX_VALUE}).</dd>
* <dt><b>Scheduler:</b></dt>
* <dd>{@code BehaviorProcessor} does not operate by default on a particular {@link io.reactivex.rxjava3.core.Scheduler} and
* the {@code Subscriber}s get notified on the thread the respective {@code onXXX} methods were invoked.</dd>
* <dt><b>Error handling:</b></dt>
* <dd>When the {@link #onError(Throwable)} is called, the {@code BehaviorProcessor} enters into a terminal state
* and emits the same {@code Throwable} instance to the last set of {@code Subscriber}s. During this emission,
* if one or more {@code Subscriber}s cancel their respective {@code Subscription}s, the
* {@code Throwable} is delivered to the global error handler via
* {@link io.reactivex.rxjava3.plugins.RxJavaPlugins#onError(Throwable)} (multiple times if multiple {@code Subscriber}s
* cancel at once).
* If there were no {@code Subscriber}s subscribed to this {@code BehaviorProcessor} when the {@code onError()}
* was called, the global error handler is not invoked.
* </dd>
* </dl>
* <p>
* Example usage:
* <pre> {@code
// subscriber will receive all events.
BehaviorProcessor<Object> processor = BehaviorProcessor.create("default");
processor.subscribe(subscriber);
processor.onNext("one");
processor.onNext("two");
processor.onNext("three");
// subscriber will receive the "one", "two" and "three" events, but not "zero"
BehaviorProcessor<Object> processor = BehaviorProcessor.create("default");
processor.onNext("zero");
processor.onNext("one");
processor.subscribe(subscriber);
processor.onNext("two");
processor.onNext("three");
// subscriber will receive only onComplete
BehaviorProcessor<Object> processor = BehaviorProcessor.create("default");
processor.onNext("zero");
processor.onNext("one");
processor.onComplete();
processor.subscribe(subscriber);
// subscriber will receive only onError
BehaviorProcessor<Object> processor = BehaviorProcessor.create("default");
processor.onNext("zero");
processor.onNext("one");
processor.onError(new RuntimeException("error"));
processor.subscribe(subscriber);
} </pre>
*
* @param <T>
* the type of item expected to be observed and emitted by the Processor
*/
public final class BehaviorProcessor<@NonNull T> extends FlowableProcessor<T> {
final AtomicReference<BehaviorSubscription<T>[]> subscribers;
static final Object[] EMPTY_ARRAY = new Object[0];
@SuppressWarnings("rawtypes")
static final BehaviorSubscription[] EMPTY = new BehaviorSubscription[0];
@SuppressWarnings("rawtypes")
static final BehaviorSubscription[] TERMINATED = new BehaviorSubscription[0];
final ReadWriteLock lock;
final Lock readLock;
final Lock writeLock;
final AtomicReference<Object> value;
final AtomicReference<Throwable> terminalEvent;
long index;
/**
* Creates a {@link BehaviorProcessor} without a default item.
*
* @param <T>
* the type of item the BehaviorProcessor will emit
* @return the constructed {@link BehaviorProcessor}
*/
@CheckReturnValue
@NonNull
public static <T> BehaviorProcessor<T> create() {
return new BehaviorProcessor<>();
}
/**
* Creates a {@link BehaviorProcessor} that emits the last item it observed and all subsequent items to each
* {@link Subscriber} that subscribes to it.
*
* @param <T>
* the type of item the BehaviorProcessor will emit
* @param defaultValue
* the item that will be emitted first to any {@link Subscriber} as long as the
* {@link BehaviorProcessor} has not yet observed any items from its source {@code Observable}
* @return the constructed {@link BehaviorProcessor}
* @throws NullPointerException if {@code defaultValue} is {@code null}
*/
@CheckReturnValue
@NonNull
public static <@NonNull T> BehaviorProcessor<T> createDefault(T defaultValue) {
Objects.requireNonNull(defaultValue, "defaultValue is null");
return new BehaviorProcessor<>(defaultValue);
}
/**
* Constructs an empty BehaviorProcessor.
* @since 2.0
*/
@SuppressWarnings("unchecked")
BehaviorProcessor() {
this.value = new AtomicReference<>();
this.lock = new ReentrantReadWriteLock();
this.readLock = lock.readLock();
this.writeLock = lock.writeLock();
this.subscribers = new AtomicReference<>(EMPTY);
this.terminalEvent = new AtomicReference<>();
}
/**
* Constructs a BehaviorProcessor with the given initial value.
* @param defaultValue the initial value, not null (verified)
* @throws NullPointerException if {@code defaultValue} is {@code null}
* @since 2.0
*/
BehaviorProcessor(T defaultValue) {
this();
this.value.lazySet(defaultValue);
}
@Override
protected void subscribeActual(@NonNull Subscriber<? super T> s) {
BehaviorSubscription<T> bs = new BehaviorSubscription<>(s, this);
s.onSubscribe(bs);
if (add(bs)) {
if (bs.cancelled) {
remove(bs);
} else {
bs.emitFirst();
}
} else {
Throwable ex = terminalEvent.get();
if (ex == ExceptionHelper.TERMINATED) {
s.onComplete();
} else {
s.onError(ex);
}
}
}
@Override
public void onSubscribe(@NonNull Subscription s) {
if (terminalEvent.get() != null) {
s.cancel();
return;
}
s.request(Long.MAX_VALUE);
}
@Override
public void onNext(@NonNull T t) {
ExceptionHelper.nullCheck(t, "onNext called with a null value.");
if (terminalEvent.get() != null) {
return;
}
Object o = NotificationLite.next(t);
setCurrent(o);
for (BehaviorSubscription<T> bs : subscribers.get()) {
bs.emitNext(o, index);
}
}
@Override
public void onError(@NonNull Throwable t) {
ExceptionHelper.nullCheck(t, "onError called with a null Throwable.");
if (!terminalEvent.compareAndSet(null, t)) {
RxJavaPlugins.onError(t);
return;
}
Object o = NotificationLite.error(t);
for (BehaviorSubscription<T> bs : terminate(o)) {
bs.emitNext(o, index);
}
}
@Override
public void onComplete() {
if (!terminalEvent.compareAndSet(null, ExceptionHelper.TERMINATED)) {
return;
}
Object o = NotificationLite.complete();
for (BehaviorSubscription<T> bs : terminate(o)) {
bs.emitNext(o, index); // relaxed read okay since this is the only mutator thread
}
}
/**
* Tries to emit the item to all currently subscribed {@link Subscriber}s if all of them
* has requested some value, returns {@code false} otherwise.
* <p>
* This method should be called in a sequential manner just like the {@code onXXX} methods
* of this {@code BehaviorProcessor}.
* <p>History: 2.0.8 - experimental
* @param t the item to emit, not {@code null}
* @return {@code true} if the item was emitted to all {@code Subscriber}s
* @throws NullPointerException if {@code t} is {@code null}
* @since 2.2
*/
@CheckReturnValue
public boolean offer(@NonNull T t) {
ExceptionHelper.nullCheck(t, "offer called with a null value.");
BehaviorSubscription<T>[] array = subscribers.get();
for (BehaviorSubscription<T> s : array) {
if (s.isFull()) {
return false;
}
}
Object o = NotificationLite.next(t);
setCurrent(o);
for (BehaviorSubscription<T> bs : array) {
bs.emitNext(o, index);
}
return true;
}
@Override
@CheckReturnValue
public boolean hasSubscribers() {
return subscribers.get().length != 0;
}
@CheckReturnValue
/* test support*/ int subscriberCount() {
return subscribers.get().length;
}
@Override
@Nullable
@CheckReturnValue
public Throwable getThrowable() {
Object o = value.get();
if (NotificationLite.isError(o)) {
return NotificationLite.getError(o);
}
return null;
}
/**
* Returns a single value the BehaviorProcessor currently has or null if no such value exists.
* <p>The method is thread-safe.
* @return a single value the BehaviorProcessor currently has or null if no such value exists
*/
@Nullable
@CheckReturnValue
public T getValue() {
Object o = value.get();
if (NotificationLite.isComplete(o) || NotificationLite.isError(o)) {
return null;
}
return NotificationLite.getValue(o);
}
@Override
@CheckReturnValue
public boolean hasComplete() {
Object o = value.get();
return NotificationLite.isComplete(o);
}
@Override
@CheckReturnValue
public boolean hasThrowable() {
Object o = value.get();
return NotificationLite.isError(o);
}
/**
* Returns true if the BehaviorProcessor has any value.
* <p>The method is thread-safe.
* @return true if the BehaviorProcessor has any value
*/
@CheckReturnValue
public boolean hasValue() {
Object o = value.get();
return o != null && !NotificationLite.isComplete(o) && !NotificationLite.isError(o);
}
boolean add(BehaviorSubscription<T> rs) {
for (;;) {
BehaviorSubscription<T>[] a = subscribers.get();
if (a == TERMINATED) {
return false;
}
int len = a.length;
@SuppressWarnings("unchecked")
BehaviorSubscription<T>[] b = new BehaviorSubscription[len + 1];
System.arraycopy(a, 0, b, 0, len);
b[len] = rs;
if (subscribers.compareAndSet(a, b)) {
return true;
}
}
}
@SuppressWarnings("unchecked")
void remove(BehaviorSubscription<T> rs) {
for (;;) {
BehaviorSubscription<T>[] a = subscribers.get();
int len = a.length;
if (len == 0) {
return;
}
int j = -1;
for (int i = 0; i < len; i++) {
if (a[i] == rs) {
j = i;
break;
}
}
if (j < 0) {
return;
}
BehaviorSubscription<T>[] b;
if (len == 1) {
b = EMPTY;
} else {
b = new BehaviorSubscription[len - 1];
System.arraycopy(a, 0, b, 0, j);
System.arraycopy(a, j + 1, b, j, len - j - 1);
}
if (subscribers.compareAndSet(a, b)) {
return;
}
}
}
@SuppressWarnings("unchecked")
BehaviorSubscription<T>[] terminate(Object terminalValue) {
setCurrent(terminalValue);
return subscribers.getAndSet(TERMINATED);
}
void setCurrent(Object o) {
Lock wl = writeLock;
wl.lock();
index++;
value.lazySet(o);
wl.unlock();
}
static final class BehaviorSubscription<@NonNull T> extends AtomicLong implements Subscription, NonThrowingPredicate<Object> {
private static final long serialVersionUID = 3293175281126227086L;
final Subscriber<? super T> downstream;
final BehaviorProcessor<T> state;
boolean next;
boolean emitting;
AppendOnlyLinkedArrayList<Object> queue;
boolean fastPath;
volatile boolean cancelled;
long index;
BehaviorSubscription(Subscriber<? super T> actual, BehaviorProcessor<T> state) {
this.downstream = actual;
this.state = state;
}
@Override
public void request(long n) {
if (SubscriptionHelper.validate(n)) {
BackpressureHelper.add(this, n);
}
}
@Override
public void cancel() {
if (!cancelled) {
cancelled = true;
state.remove(this);
}
}
void emitFirst() {
if (cancelled) {
return;
}
Object o;
synchronized (this) {
if (cancelled) {
return;
}
if (next) {
return;
}
BehaviorProcessor<T> s = state;
Lock readLock = s.readLock;
readLock.lock();
index = s.index;
o = s.value.get();
readLock.unlock();
emitting = o != null;
next = true;
}
if (o != null) {
if (test(o)) {
return;
}
emitLoop();
}
}
void emitNext(Object value, long stateIndex) {
if (cancelled) {
return;
}
if (!fastPath) {
synchronized (this) {
if (cancelled) {
return;
}
if (index == stateIndex) {
return;
}
if (emitting) {
AppendOnlyLinkedArrayList<Object> q = queue;
if (q == null) {
q = new AppendOnlyLinkedArrayList<>(4);
queue = q;
}
q.add(value);
return;
}
next = true;
}
fastPath = true;
}
test(value);
}
@Override
public boolean test(Object o) {
if (cancelled) {
return true;
}
if (NotificationLite.isComplete(o)) {
downstream.onComplete();
return true;
} else
if (NotificationLite.isError(o)) {
downstream.onError(NotificationLite.getError(o));
return true;
}
long r = get();
if (r != 0L) {
downstream.onNext(NotificationLite.<T>getValue(o));
if (r != Long.MAX_VALUE) {
decrementAndGet();
}
return false;
}
cancel();
downstream.onError(MissingBackpressureException.createDefault());
return true;
}
void emitLoop() {
for (;;) {
if (cancelled) {
return;
}
AppendOnlyLinkedArrayList<Object> q;
synchronized (this) {
q = queue;
if (q == null) {
emitting = false;
return;
}
queue = null;
}
q.forEachWhile(this);
}
}
public boolean isFull() {
return get() == 0L;
}
}
}