/**

public static <T> ReplaySubject<T> create() {

return create(16);

}

/**

public static <T> ReplaySubject<T> create(int capacity) {

final UnboundedReplayState<T> state = new UnboundedReplayState<T>(capacity);

SubjectSubscriptionManager<T> ssm = new SubjectSubscriptionManager<T>();

ssm.onStart = new Action1<SubjectObserver<T>>() {

@Override

public void call(SubjectObserver<T> o) {

// replay history for this observer using the subscribing thread

int lastIndex = state.replayObserverFromIndex(0, o);

// now that it is caught up add to observers

o.index(lastIndex);

}

};

ssm.onAdded = new Action1<SubjectObserver<T>>() {

@Override

public void call(SubjectObserver<T> o) {

synchronized (o) {

if (!o.first || o.emitting) {

return;

}

o.first = false;

o.emitting = true;

}

boolean skipFinal = false;

try {

for (;;) {

int idx = o.<Integer>index();

int sidx = state.index;

if (idx != sidx) {

Integer j = state.replayObserverFromIndex(idx, o);

o.index(j);

}

synchronized (o) {

if (sidx == state.index) {

o.emitting = false;

skipFinal = true;

break;

}

}

}

} finally {

if (!skipFinal) {

synchronized (o) {

o.emitting = false;

}

}

}

}

};

ssm.onTerminated = new Action1<SubjectObserver<T>>() {

@Override

public void call(SubjectObserver<T> o) {

Integer idx = o.index();

if (idx == null) {

idx = 0;

}

// we will finish replaying if there is anything left

state.replayObserverFromIndex(idx, o);

}

};

return new ReplaySubject<T>(ssm, ssm, state);

}

/**

/* public */ static <T> ReplaySubject<T> createUnbounded() {

final BoundedState<T> state = new BoundedState<T>(

new EmptyEvictionPolicy(),

UtilityFunctions.identity(),

UtilityFunctions.identity()

);

return createWithState(state, new DefaultOnAdd<T>(state));

}

/**

public static <T> ReplaySubject<T> createWithSize(int size) {

final BoundedState<T> state = new BoundedState<T>(

new SizeEvictionPolicy(size),

UtilityFunctions.identity(),

UtilityFunctions.identity()

);

return createWithState(state, new DefaultOnAdd<T>(state));

}

/**

public static <T> ReplaySubject<T> createWithTime(long time, TimeUnit unit, final Scheduler scheduler) {

final BoundedState<T> state = new BoundedState<T>(

new TimeEvictionPolicy(unit.toMillis(time), scheduler),

new AddTimestamped(scheduler),

new RemoveTimestamped()

);

return createWithState(state, new TimedOnAdd<T>(state, scheduler));

}

/**

public static <T> ReplaySubject<T> createWithTimeAndSize(long time, TimeUnit unit, int size, final Scheduler scheduler) {

final BoundedState<T> state = new BoundedState<T>(

new PairEvictionPolicy(

new SizeEvictionPolicy(size),

new TimeEvictionPolicy(unit.toMillis(time), scheduler)

),

new AddTimestamped(scheduler),

new RemoveTimestamped()

);

return createWithState(state, new TimedOnAdd<T>(state, scheduler));

}

/**

static final <T> ReplaySubject<T> createWithState(final BoundedState<T> state,

Action1<SubjectObserver<T>> onStart) {

SubjectSubscriptionManager<T> ssm = new SubjectSubscriptionManager<T>();

ssm.onStart = onStart;

ssm.onAdded = new Action1<SubjectObserver<T>>() {

@Override

public void call(SubjectObserver<T> o) {

synchronized (o) {

if (!o.first || o.emitting) {

return;

}

o.first = false;

o.emitting = true;

}

boolean skipFinal = false;

try {

for (;;) {

NodeList.Node<Object> idx = o.index();

NodeList.Node<Object> sidx = state.tail();

if (idx != sidx) {

NodeList.Node<Object> j = state.replayObserverFromIndex(idx, o);

o.index(j);

}

synchronized (o) {

if (sidx == state.tail()) {

o.emitting = false;

skipFinal = true;

break;

}

}

}

} finally {

if (!skipFinal) {

synchronized (o) {

o.emitting = false;

}

}

}

}

};

ssm.onTerminated = new Action1<SubjectObserver<T>>() {

@Override

public void call(SubjectObserver<T> t1) {

NodeList.Node<Object> l = t1.index();

if (l == null) {

l = state.head();

}

state.replayObserverFromIndex(l, t1);

}

};

return new ReplaySubject<T>(ssm, ssm, state);

}

/** The state storing the history and the references. */

