src/main/java/rx/internal/operators/OperatorReplay.java
/**
* 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.internal.operators;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.*;
import rx.*;
import rx.Observable;
import rx.exceptions.*;
import rx.functions.*;
import rx.internal.util.OpenHashSet;
import rx.observables.ConnectableObservable;
import rx.schedulers.Timestamped;
import rx.subscriptions.Subscriptions;
public final class OperatorReplay<T> extends ConnectableObservable<T> implements Subscription {
/** The source observable. */
final Observable<? extends T> source;
/** Holds the current subscriber that is, will be or just was subscribed to the source observable. */
final AtomicReference<ReplaySubscriber<T>> current;
/** A factory that creates the appropriate buffer for the ReplaySubscriber. */
final Func0<? extends ReplayBuffer<T>> bufferFactory;
@SuppressWarnings("rawtypes")
static final Func0 DEFAULT_UNBOUNDED_FACTORY = new Func0() {
@Override
public Object call() {
return new UnboundedReplayBuffer<Object>(16);
}
};
/**
* Given a connectable observable factory, it multicasts over the generated
* ConnectableObservable via a selector function.
* @param <T> the upstream's value type
* @param <U> the intermediate value type of the ConnectableObservable
* @param <R> the final value type provided by the selector function
* @param connectableFactory the factory that returns a ConnectableObservable instance
* @param selector the function applied on the ConnectableObservable and returns the Observable
* the downstream will subscribe to.
* @return the Observable multicasting over a transformation of a ConnectableObservable
*/
public static <T, U, R> Observable<R> multicastSelector(
final Func0<? extends ConnectableObservable<U>> connectableFactory,
final Func1<? super Observable<U>, ? extends Observable<R>> selector) {
return Observable.unsafeCreate(new OnSubscribe<R>() {
@Override
public void call(final Subscriber<? super R> child) {
ConnectableObservable<U> co;
Observable<R> observable;
try {
co = connectableFactory.call();
observable = selector.call(co);
} catch (Throwable e) {
Exceptions.throwOrReport(e, child);
return;
}
observable.subscribe(child);
co.connect(new Action1<Subscription>() {
@Override
public void call(Subscription t) {
child.add(t);
}
});
}
});
}
/**
* Child Subscribers will observe the events of the ConnectableObservable on the
* specified scheduler.
* @param <T> the value type
* @param co the ConnectableObservable to schedule on the specified scheduler
* @param scheduler the target Scheduler instance
* @return the ConnectableObservable instance that is observed on the specified scheduler
*/
public static <T> ConnectableObservable<T> observeOn(final ConnectableObservable<T> co, final Scheduler scheduler) {
final Observable<T> observable = co.observeOn(scheduler);
OnSubscribe<T> onSubscribe = new OnSubscribe<T>() {
@Override
public void call(final Subscriber<? super T> child) {
// apply observeOn and prevent calling onStart() again
observable.unsafeSubscribe(new Subscriber<T>(child) {
@Override
public void onNext(T t) {
child.onNext(t);
}
@Override
public void onError(Throwable e) {
child.onError(e);
}
@Override
public void onCompleted() {
child.onCompleted();
}
});
}
};
return new ConnectableObservable<T>(onSubscribe) {
@Override
public void connect(Action1<? super Subscription> connection) {
co.connect(connection);
}
};
}
/**
* Creates a replaying ConnectableObservable with an unbounded buffer.
* @param <T> the value type
* @param source the source Observable
* @return the replaying ConnectableObservable
*/
@SuppressWarnings("unchecked")
public static <T> ConnectableObservable<T> create(Observable<? extends T> source) {
return create(source, DEFAULT_UNBOUNDED_FACTORY);
}
/**
* Creates a replaying ConnectableObservable with a size bound buffer.
* @param <T> the value type
* @param source the source Observable
* @param bufferSize the maximum number of elements buffered
* @return the replaying ConnectableObservable
*/
@SuppressWarnings("cast")
public static <T> ConnectableObservable<T> create(Observable<? extends T> source,
final int bufferSize) {
if (bufferSize == Integer.MAX_VALUE) {
return (ConnectableObservable<T>)create(source);
}
return create(source, new Func0<ReplayBuffer<T>>() {
@Override
public ReplayBuffer<T> call() {
return new SizeBoundReplayBuffer<T>(bufferSize);
}
});
}
/**
* Creates a replaying ConnectableObservable with a time bound buffer.
* @param <T> the value type
* @param source the source Observable
* @param maxAge the maximum age (exclusive) of each item when timestamped with the given scheduler
* @param unit the time unit of the maximum age
* @param scheduler the scheduler providing the notion of current time
* @return the replaying ConnectableObservable
*/
@SuppressWarnings("cast")
public static <T> ConnectableObservable<T> create(Observable<? extends T> source,
long maxAge, TimeUnit unit, Scheduler scheduler) {
return (ConnectableObservable<T>)create(source, maxAge, unit, scheduler, Integer.MAX_VALUE);
}
/**
* Creates a replaying ConnectableObservable with a size and time bound buffer.
* @param <T> the value type
* @param source the source Observable
* @param maxAge the maximum age (exclusive) of each item when timestamped with the given scheduler
* @param unit the time unit of the maximum age
* @param scheduler the scheduler providing the notion of current time
* @param bufferSize the maximum number of elements buffered
* @return the replaying ConnectableObservable
*/
public static <T> ConnectableObservable<T> create(Observable<? extends T> source,
long maxAge, TimeUnit unit, final Scheduler scheduler, final int bufferSize) {
final long maxAgeInMillis = unit.toMillis(maxAge);
return create(source, new Func0<ReplayBuffer<T>>() {
@Override
public ReplayBuffer<T> call() {
return new SizeAndTimeBoundReplayBuffer<T>(bufferSize, maxAgeInMillis, scheduler);
}
});
}
/**
* Creates a OperatorReplay instance to replay values of the given source observable.
* @param source the source observable
* @param bufferFactory the factory to instantiate the appropriate buffer when the observable becomes active
* @return the connectable observable
*/
static <T> ConnectableObservable<T> create(Observable<? extends T> source,
final Func0<? extends ReplayBuffer<T>> bufferFactory) {
// the current connection to source needs to be shared between the operator and its onSubscribe call
final AtomicReference<ReplaySubscriber<T>> curr = new AtomicReference<ReplaySubscriber<T>>();
OnSubscribe<T> onSubscribe = new OnSubscribe<T>() {
@Override
public void call(Subscriber<? super T> child) {
// concurrent connection/disconnection may change the state,
// we loop to be atomic while the child subscribes
for (;;) {
// get the current subscriber-to-source
ReplaySubscriber<T> r = curr.get();
// if there isn't one
if (r == null) {
// create a new subscriber to source
ReplaySubscriber<T> u = new ReplaySubscriber<T>(bufferFactory.call());
// perform extra initialization to avoid 'this' to escape during construction
u.init();
// let's try setting it as the current subscriber-to-source
if (!curr.compareAndSet(r, u)) {
// didn't work, maybe someone else did it or the current subscriber
// to source has just finished
continue;
}
// we won, let's use it going onwards
r = u;
}
// create the backpressure-managing producer for this child
InnerProducer<T> inner = new InnerProducer<T>(r, child);
// we try to add it to the array of producers
// if it fails, no worries because we will still have its buffer
// so it is going to replay it for us
r.add(inner);
// the producer has been registered with the current subscriber-to-source so
// at least it will receive the next terminal event
child.add(inner);
// pin the head of the buffer here, shouldn't affect anything else
r.buffer.replay(inner);
// setting the producer will trigger the first request to be considered by
// the subscriber-to-source.
child.setProducer(inner);
break; // NOPMD
}
}
};
return new OperatorReplay<T>(onSubscribe, source, curr, bufferFactory);
}
private OperatorReplay(OnSubscribe<T> onSubscribe, Observable<? extends T> source,
final AtomicReference<ReplaySubscriber<T>> current,
final Func0<? extends ReplayBuffer<T>> bufferFactory) {
super(onSubscribe);
this.source = source;
this.current = current;
this.bufferFactory = bufferFactory;
}
@Override
public void unsubscribe() {
current.lazySet(null);
}
@Override
public boolean isUnsubscribed() {
ReplaySubscriber<T> ps = current.get();
return ps == null || ps.isUnsubscribed();
}
@Override
public void connect(Action1<? super Subscription> connection) {
boolean doConnect;
ReplaySubscriber<T> ps;
// we loop because concurrent connect/disconnect and termination may change the state
for (;;) {
// retrieve the current subscriber-to-source instance
ps = current.get();
// if there is none yet or the current has unsubscribed
if (ps == null || ps.isUnsubscribed()) {
// create a new subscriber-to-source
ReplaySubscriber<T> u = new ReplaySubscriber<T>(bufferFactory.call());
// initialize out the constructor to avoid 'this' to escape
u.init();
// try setting it as the current subscriber-to-source
if (!current.compareAndSet(ps, u)) {
// did not work, perhaps a new subscriber arrived
// and created a new subscriber-to-source as well, retry
continue;
}
ps = u;
}
// if connect() was called concurrently, only one of them should actually
// connect to the source
doConnect = !ps.shouldConnect.get() && ps.shouldConnect.compareAndSet(false, true);
break; // NOPMD
}
/*
* Notify the callback that we have a (new) connection which it can unsubscribe
* but since ps is unique to a connection, multiple calls to connect() will return the
* same Subscription and even if there was a connect-disconnect-connect pair, the older
* references won't disconnect the newer connection.
* Synchronous source consumers have the opportunity to disconnect via unsubscribe on the
* Subscription as unsafeSubscribe may never return in its own.
*
* Note however, that asynchronously disconnecting a running source might leave
* child-subscribers without any terminal event; ReplaySubject does not have this
* issue because the unsubscription was always triggered by the child-subscribers
* themselves.
*/
connection.call(ps);
if (doConnect) {
source.unsafeSubscribe(ps);
}
}
@SuppressWarnings("rawtypes")
static final class ReplaySubscriber<T> extends Subscriber<T> implements Subscription {
/** Holds notifications from upstream. */
final ReplayBuffer<T> buffer;
/** Contains either an onCompleted or an onError token from upstream. */
boolean done;
/** Indicates an empty array of inner producers. */
static final InnerProducer[] EMPTY = new InnerProducer[0];
/** Indicates a terminated ReplaySubscriber. */
static final InnerProducer[] TERMINATED = new InnerProducer[0];
/** Indicates no further InnerProducers are accepted. */
volatile boolean terminated;
/** Tracks the subscribed producers. Guarded by itself. */
final OpenHashSet<InnerProducer<T>> producers;
/** Contains a copy of the producers. Modified only from the source side. */
InnerProducer<T>[] producersCache;
/** Contains number of modifications to the producers set.*/
volatile long producersVersion;
/** Contains the number of modifications that the producersCache holds. */
long producersCacheVersion;
/**
* Atomically changed from false to true by connect to make sure the
* connection is only performed by one thread.
*/
final AtomicBoolean shouldConnect;
/** Guarded by this. */
boolean emitting;
/** Guarded by this. */
boolean missed;
/** Contains the maximum element index the child Subscribers requested so far. Accessed while emitting is true. */
long maxChildRequested;
/** Counts the outstanding upstream requests until the producer arrives. */
long maxUpstreamRequested;
/** The upstream producer. */
volatile Producer producer;
/** The queue that holds producers with request changes that need to be coordinated. */
List<InnerProducer<T>> coordinationQueue;
/** Indicate that all request amounts should be considered. */
boolean coordinateAll;
@SuppressWarnings("unchecked")
public ReplaySubscriber(ReplayBuffer<T> buffer) {
this.buffer = buffer;
this.producers = new OpenHashSet<InnerProducer<T>>();
this.producersCache = EMPTY;
this.shouldConnect = new AtomicBoolean();
// make sure the source doesn't produce values until the child subscribers
// expressed their request amounts
this.request(0);
}
/** Should be called after the constructor finished to setup nulling-out the current reference. */
void init() {
add(Subscriptions.create(new Action0() {
@Override
public void call() {
if (!terminated) {
synchronized (producers) {
if (!terminated) {
producers.terminate();
producersVersion++;
terminated = true;
}
}
}
// unlike OperatorPublish, we can't null out the terminated so
// late subscribers can still get replay
// current.compareAndSet(ReplaySubscriber.this, null);
// we don't care if it fails because it means the current has
// been replaced in the meantime
}
}));
}
/**
* Atomically try adding a new InnerProducer to this Subscriber or return false if this
* Subscriber was terminated.
* @param producer the producer to add
* @return true if succeeded, false otherwise
*/
boolean add(InnerProducer<T> producer) {
if (producer == null) {
throw new NullPointerException();
}
if (terminated) {
return false;
}
synchronized (producers) {
if (terminated) {
return false;
}
producers.add(producer);
producersVersion++;
}
return true;
}
/**
* Atomically removes the given producer from the producers array.
* @param producer the producer to remove
*/
@SuppressWarnings("unchecked")
void remove(InnerProducer<T> producer) {
if (terminated) {
return;
}
synchronized (producers) {
if (terminated) {
return;
}
producers.remove(producer);
if (producers.isEmpty()) {
producersCache = EMPTY;
}
producersVersion++;
}
}
@Override
public void setProducer(Producer p) {
Producer p0 = producer;
if (p0 != null) {
throw new IllegalStateException("Only a single producer can be set on a Subscriber.");
}
producer = p;
manageRequests(null);
replay();
}
@Override
public void onNext(T t) {
if (!done) {
buffer.next(t);
replay();
}
}
@Override
public void onError(Throwable e) {
// The observer front is accessed serially as required by spec so
// no need to CAS in the terminal value
if (!done) {
done = true;
try {
buffer.error(e);
replay();
} finally {
unsubscribe(); // expectation of testIssue2191
}
}
}
@Override
public void onCompleted() {
// The observer front is accessed serially as required by spec so
// no need to CAS in the terminal value
if (!done) {
done = true;
try {
buffer.complete();
replay();
} finally {
unsubscribe();
}
}
}
/**
* Coordinates the request amounts of various child Subscribers.
*/
void manageRequests(InnerProducer<T> inner) {
// if the upstream has completed, no more requesting is possible
if (isUnsubscribed()) {
return;
}
synchronized (this) {
if (emitting) {
if (inner != null) {
List<InnerProducer<T>> q = coordinationQueue;
if (q == null) {
q = new ArrayList<InnerProducer<T>>();
coordinationQueue = q;
}
q.add(inner);
} else {
coordinateAll = true;
}
missed = true;
return;
}
emitting = true;
}
long ri = maxChildRequested;
long maxTotalRequested;
if (inner != null) {
maxTotalRequested = Math.max(ri, inner.totalRequested.get());
} else {
maxTotalRequested = ri;
InnerProducer<T>[] a = copyProducers();
for (InnerProducer<T> rp : a) {
if (rp != null) {
maxTotalRequested = Math.max(maxTotalRequested, rp.totalRequested.get());
}
}
}
makeRequest(maxTotalRequested, ri);
for (;;) {
// if the upstream has completed, no more requesting is possible
if (isUnsubscribed()) {
return;
}
List<InnerProducer<T>> q;
boolean all;
synchronized (this) {
if (!missed) {
emitting = false;
return;
}
missed = false;
q = coordinationQueue;
coordinationQueue = null;
all = coordinateAll;
coordinateAll = false;
}
ri = maxChildRequested;
maxTotalRequested = ri;
if (q != null) {
for (InnerProducer<T> rp : q) {
maxTotalRequested = Math.max(maxTotalRequested, rp.totalRequested.get());
}
}
if (all) {
InnerProducer<T>[] a = copyProducers();
for (InnerProducer<T> rp : a) {
if (rp != null) {
maxTotalRequested = Math.max(maxTotalRequested, rp.totalRequested.get());
}
}
}
makeRequest(maxTotalRequested, ri);
}
}
InnerProducer<T>[] copyProducers() {
synchronized (producers) {
Object[] a = producers.values();
int n = a.length;
@SuppressWarnings("unchecked")
InnerProducer<T>[] result = new InnerProducer[n];
System.arraycopy(a, 0, result, 0, n);
return result;
}
}
void makeRequest(long maxTotalRequests, long previousTotalRequests) {
long ur = maxUpstreamRequested;
Producer p = producer;
long diff = maxTotalRequests - previousTotalRequests;
if (diff != 0) {
maxChildRequested = maxTotalRequests;
if (p != null) {
if (ur != 0L) {
maxUpstreamRequested = 0L;
p.request(ur + diff);
} else {
p.request(diff);
}
} else {
// collect upstream request amounts until there is a producer for them
long u = ur + diff;
if (u < 0) {
u = Long.MAX_VALUE;
}
maxUpstreamRequested = u;
}
} else
// if there were outstanding upstream requests and we have a producer
if (ur != 0L && p != null) {
maxUpstreamRequested = 0L;
// fire the accumulated requests
p.request(ur);
}
}
/**
* Tries to replay the buffer contents to all known subscribers.
*/
@SuppressWarnings("unchecked")
void replay() {
InnerProducer<T>[] pc = producersCache;
if (producersCacheVersion != producersVersion) {
synchronized (producers) {
pc = producersCache;
// if the producers hasn't changed do nothing
// otherwise make a copy of the current set of producers
Object[] a = producers.values();
int n = a.length;
if (pc.length != n) {
pc = new InnerProducer[n];
producersCache = pc;
}
System.arraycopy(a, 0, pc, 0, n);
producersCacheVersion = producersVersion;
}
}
ReplayBuffer<T> b = buffer;
for (InnerProducer<T> rp : pc) {
if (rp != null) {
b.replay(rp);
}
}
}
}
/**
* A Producer and Subscription that manages the request and unsubscription state of a
* child subscriber in thread-safe manner.
* We use AtomicLong as a base class to save on extra allocation of an AtomicLong and also
* save the overhead of the AtomicIntegerFieldUpdater.
* @param <T> the value type
*/
static final class InnerProducer<T> extends AtomicLong implements Producer, Subscription {
/** */
private static final long serialVersionUID = -4453897557930727610L;
/**
* The parent subscriber-to-source used to allow removing the child in case of
* child unsubscription.
*/
final ReplaySubscriber<T> parent;
/** The actual child subscriber. */
Subscriber<? super T> child;
/**
* Holds an object that represents the current location in the buffer.
* Guarded by the emitter loop.
*/
Object index;
/**
* Keeps the sum of all requested amounts.
*/
final AtomicLong totalRequested;
/** Indicates an emission state. Guarded by this. */
boolean emitting;
/** Indicates a missed update. Guarded by this. */
boolean missed;
/**
* Indicates this child has been unsubscribed: the state is swapped in atomically and
* will prevent the dispatch() to emit (too many) values to a terminated child subscriber.
*/
static final long UNSUBSCRIBED = Long.MIN_VALUE;
public InnerProducer(ReplaySubscriber<T> parent, Subscriber<? super T> child) {
this.parent = parent;
this.child = child;
this.totalRequested = new AtomicLong();
}
@Override
public void request(long n) {
// ignore negative requests
if (n < 0) {
return;
}
// In general, RxJava doesn't prevent concurrent requests (with each other or with
// an unsubscribe) so we need a CAS-loop, but we need to handle
// request overflow and unsubscribed/not requested state as well.
for (;;) {
// get the current request amount
long r = get();
// if child called unsubscribe() do nothing
if (r == UNSUBSCRIBED) {
return;
}
// ignore zero requests except any first that sets in zero
if (r >= 0L && n == 0) {
return;
}
// otherwise, increase the request count
long u = r + n;
// and check for long overflow
if (u < 0) {
// cap at max value, which is essentially unlimited
u = Long.MAX_VALUE;
}
// try setting the new request value
if (compareAndSet(r, u)) {
// increment the total request counter
addTotalRequested(n);
// if successful, notify the parent dispatcher this child can receive more
// elements
parent.manageRequests(this);
parent.buffer.replay(this);
return;
}
// otherwise, someone else changed the state (perhaps a concurrent
// request or unsubscription so retry
}
}
/**
* Increments the total requested amount.
* @param n the additional request amount
*/
void addTotalRequested(long n) {
for (;;) {
long r = totalRequested.get();
long u = r + n;
if (u < 0) {
u = Long.MAX_VALUE;
}
if (totalRequested.compareAndSet(r, u)) {
return;
}
}
}
/**
* Indicate that values have been emitted to this child subscriber by the dispatch() method.
* @param n the number of items emitted
* @return the updated request value (may indicate how much can be produced or a terminal state)
*/
public long produced(long n) {
// we don't allow producing zero or less: it would be a bug in the operator
if (n <= 0) {
throw new IllegalArgumentException("Cant produce zero or less");
}
for (;;) {
// get the current request value
long r = get();
// if the child has unsubscribed, simply return and indicate this
if (r == UNSUBSCRIBED) {
return UNSUBSCRIBED;
}
// reduce the requested amount
long u = r - n;
// if the new amount is less than zero, we have a bug in this operator
if (u < 0) {
throw new IllegalStateException("More produced (" + n + ") than requested (" + r + ")");
}
// try updating the request value
if (compareAndSet(r, u)) {
// and return the updated value
return u;
}
// otherwise, some concurrent activity happened and we need to retry
}
}
@Override
public boolean isUnsubscribed() {
return get() == UNSUBSCRIBED;
}
@Override
public void unsubscribe() {
long r = get();
// let's see if we are unsubscribed
if (r != UNSUBSCRIBED) {
// if not, swap in the terminal state, this is idempotent
// because other methods using CAS won't overwrite this value,
// concurrent calls to unsubscribe will atomically swap in the same
// terminal value
r = getAndSet(UNSUBSCRIBED);
// and only one of them will see a non-terminated value before the swap
if (r != UNSUBSCRIBED) {
// remove this from the parent
parent.remove(this);
// After removal, we might have unblocked the other child subscribers:
// let's assume this child had 0 requested before the unsubscription while
// the others had non-zero. By removing this 'blocking' child, the others
// are now free to receive events
parent.manageRequests(this);
// break the reference
child = null;
}
}
}
/**
* Convenience method to auto-cast the index object.
* @return the associated index object or null
*/
@SuppressWarnings("unchecked")
<U> U index() {
return (U)index;
}
}
/**
* The interface for interacting with various buffering logic.
*
* @param <T> the value type
*/
interface ReplayBuffer<T> {
/**
* Adds a regular value to the buffer.
* @param value the value to buffer
*/
void next(T value);
/**
* Adds a terminal exception to the buffer
* @param e the Throwable to buffer
*/
void error(Throwable e);
/**
* Adds a completion event to the buffer
*/
void complete();
/**
* Tries to replay the buffered values to the
* subscriber inside the output if there
* is new value and requests available at the
* same time.
* @param output the producer of the downstream consumer
*/
void replay(InnerProducer<T> output);
}
/**
* Holds an unbounded list of events.
*
* @param <T> the value type
*/
static final class UnboundedReplayBuffer<T> extends ArrayList<Object> implements ReplayBuffer<T> {
/** */
private static final long serialVersionUID = 7063189396499112664L;
/** The total number of events in the buffer. */
volatile int size;
public UnboundedReplayBuffer(int capacityHint) {
super(capacityHint);
}
@Override
public void next(T value) {
add(NotificationLite.next(value));
size++;
}
@Override
public void error(Throwable e) {
add(NotificationLite.error(e));
size++;
}
@Override
public void complete() {
add(NotificationLite.completed());
size++;
}
@Override
public void replay(InnerProducer<T> output) {
synchronized (output) {
if (output.emitting) {
output.missed = true;
return;
}
output.emitting = true;
}
for (;;) {
if (output.isUnsubscribed()) {
return;
}
int sourceIndex = size;
Integer destinationIndexObject = output.index();
int destinationIndex = destinationIndexObject != null ? destinationIndexObject : 0;
Subscriber<? super T> child = output.child;
if (child == null) {
return;
}
long r = output.get();
long e = 0L;
while (e != r && destinationIndex < sourceIndex) {
Object o = get(destinationIndex);
try {
if (NotificationLite.accept(child, o)) {
return;
}
} catch (Throwable err) {
Exceptions.throwIfFatal(err);
output.unsubscribe();
if (!NotificationLite.isError(o) && !NotificationLite.isCompleted(o)) {
child.onError(OnErrorThrowable.addValueAsLastCause(err, NotificationLite.getValue(o)));
}
return;
}
if (output.isUnsubscribed()) {
return;
}
destinationIndex++;
e++;
}
if (e != 0L) {
output.index = destinationIndex;
if (r != Long.MAX_VALUE) {
output.produced(e);
}
}
synchronized (output) {
if (!output.missed) {
output.emitting = false;
return;
}
output.missed = false;
}
}
}
}
/**
* Represents a node in a bounded replay buffer's linked list.
*/
static final class Node extends AtomicReference<Node> {
/** */
private static final long serialVersionUID = 245354315435971818L;
/** The contained value. */
final Object value;
/** The absolute index of the value. */
final long index;
public Node(Object value, long index) {
this.value = value;
this.index = index;
}
}
/**
* Base class for bounded buffering with options to specify an
* enter and leave transforms and custom truncation behavior.
*
* @param <T> the value type
*/
static class BoundedReplayBuffer<T> extends AtomicReference<Node> implements ReplayBuffer<T> {
/** */
private static final long serialVersionUID = 2346567790059478686L;
Node tail;
int size;
/** The total number of received values so far. */
long index;
public BoundedReplayBuffer() {
Node n = new Node(null, 0);
tail = n;
set(n);
}
/**
* Add a new node to the linked list.
* @param n the node to add as last
*/
final void addLast(Node n) {
tail.set(n);
tail = n;
size++;
}
/**
* Remove the first node from the linked list.
*/
final void removeFirst() {
Node head = get();
Node next = head.get();
if (next == null) {
throw new IllegalStateException("Empty list!");
}
size--;
// can't just move the head because it would retain the very first value
// can't null out the head's value because of late replayers would see null
setFirst(next);
}
/* test */ final void removeSome(int n) {
Node head = get();
while (n > 0) {
head = head.get();
n--;
size--;
}
setFirst(head);
}
/**
* Arranges the given node is the new head from now on.
* @param n the node to set as first
*/
final void setFirst(Node n) {
set(n);
}
/**
* Returns the current head for initializing the replay location
* for a new subscriber.
* Override it to consider linked but outdated elements.
* @return the current head
*/
Node getInitialHead() {
return get();
}
@Override
public final void next(T value) {
Object o = enterTransform(NotificationLite.next(value));
Node n = new Node(o, ++index);
addLast(n);
truncate();
}
@Override
public final void error(Throwable e) {
Object o = enterTransform(NotificationLite.error(e));
Node n = new Node(o, ++index);
addLast(n);
truncateFinal();
}
@Override
public final void complete() {
Object o = enterTransform(NotificationLite.completed());
Node n = new Node(o, ++index);
addLast(n);
truncateFinal();
}
@Override
public final void replay(InnerProducer<T> output) {
synchronized (output) {
if (output.emitting) {
output.missed = true;
return;
}
output.emitting = true;
}
for (;;) {
if (output.isUnsubscribed()) {
return;
}
Node node = output.index();
if (node == null) {
node = getInitialHead();
output.index = node;
/*
* Since this is a latecomer, fix its total requested amount
* as if it got all the values up to the node.index
*/
output.addTotalRequested(node.index);
}
if (output.isUnsubscribed()) {
return;
}
Subscriber<? super T> child = output.child;
if (child == null) {
return;
}
long r = output.get();
long e = 0L;
while (e != r) {
Node v = node.get();
if (v != null) {
Object o = leaveTransform(v.value);
try {
if (NotificationLite.accept(child, o)) {
output.index = null;
return;
}
} catch (Throwable err) {
output.index = null;
Exceptions.throwIfFatal(err);
output.unsubscribe();
if (!NotificationLite.isError(o) && !NotificationLite.isCompleted(o)) {
child.onError(OnErrorThrowable.addValueAsLastCause(err, NotificationLite.getValue(o)));
}
return;
}
e++;
node = v;
} else {
break;
}
if (output.isUnsubscribed()) {
return;
}
}
if (e != 0L) {
output.index = node;
if (r != Long.MAX_VALUE) {
output.produced(e);
}
}
synchronized (output) {
if (!output.missed) {
output.emitting = false;
return;
}
output.missed = false;
}
}
}
/**
* Override this to wrap the NotificationLite object into a
* container to be used later by truncate.
* @param value the value to transform into the internal representation
* @return the internal representation of the value
*/
Object enterTransform(Object value) {
return value;
}
/**
* Override this to unwrap the transformed value into a
* NotificationLite object.
* @param value the value to transform back to external representation from
* the internal representation
* @return the external representation of the value
*/
Object leaveTransform(Object value) {
return value;
}
/**
* Override this method to truncate a non-terminated buffer
* based on its current properties.
*/
void truncate() {
// no op by default
}
/**
* Override this method to truncate a terminated buffer
* based on its properties (i.e., truncate but the very last node).
*/
void truncateFinal() {
// no op by default
}
/* test */ final void collect(Collection<? super T> output) {
Node n = getInitialHead();
for (;;) {
Node next = n.get();
if (next != null) {
Object o = next.value;
Object v = leaveTransform(o);
if (NotificationLite.isCompleted(v) || NotificationLite.isError(v)) {
break;
}
output.add(NotificationLite.<T>getValue(v));
n = next;
} else {
break;
}
}
}
/* test */ boolean hasError() {
return tail.value != null && NotificationLite.isError(leaveTransform(tail.value));
}
/* test */ boolean hasCompleted() {
return tail.value != null && NotificationLite.isCompleted(leaveTransform(tail.value));
}
}
/**
* A bounded replay buffer implementation with size limit only.
*
* @param <T> the value type
*/
static final class SizeBoundReplayBuffer<T> extends BoundedReplayBuffer<T> {
/** */
private static final long serialVersionUID = -5898283885385201806L;
final int limit;
public SizeBoundReplayBuffer(int limit) {
this.limit = limit;
}
@Override
void truncate() {
// overflow can be at most one element
if (size > limit) {
removeFirst();
}
}
// no need for final truncation because values are truncated one by one
}
/**
* Size and time bound replay buffer.
*
* @param <T> the buffered value type
*/
static final class SizeAndTimeBoundReplayBuffer<T> extends BoundedReplayBuffer<T> {
/** */
private static final long serialVersionUID = 3457957419649567404L;
final Scheduler scheduler;
final long maxAgeInMillis;
final int limit;
public SizeAndTimeBoundReplayBuffer(int limit, long maxAgeInMillis, Scheduler scheduler) {
this.scheduler = scheduler;
this.limit = limit;
this.maxAgeInMillis = maxAgeInMillis;
}
@Override
Object enterTransform(Object value) {
return new Timestamped<Object>(scheduler.now(), value);
}
@Override
Object leaveTransform(Object value) {
return ((Timestamped<?>)value).getValue();
}
@Override
Node getInitialHead() {
long timeLimit = scheduler.now() - maxAgeInMillis;
Node prev = get();
Node next = prev.get();
while (next != null) {
Object o = next.value;
Object v = leaveTransform(o);
if (NotificationLite.isCompleted(v) || NotificationLite.isError(v)) {
break;
}
if (((Timestamped<?>)o).getTimestampMillis() <= timeLimit) {
prev = next;
next = next.get();
} else {
break;
}
}
return prev;
}
@Override
void truncate() {
long timeLimit = scheduler.now() - maxAgeInMillis;
Node prev = get();
Node next = prev.get();
int e = 0;
for (;;) {
if (next != null) {
if (size > limit) {
e++;
size--;
prev = next;
next = next.get();
} else {
Timestamped<?> v = (Timestamped<?>)next.value;
if (v.getTimestampMillis() <= timeLimit) {
e++;
size--;
prev = next;
next = next.get();
} else {
break;
}
}
} else {
break;
}
}
if (e != 0) {
setFirst(prev);
}
}
@Override
void truncateFinal() {
long timeLimit = scheduler.now() - maxAgeInMillis;
Node prev = get();
Node next = prev.get();
int e = 0;
for (;;) {
if (next != null && size > 1) {
Timestamped<?> v = (Timestamped<?>)next.value;
if (v.getTimestampMillis() <= timeLimit) {
e++;
size--;
prev = next;
next = next.get();
} else {
break;
}
} else {
break;
}
}
if (e != 0) {
setFirst(prev);
}
}
}
}