final ReplayState<T, ?> state;

/** The manager of subscribers. */

final SubjectSubscriptionManager<T> ssm;

ReplaySubject(OnSubscribe<T> onSubscribe, SubjectSubscriptionManager<T> ssm, ReplayState<T, ?> state) {

super(onSubscribe);

this.ssm = ssm;

this.state = state;

}

@Override

public void onNext(T t) {

if (ssm.active) {

state.next(t);

for (SubjectSubscriptionManager.SubjectObserver<? super T> o : ssm.observers()) {

if (caughtUp(o)) {

o.onNext(t);

}

}

}

}

@Override

public void onError(final Throwable e) {

if (ssm.active) {

state.error(e);

List<Throwable> errors = null;

for (SubjectObserver<? super T> o : ssm.terminate(NotificationLite.instance().error(e))) {

try {

if (caughtUp(o)) {

o.onError(e);

}

} catch (Throwable e2) {

if (errors == null) {

errors = new ArrayList<Throwable>();

}

errors.add(e2);

}

}

Exceptions.throwIfAny(errors);

}

}

@Override

public void onCompleted() {

if (ssm.active) {

state.complete();

for (SubjectObserver<? super T> o : ssm.terminate(NotificationLite.instance().completed())) {

if (caughtUp(o)) {

o.onCompleted();

}

}

}

}

/**

/* Support test. */int subscriberCount() {

return ssm.state.observers.length;

}

@Override

public boolean hasObservers() {

return ssm.observers().length > 0;

}

private boolean caughtUp(SubjectObserver<? super T> o) {

if (!o.caughtUp) {

if (state.replayObserver(o)) {

o.caughtUp = true;

o.index(null); // once caught up, no need for the index anymore

}

return false;

} else {

// it was caught up so proceed the "raw route"

return true;

}

}

// *********************

// State implementations

// *********************

/**

static final class UnboundedReplayState<T> implements ReplayState<T, Integer> {

private final NotificationLite<T> nl = NotificationLite.instance();

/** The buffer. */

private final ArrayList<Object> list;

/** The termination flag. */

private volatile boolean terminated;

/** The size of the buffer. */

volatile int index;

@SuppressWarnings("rawtypes")

static final AtomicIntegerFieldUpdater<UnboundedReplayState> INDEX_UPDATER

= AtomicIntegerFieldUpdater.newUpdater(UnboundedReplayState.class, "index");

public UnboundedReplayState(int initialCapacity) {

list = new ArrayList<Object>(initialCapacity);

}

@Override

public void next(T n) {

if (!terminated) {

list.add(nl.next(n));

INDEX_UPDATER.getAndIncrement(this); // release index

}

}

public void accept(Observer<? super T> o, int idx) {

nl.accept(o, list.get(idx));

}

@Override

public void complete() {

if (!terminated) {

terminated = true;

list.add(nl.completed());

INDEX_UPDATER.getAndIncrement(this); // release index

}

}

@Override

public void error(Throwable e) {

if (!terminated) {

terminated = true;

list.add(nl.error(e));

INDEX_UPDATER.getAndIncrement(this); // release index

}

}

@Override

public boolean terminated() {

return terminated;

}

@Override

public boolean replayObserver(SubjectObserver<? super T> observer) {

synchronized (observer) {

observer.first = false;

if (observer.emitting) {

return false;

}

}

Integer lastEmittedLink = observer.index();

if (lastEmittedLink != null) {

int l = replayObserverFromIndex(lastEmittedLink, observer);

observer.index(l);

return true;

} else {

throw new IllegalStateException("failed to find lastEmittedLink for: " + observer);

}

}

@Override

public Integer replayObserverFromIndex(Integer idx, SubjectObserver<? super T> observer) {

int i = idx;

while (i < index) {

accept(observer, i);

i++;

}

return i;

}

@Override

public Integer replayObserverFromIndexTest(Integer idx, SubjectObserver<? super T> observer, long now) {

return replayObserverFromIndex(idx, observer);

}

@Override

public int size() {

int idx = index; // aquire

if (idx > 0) {

Object o = list.get(idx - 1);

if (nl.isCompleted(o) || nl.isError(o)) {

return idx - 1; // do not report a terminal event as part of size

}

}

return idx;

}

@Override

public boolean isEmpty() {

return size() == 0;

}

@Override

@SuppressWarnings("unchecked")

public T[] toArray(T[] a) {

int s = size();

if (s > 0) {

if (s > a.length) {

a = (T[])Array.newInstance(a.getClass().getComponentType(), s);

}

for (int i = 0; i < s; i++) {

a[i] = (T)list.get(i);

}

if (a.length > s) {

a[s] = null;

}

} else

if (a.length > 0) {

a[0] = null;

}

return a;

}

@Override

public T latest() {

int idx = index;

if (idx > 0) {

Object o = list.get(idx - 1);

if (nl.isCompleted(o) || nl.isError(o)) {

if (idx > 1) {

return nl.getValue(list.get(idx - 2));

}

return null;

}

return nl.getValue(o);

}

return null;

}

}

/**

static final class BoundedState<T> implements ReplayState<T, NodeList.Node<Object>> {

final NodeList<Object> list;

final EvictionPolicy evictionPolicy;

final Func1<Object, Object> enterTransform;

final Func1<Object, Object> leaveTransform;

final NotificationLite<T> nl = NotificationLite.instance();

volatile boolean terminated;

volatile NodeList.Node<Object> tail;

public BoundedState(EvictionPolicy evictionPolicy, Func1<Object, Object> enterTransform,

Func1<Object, Object> leaveTransform) {

this.list = new NodeList<Object>();

this.tail = list.tail;

this.evictionPolicy = evictionPolicy;

this.enterTransform = enterTransform;

this.leaveTransform = leaveTransform;

}

@Override

public void next(T value) {

if (!terminated) {

list.addLast(enterTransform.call(nl.next(value)));

evictionPolicy.evict(list);

tail = list.tail;

}

}

@Override

public void complete() {

if (!terminated) {

terminated = true;

list.addLast(enterTransform.call(nl.completed()));

evictionPolicy.evictFinal(list);

tail = list.tail;

}

}

@Override

public void error(Throwable e) {

if (!terminated) {

terminated = true;

list.addLast(enterTransform.call(nl.error(e)));

// don't evict the terminal value

evictionPolicy.evictFinal(list);

tail = list.tail;

}

}

public void accept(Observer<? super T> o, NodeList.Node<Object> node) {

nl.accept(o, leaveTransform.call(node.value));

}

/**

public void acceptTest(Observer<? super T> o, NodeList.Node<Object> node, long now) {

Object v = node.value;

if (!evictionPolicy.test(v, now)) {

nl.accept(o, leaveTransform.call(v));

}

}

public Node<Object> head() {

return list.head;

}

public Node<Object> tail() {

return tail;

}

@Override

public boolean replayObserver(SubjectObserver<? super T> observer) {

synchronized (observer) {

observer.first = false;

if (observer.emitting) {

return false;

}

}

NodeList.Node<Object> lastEmittedLink = observer.index();

NodeList.Node<Object> l = replayObserverFromIndex(lastEmittedLink, observer);

observer.index(l);

return true;

}

@Override

public NodeList.Node<Object> replayObserverFromIndex(

NodeList.Node<Object> l, SubjectObserver<? super T> observer) {

while (l != tail()) {

accept(observer, l.next);

l = l.next;

}

return l;

}

@Override

public NodeList.Node<Object> replayObserverFromIndexTest(

NodeList.Node<Object> l, SubjectObserver<? super T> observer, long now) {

while (l != tail()) {

acceptTest(observer, l.next, now);

l = l.next;

}

return l;

}

@Override

public boolean terminated() {

return terminated;

}

@Override

public int size() {

int size = 0;

NodeList.Node<Object> l = head();

NodeList.Node<Object> next = l.next;

while (next != null) {

size++;

l = next;

next = next.next;

}

if (l.value != null) {

Object value = leaveTransform.call(l.value);

if (value != null && (nl.isError(value) || nl.isCompleted(value))) {

return size - 1;

}

}

return size;

}

@Override

public boolean isEmpty() {

NodeList.Node<Object> l = head();

NodeList.Node<Object> next = l.next;

if (next == null) {

return true;

}

Object value = leaveTransform.call(next.value);

return nl.isError(value) || nl.isCompleted(value);

}

@Override

@SuppressWarnings("unchecked")

public T[] toArray(T[] a) {

List<T> list = new ArrayList<T>();

NodeList.Node<Object> l = head();

NodeList.Node<Object> next = l.next;

while (next != null) {

Object o = leaveTransform.call(next.value);

if (next.next == null && (nl.isError(o) || nl.isCompleted(o))) {

break;

} else {

list.add((T)o);

}

l = next;

next = next.next;

}

return list.toArray(a);

}

@Override

public T latest() {

Node<Object> h = head().next;

if (h == null) {

return null;

}

Node<Object> p = null;

while (h != tail()) {

p = h;

h = h.next;

}

Object value = leaveTransform.call(h.value);

if (nl.isError(value) || nl.isCompleted(value)) {

if (p != null) {

value = leaveTransform.call(p.value);

return nl.getValue(value);

}

return null;

}

return nl.getValue(value);

}

}

// **************

// API interfaces

// **************

/**

interface ReplayState<T, I> {

/** @return true if the subject has reached a terminal state. */

boolean terminated();

/**

boolean replayObserver(SubjectObserver<? super T> observer);

/**

I replayObserverFromIndex(

I idx, SubjectObserver<? super T> observer);

/**

I replayObserverFromIndexTest(

I idx, SubjectObserver<? super T> observer, long now);

/**

void next(T value);

/**

void error(Throwable e);

/**

void complete();

/**

int size();

/**

boolean isEmpty();

/**

T[] toArray(T[] a);

/**

T latest();

}

/** Interface to manage eviction checking. */

interface EvictionPolicy {

/**

boolean test(Object value, long now);

/**

void evict(NodeList<Object> list);

/**

void evictFinal(NodeList<Object> list);

}

// ************************

// Callback implementations

// ************************

/**

static final class SizeEvictionPolicy implements EvictionPolicy {

final int maxSize;

public SizeEvictionPolicy(int maxSize) {

this.maxSize = maxSize;

}

@Override

public void evict(NodeList<Object> t1) {

while (t1.size() > maxSize) {

t1.removeFirst();

}

}

@Override

public boolean test(Object value, long now) {

return false; // size gets never stale

}

@Override

public void evictFinal(NodeList<Object> t1) {

while (t1.size() > maxSize + 1) {

t1.removeFirst();

}

}

}

/**

static final class TimeEvictionPolicy implements EvictionPolicy {

final long maxAgeMillis;

final Scheduler scheduler;

public TimeEvictionPolicy(long maxAgeMillis, Scheduler scheduler) {

this.maxAgeMillis = maxAgeMillis;

this.scheduler = scheduler;

}

@Override

public void evict(NodeList<Object> t1) {

long now = scheduler.now();

while (!t1.isEmpty()) {

NodeList.Node<Object> n = t1.head.next;

if (test(n.value, now)) {

t1.removeFirst();

} else {

break;

}

}

}

@Override

public void evictFinal(NodeList<Object> t1) {

long now = scheduler.now();

while (t1.size > 1) {

NodeList.Node<Object> n = t1.head.next;

if (test(n.value, now)) {

t1.removeFirst();

} else {

break;

}

}

}

@Override

public boolean test(Object value, long now) {

Timestamped<?> ts = (Timestamped<?>)value;

return ts.getTimestampMillis() <= now - maxAgeMillis;

}

}

/**

static final class PairEvictionPolicy implements EvictionPolicy {

final EvictionPolicy first;

final EvictionPolicy second;

public PairEvictionPolicy(EvictionPolicy first, EvictionPolicy second) {

this.first = first;

this.second = second;

}

@Override

public void evict(NodeList<Object> t1) {

first.evict(t1);

second.evict(t1);

}

@Override

public void evictFinal(NodeList<Object> t1) {

first.evictFinal(t1);

second.evictFinal(t1);

}

@Override

public boolean test(Object value, long now) {

return first.test(value, now) || second.test(value, now);

}

};

/** Maps the values to Timestamped. */

static final class AddTimestamped implements Func1<Object, Object> {

final Scheduler scheduler;

public AddTimestamped(Scheduler scheduler) {

this.scheduler = scheduler;

}

@Override

public Object call(Object t1) {

return new Timestamped<Object>(scheduler.now(), t1);

}

}

/** Maps timestamped values back to raw objects. */

static final class RemoveTimestamped implements Func1<Object, Object> {

@Override

@SuppressWarnings("unchecked")

public Object call(Object t1) {

return ((Timestamped<Object>)t1).getValue();

}

}

/**

static final class DefaultOnAdd<T> implements Action1<SubjectObserver<T>> {

final BoundedState<T> state;

public DefaultOnAdd(BoundedState<T> state) {

this.state = state;

}

@Override

public void call(SubjectObserver<T> t1) {

NodeList.Node<Object> l = state.replayObserverFromIndex(state.head(), t1);

t1.index(l);

}