b = clients; b != null; b = next) {
next = b.next;
if (b.isClosed()) {
b.next = null;
if (pred == null)
clients = next;
else
pred.next = next;
}
else {
pred = b;
++count;
}
}
return count;
}
/**
* Publishes the given item to each current subscriber by
* asynchronously invoking its {@link Flow.Subscriber#onNext(Object)
* onNext} method, blocking uninterruptibly while resources for any
* subscriber are unavailable. This method returns an estimate of
* the maximum lag (number of items submitted but not yet consumed)
* among all current subscribers. This value is at least one
* (accounting for this submitted item) if there are any
* subscribers, else zero.
*
* If the Executor for this publisher throws a
* RejectedExecutionException (or any other RuntimeException or
* Error) when attempting to asynchronously notify subscribers,
* then this exception is rethrown, in which case not all
* subscribers will have been issued this item.
*
* @param item the (non-null) item to publish
* @return the estimated maximum lag among subscribers
* @throws IllegalStateException if closed
* @throws NullPointerException if item is null
* @throws RejectedExecutionException if thrown by Executor
*/
public int submit(T item) {
return doOffer(item, Long.MAX_VALUE, null);
}
/**
* Publishes the given item, if possible, to each current subscriber
* by asynchronously invoking its {@link
* Flow.Subscriber#onNext(Object) onNext} method. The item may be
* dropped by one or more subscribers if resource limits are
* exceeded, in which case the given handler (if non-null) is
* invoked, and if it returns true, retried once. Other calls to
* methods in this class by other threads are blocked while the
* handler is invoked. Unless recovery is assured, options are
* usually limited to logging the error and/or issuing an {@link
* Flow.Subscriber#onError(Throwable) onError} signal to the
* subscriber.
*
*
This method returns a status indicator: If negative, it
* represents the (negative) number of drops (failed attempts to
* issue the item to a subscriber). Otherwise it is an estimate of
* the maximum lag (number of items submitted but not yet
* consumed) among all current subscribers. This value is at least
* one (accounting for this submitted item) if there are any
* subscribers, else zero.
*
*
If the Executor for this publisher throws a
* RejectedExecutionException (or any other RuntimeException or
* Error) when attempting to asynchronously notify subscribers, or
* the drop handler throws an exception when processing a dropped
* item, then this exception is rethrown.
*
* @param item the (non-null) item to publish
* @param onDrop if non-null, the handler invoked upon a drop to a
* subscriber, with arguments of the subscriber and item; if it
* returns true, an offer is re-attempted (once)
* @return if negative, the (negative) number of drops; otherwise
* an estimate of maximum lag
* @throws IllegalStateException if closed
* @throws NullPointerException if item is null
* @throws RejectedExecutionException if thrown by Executor
*/
public int offer(T item,
BiPredicate, ? super T> onDrop) {
return doOffer(item, 0L, onDrop);
}
/**
* Publishes the given item, if possible, to each current subscriber
* by asynchronously invoking its {@link
* Flow.Subscriber#onNext(Object) onNext} method, blocking while
* resources for any subscription are unavailable, up to the
* specified timeout or until the caller thread is interrupted, at
* which point the given handler (if non-null) is invoked, and if it
* returns true, retried once. (The drop handler may distinguish
* timeouts from interrupts by checking whether the current thread
* is interrupted.) Other calls to methods in this class by other
* threads are blocked while the handler is invoked. Unless
* recovery is assured, options are usually limited to logging the
* error and/or issuing an {@link Flow.Subscriber#onError(Throwable)
* onError} signal to the subscriber.
*
* This method returns a status indicator: If negative, it
* represents the (negative) number of drops (failed attempts to
* issue the item to a subscriber). Otherwise it is an estimate of
* the maximum lag (number of items submitted but not yet
* consumed) among all current subscribers. This value is at least
* one (accounting for this submitted item) if there are any
* subscribers, else zero.
*
*
If the Executor for this publisher throws a
* RejectedExecutionException (or any other RuntimeException or
* Error) when attempting to asynchronously notify subscribers, or
* the drop handler throws an exception when processing a dropped
* item, then this exception is rethrown.
*
* @param item the (non-null) item to publish
* @param timeout how long to wait for resources for any subscriber
* before giving up, in units of {@code unit}
* @param unit a {@code TimeUnit} determining how to interpret the
* {@code timeout} parameter
* @param onDrop if non-null, the handler invoked upon a drop to a
* subscriber, with arguments of the subscriber and item; if it
* returns true, an offer is re-attempted (once)
* @return if negative, the (negative) number of drops; otherwise
* an estimate of maximum lag
* @throws IllegalStateException if closed
* @throws NullPointerException if item is null
* @throws RejectedExecutionException if thrown by Executor
*/
public int offer(T item, long timeout, TimeUnit unit,
BiPredicate, ? super T> onDrop) {
long nanos = unit.toNanos(timeout);
// distinguishes from untimed (only wrt interrupt policy)
if (nanos == Long.MAX_VALUE) --nanos;
return doOffer(item, nanos, onDrop);
}
/**
* Unless already closed, issues {@link
* Flow.Subscriber#onComplete() onComplete} signals to current
* subscribers, and disallows subsequent attempts to publish.
* Upon return, this method does NOT guarantee that all
* subscribers have yet completed.
*/
public void close() {
ReentrantLock lock = this.lock;
if (!closed) {
BufferedSubscription b;
lock.lock();
try {
// no need to re-check closed here
b = clients;
clients = null;
owner = null;
closed = true;
} finally {
lock.unlock();
}
while (b != null) {
BufferedSubscription next = b.next;
b.next = null;
b.onComplete();
b = next;
}
}
}
/**
* Unless already closed, issues {@link
* Flow.Subscriber#onError(Throwable) onError} signals to current
* subscribers with the given error, and disallows subsequent
* attempts to publish. Future subscribers also receive the given
* error. Upon return, this method does NOT guarantee
* that all subscribers have yet completed.
*
* @param error the {@code onError} argument sent to subscribers
* @throws NullPointerException if error is null
*/
public void closeExceptionally(Throwable error) {
if (error == null)
throw new NullPointerException();
ReentrantLock lock = this.lock;
if (!closed) {
BufferedSubscription b;
lock.lock();
try {
b = clients;
if (!closed) { // don't clobber racing close
closedException = error;
clients = null;
owner = null;
closed = true;
}
} finally {
lock.unlock();
}
while (b != null) {
BufferedSubscription next = b.next;
b.next = null;
b.onError(error);
b = next;
}
}
}
/**
* Returns true if this publisher is not accepting submissions.
*
* @return true if closed
*/
public boolean isClosed() {
return closed;
}
/**
* Returns the exception associated with {@link
* #closeExceptionally(Throwable) closeExceptionally}, or null if
* not closed or if closed normally.
*
* @return the exception, or null if none
*/
public Throwable getClosedException() {
return closedException;
}
/**
* Returns true if this publisher has any subscribers.
*
* @return true if this publisher has any subscribers
*/
public boolean hasSubscribers() {
boolean nonEmpty = false;
ReentrantLock lock = this.lock;
lock.lock();
try {
for (BufferedSubscription b = clients; b != null;) {
BufferedSubscription next = b.next;
if (b.isClosed()) {
b.next = null;
b = clients = next;
}
else {
nonEmpty = true;
break;
}
}
} finally {
lock.unlock();
}
return nonEmpty;
}
/**
* Returns the number of current subscribers.
*
* @return the number of current subscribers
*/
public int getNumberOfSubscribers() {
int n;
ReentrantLock lock = this.lock;
lock.lock();
try {
n = cleanAndCount();
} finally {
lock.unlock();
}
return n;
}
/**
* Returns the Executor used for asynchronous delivery.
*
* @return the Executor used for asynchronous delivery
*/
public Executor getExecutor() {
return executor;
}
/**
* Returns the maximum per-subscriber buffer capacity.
*
* @return the maximum per-subscriber buffer capacity
*/
public int getMaxBufferCapacity() {
return maxBufferCapacity;
}
/**
* Returns a list of current subscribers for monitoring and
* tracking purposes, not for invoking {@link Flow.Subscriber}
* methods on the subscribers.
*
* @return list of current subscribers
*/
public List> getSubscribers() {
ArrayList> subs = new ArrayList<>();
ReentrantLock lock = this.lock;
lock.lock();
try {
BufferedSubscription pred = null, next;
for (BufferedSubscription b = clients; b != null; b = next) {
next = b.next;
if (b.isClosed()) {
b.next = null;
if (pred == null)
clients = next;
else
pred.next = next;
}
else {
subs.add(b.subscriber);
pred = b;
}
}
} finally {
lock.unlock();
}
return subs;
}
/**
* Returns true if the given Subscriber is currently subscribed.
*
* @param subscriber the subscriber
* @return true if currently subscribed
* @throws NullPointerException if subscriber is null
*/
public boolean isSubscribed(Subscriber subscriber) {
if (subscriber == null) throw new NullPointerException();
boolean subscribed = false;
ReentrantLock lock = this.lock;
if (!closed) {
lock.lock();
try {
BufferedSubscription pred = null, next;
for (BufferedSubscription b = clients; b != null; b = next) {
next = b.next;
if (b.isClosed()) {
b.next = null;
if (pred == null)
clients = next;
else
pred.next = next;
}
else if (subscribed = subscriber.equals(b.subscriber))
break;
else
pred = b;
}
} finally {
lock.unlock();
}
}
return subscribed;
}
/**
* Returns an estimate of the minimum number of items requested
* (via {@link Flow.Subscription#request(long) request}) but not
* yet produced, among all current subscribers.
*
* @return the estimate, or zero if no subscribers
*/
public long estimateMinimumDemand() {
long min = Long.MAX_VALUE;
boolean nonEmpty = false;
ReentrantLock lock = this.lock;
lock.lock();
try {
BufferedSubscription pred = null, next;
for (BufferedSubscription b = clients; b != null; b = next) {
int n; long d;
next = b.next;
if ((n = b.estimateLag()) < 0) {
b.next = null;
if (pred == null)
clients = next;
else
pred.next = next;
}
else {
if ((d = b.demand - n) < min)
min = d;
nonEmpty = true;
pred = b;
}
}
} finally {
lock.unlock();
}
return nonEmpty ? min : 0;
}
/**
* Returns an estimate of the maximum number of items produced but
* not yet consumed among all current subscribers.
*
* @return the estimate
*/
public int estimateMaximumLag() {
int max = 0;
ReentrantLock lock = this.lock;
lock.lock();
try {
BufferedSubscription pred = null, next;
for (BufferedSubscription b = clients; b != null; b = next) {
int n;
next = b.next;
if ((n = b.estimateLag()) < 0) {
b.next = null;
if (pred == null)
clients = next;
else
pred.next = next;
}
else {
if (n > max)
max = n;
pred = b;
}
}
} finally {
lock.unlock();
}
return max;
}
/**
* Processes all published items using the given Consumer function.
* Returns a CompletableFuture that is completed normally when this
* publisher signals {@link Flow.Subscriber#onComplete()
* onComplete}, or completed exceptionally upon any error, or an
* exception is thrown by the Consumer, or the returned
* CompletableFuture is cancelled, in which case no further items
* are processed.
*
* @param consumer the function applied to each onNext item
* @return a CompletableFuture that is completed normally
* when the publisher signals onComplete, and exceptionally
* upon any error or cancellation
* @throws NullPointerException if consumer is null
*/
public CompletableFuture consume(Consumer consumer) {
if (consumer == null)
throw new NullPointerException();
CompletableFuture status = new CompletableFuture<>();
subscribe(new ConsumerSubscriber(status, consumer));
return status;
}
/** Subscriber for method consume */
static final class ConsumerSubscriber implements Subscriber {
final CompletableFuture status;
final Consumer consumer;
Subscription subscription;
ConsumerSubscriber(CompletableFuture status,
Consumer consumer) {
this.status = status; this.consumer = consumer;
}
public final void onSubscribe(Subscription subscription) {
this.subscription = subscription;
status.whenComplete((v, e) -> subscription.cancel());
if (!status.isDone())
subscription.request(Long.MAX_VALUE);
}
public final void onError(Throwable ex) {
status.completeExceptionally(ex);
}
public final void onComplete() {
status.complete(null);
}
public final void onNext(T item) {
try {
consumer.accept(item);
} catch (Throwable ex) {
subscription.cancel();
status.completeExceptionally(ex);
}
}
}
/**
* A task for consuming buffer items and signals, created and
* executed whenever they become available. A task consumes as
* many items/signals as possible before terminating, at which
* point another task is created when needed. The dual Runnable
* and ForkJoinTask declaration saves overhead when executed by
* ForkJoinPools, without impacting other kinds of Executors.
*/
@SuppressWarnings("serial")
static final class ConsumerTask extends ForkJoinTask
implements Runnable, CompletableFuture.AsynchronousCompletionTask {
final BufferedSubscription consumer;
ConsumerTask(BufferedSubscription consumer) {
this.consumer = consumer;
}
public final Void getRawResult() { return null; }
public final void setRawResult(Void v) {}
public final boolean exec() { consumer.consume(); return false; }
public final void run() { consumer.consume(); }
}
/**
* A resizable array-based ring buffer with integrated control to
* start a consumer task whenever items are available. The buffer
* algorithm is specialized for the case of at most one concurrent
* producer and consumer, and power of two buffer sizes. It relies
* primarily on atomic operations (CAS or getAndSet) at the next
* array slot to put or take an element, at the "tail" and "head"
* indices written only by the producer and consumer respectively.
*
* We ensure internally that there is at most one active consumer
* task at any given time. The publisher guarantees a single
* producer via its lock. Sync among producers and consumers
* relies on volatile fields "ctl", "demand", and "waiting" (along
* with element access). Other variables are accessed in plain
* mode, relying on outer ordering and exclusion, and/or enclosing
* them within other volatile accesses. Some atomic operations are
* avoided by tracking single threaded ownership by producers (in
* the style of biased locking).
*
* Execution control and protocol state are managed using field
* "ctl". Methods to subscribe, close, request, and cancel set
* ctl bits (mostly using atomic boolean method getAndBitwiseOr),
* and ensure that a task is running. (The corresponding consumer
* side actions are in method consume.) To avoid starting a new
* task on each action, ctl also includes a keep-alive bit
* (ACTIVE) that is refreshed if needed on producer actions.
* (Maintaining agreement about keep-alives requires most atomic
* updates to be full SC/Volatile strength, which is still much
* cheaper than using one task per item.) Error signals
* additionally null out items and/or fields to reduce termination
* latency. The cancel() method is supported by treating as ERROR
* but suppressing onError signal.
*
* Support for blocking also exploits the fact that there is only
* one possible waiter. ManagedBlocker-compatible control fields
* are placed in this class itself rather than in wait-nodes.
* Blocking control relies on the "waiting" and "waiter"
* fields. Producers set them before trying to block. Signalling
* unparks and clears fields. If the producer and/or consumer are
* using a ForkJoinPool, the producer attempts to help run
* consumer tasks via ForkJoinPool.helpAsyncBlocker before
* blocking.
*
* Usages of this class may encounter any of several forms of
* memory contention. We try to ameliorate across them without
* unduly impacting footprints in low-contention usages where it
* isn't needed. Buffer arrays start out small and grow only as
* needed. The class uses @Contended and heuristic field
* declaration ordering to reduce false-sharing memory contention
* across instances of BufferedSubscription (as in, multiple
* subscribers per publisher). We additionally segregate some
* fields that would otherwise nearly always encounter cache line
* contention among producers and consumers. To reduce contention
* across time (vs space), consumers only periodically update
* other fields (see method takeItems), at the expense of possibly
* staler reporting of lags and demand (bounded at 12.5% == 1/8
* capacity) and possibly more atomic operations.
*
* Other forms of imbalance and slowdowns can occur during startup
* when producer and consumer methods are compiled and/or memory
* is allocated at different rates. This is ameliorated by
* artificially subdividing some consumer methods, including
* isolation of all subscriber callbacks. This code also includes
* typical power-of-two array screening idioms to avoid compilers
* generating traps, along with the usual SSA-based inline
* assignment coding style. Also, all methods and fields have
* default visibility to simplify usage by callers.
*/
@SuppressWarnings("serial")
@jdk.internal.vm.annotation.Contended
static final class BufferedSubscription
implements Subscription, ForkJoinPool.ManagedBlocker {
long timeout; // Long.MAX_VALUE if untimed wait
int head; // next position to take
int tail; // next position to put
final int maxCapacity; // max buffer size
volatile int ctl; // atomic run state flags
Object[] array; // buffer
final Subscriber subscriber;
final BiConsumer, ? super Throwable> onNextHandler;
Executor executor; // null on error
Thread waiter; // blocked producer thread
Throwable pendingError; // holds until onError issued
BufferedSubscription next; // used only by publisher
BufferedSubscription nextRetry; // used only by publisher
@jdk.internal.vm.annotation.Contended("c") // segregate
volatile long demand; // # unfilled requests
@jdk.internal.vm.annotation.Contended("c")
volatile int waiting; // nonzero if producer blocked
// ctl bit values
static final int CLOSED = 0x01; // if set, other bits ignored
static final int ACTIVE = 0x02; // keep-alive for consumer task
static final int REQS = 0x04; // (possibly) nonzero demand
static final int ERROR = 0x08; // issues onError when noticed
static final int COMPLETE = 0x10; // issues onComplete when done
static final int RUN = 0x20; // task is or will be running
static final int OPEN = 0x40; // true after subscribe
static final long INTERRUPTED = -1L; // timeout vs interrupt sentinel
BufferedSubscription(Subscriber subscriber,
Executor executor,
BiConsumer,
? super Throwable> onNextHandler,
Object[] array,
int maxBufferCapacity) {
this.subscriber = subscriber;
this.executor = executor;
this.onNextHandler = onNextHandler;
this.array = array;
this.maxCapacity = maxBufferCapacity;
}
// Wrappers for some VarHandle methods
final boolean weakCasCtl(int cmp, int val) {
return CTL.weakCompareAndSet(this, cmp, val);
}
final int getAndBitwiseOrCtl(int bits) {
return (int)CTL.getAndBitwiseOr(this, bits);
}
final long subtractDemand(int k) {
long n = (long)(-k);
return n + (long)DEMAND.getAndAdd(this, n);
}
final boolean casDemand(long cmp, long val) {
return DEMAND.compareAndSet(this, cmp, val);
}
// Utilities used by SubmissionPublisher
/**
* Returns true if closed (consumer task may still be running).
*/
final boolean isClosed() {
return (ctl & CLOSED) != 0;
}
/**
* Returns estimated number of buffered items, or negative if
* closed.
*/
final int estimateLag() {
int c = ctl, n = tail - head;
return ((c & CLOSED) != 0) ? -1 : (n < 0) ? 0 : n;
}
// Methods for submitting items
/**
* Tries to add item and start consumer task if necessary.
* @return negative if closed, 0 if saturated, else estimated lag
*/
final int offer(T item, boolean unowned) {
Object[] a;
int stat = 0, cap = ((a = array) == null) ? 0 : a.length;
int t = tail, i = t & (cap - 1), n = t + 1 - head;
if (cap > 0) {
boolean added;
if (n >= cap && cap < maxCapacity) // resize
added = growAndOffer(item, a, t);
else if (n >= cap || unowned) // need volatile CAS
added = QA.compareAndSet(a, i, null, item);
else { // can use release mode
QA.setRelease(a, i, item);
added = true;
}
if (added) {
tail = t + 1;
stat = n;
}
}
return startOnOffer(stat);
}
/**
* Tries to expand buffer and add item, returning true on
* success. Currently fails only if out of memory.
*/
final boolean growAndOffer(T item, Object[] a, int t) {
int cap = 0, newCap = 0;
Object[] newArray = null;
if (a != null && (cap = a.length) > 0 && (newCap = cap << 1) > 0) {
try {
newArray = new Object[newCap];
} catch (OutOfMemoryError ex) {
}
}
if (newArray == null)
return false;
else { // take and move items
int newMask = newCap - 1;
newArray[t-- & newMask] = item;
for (int mask = cap - 1, k = mask; k >= 0; --k) {
Object x = QA.getAndSet(a, t & mask, null);
if (x == null)
break; // already consumed
else
newArray[t-- & newMask] = x;
}
array = newArray;
VarHandle.releaseFence(); // release array and slots
return true;
}
}
/**
* Version of offer for retries (no resize or bias)
*/
final int retryOffer(T item) {
Object[] a;
int stat = 0, t = tail, h = head, cap;
if ((a = array) != null && (cap = a.length) > 0 &&
QA.compareAndSet(a, (cap - 1) & t, null, item))
stat = (tail = t + 1) - h;
return startOnOffer(stat);
}
/**
* Tries to start consumer task after offer.
* @return negative if now closed, else argument
*/
final int startOnOffer(int stat) {
int c; // start or keep alive if requests exist and not active
if (((c = ctl) & (REQS | ACTIVE)) == REQS &&
((c = getAndBitwiseOrCtl(RUN | ACTIVE)) & (RUN | CLOSED)) == 0)
tryStart();
else if ((c & CLOSED) != 0)
stat = -1;
return stat;
}
/**
* Tries to start consumer task. Sets error state on failure.
*/
final void tryStart() {
try {
Executor e;
ConsumerTask task = new ConsumerTask(this);
if ((e = executor) != null) // skip if disabled on error
e.execute(task);
} catch (RuntimeException | Error ex) {
getAndBitwiseOrCtl(ERROR | CLOSED);
throw ex;
}
}
// Signals to consumer tasks
/**
* Sets the given control bits, starting task if not running or closed.
* @param bits state bits, assumed to include RUN but not CLOSED
*/
final void startOnSignal(int bits) {
if ((ctl & bits) != bits &&
(getAndBitwiseOrCtl(bits) & (RUN | CLOSED)) == 0)
tryStart();
}
final void onSubscribe() {
startOnSignal(RUN | ACTIVE);
}
final void onComplete() {
startOnSignal(RUN | ACTIVE | COMPLETE);
}
final void onError(Throwable ex) {
int c; Object[] a; // to null out buffer on async error
if (ex != null)
pendingError = ex; // races are OK
if (((c = getAndBitwiseOrCtl(ERROR | RUN | ACTIVE)) & CLOSED) == 0) {
if ((c & RUN) == 0)
tryStart();
else if ((a = array) != null)
Arrays.fill(a, null);
}
}
public final void cancel() {
onError(null);
}
public final void request(long n) {
if (n > 0L) {
for (;;) {
long p = demand, d = p + n; // saturate
if (casDemand(p, d < p ? Long.MAX_VALUE : d))
break;
}
startOnSignal(RUN | ACTIVE | REQS);
}
else
onError(new IllegalArgumentException(
"non-positive subscription request"));
}
// Consumer task actions
/**
* Consumer loop, called from ConsumerTask, or indirectly when
* helping during submit.
*/
final void consume() {
Subscriber s;
if ((s = subscriber) != null) { // hoist checks
subscribeOnOpen(s);
long d = demand;
for (int h = head, t = tail;;) {
int c, taken; boolean empty;
if (((c = ctl) & ERROR) != 0) {
closeOnError(s, null);
break;
}
else if ((taken = takeItems(s, d, h)) > 0) {
head = h += taken;
d = subtractDemand(taken);
}
else if ((d = demand) == 0L && (c & REQS) != 0)
weakCasCtl(c, c & ~REQS); // exhausted demand
else if (d != 0L && (c & REQS) == 0)
weakCasCtl(c, c | REQS); // new demand
else if (t == (t = tail)) { // stability check
if ((empty = (t == h)) && (c & COMPLETE) != 0) {
closeOnComplete(s); // end of stream
break;
}
else if (empty || d == 0L) {
int bit = ((c & ACTIVE) != 0) ? ACTIVE : RUN;
if (weakCasCtl(c, c & ~bit) && bit == RUN)
break; // un-keep-alive or exit
}
}
}
}
}
/**
* Consumes some items until unavailable or bound or error.
*
* @param s subscriber
* @param d current demand
* @param h current head
* @return number taken
*/
final int takeItems(Subscriber s, long d, int h) {
Object[] a;
int k = 0, cap;
if ((a = array) != null && (cap = a.length) > 0) {
int m = cap - 1, b = (m >>> 3) + 1; // min(1, cap/8)
int n = (d < (long)b) ? (int)d : b;
for (; k < n; ++h, ++k) {
Object x = QA.getAndSet(a, h & m, null);
if (waiting != 0)
signalWaiter();
if (x == null)
break;
else if (!consumeNext(s, x))
break;
}
}
return k;
}
final boolean consumeNext(Subscriber s, Object x) {
try {
@SuppressWarnings("unchecked") T y = (T) x;
if (s != null)
s.onNext(y);
return true;
} catch (Throwable ex) {
handleOnNext(s, ex);
return false;
}
}
/**
* Processes exception in Subscriber.onNext.
*/
final void handleOnNext(Subscriber s, Throwable ex) {
BiConsumer, ? super Throwable> h;
try {
if ((h = onNextHandler) != null)
h.accept(s, ex);
} catch (Throwable ignore) {
}
closeOnError(s, ex);
}
/**
* Issues subscriber.onSubscribe if this is first signal.
*/
final void subscribeOnOpen(Subscriber s) {
if ((ctl & OPEN) == 0 && (getAndBitwiseOrCtl(OPEN) & OPEN) == 0)
consumeSubscribe(s);
}
final void consumeSubscribe(Subscriber s) {
try {
if (s != null) // ignore if disabled
s.onSubscribe(this);
} catch (Throwable ex) {
closeOnError(s, ex);
}
}
/**
* Issues subscriber.onComplete unless already closed.
*/
final void closeOnComplete(Subscriber s) {
if ((getAndBitwiseOrCtl(CLOSED) & CLOSED) == 0)
consumeComplete(s);
}
final void consumeComplete(Subscriber s) {
try {
if (s != null)
s.onComplete();
} catch (Throwable ignore) {
}
}
/**
* Issues subscriber.onError, and unblocks producer if needed.
*/
final void closeOnError(Subscriber s, Throwable ex) {
if ((getAndBitwiseOrCtl(ERROR | CLOSED) & CLOSED) == 0) {
if (ex == null)
ex = pendingError;
pendingError = null; // detach
executor = null; // suppress racing start calls
signalWaiter();
consumeError(s, ex);
}
}
final void consumeError(Subscriber s, Throwable ex) {
try {
if (ex != null && s != null)
s.onError(ex);
} catch (Throwable ignore) {
}
}
// Blocking support
/**
* Unblocks waiting producer.
*/
final void signalWaiter() {
Thread w;
waiting = 0;
if ((w = waiter) != null)
LockSupport.unpark(w);
}
/**
* Returns true if closed or space available.
* For ManagedBlocker.
*/
public final boolean isReleasable() {
Object[] a; int cap;
return ((ctl & CLOSED) != 0 ||
((a = array) != null && (cap = a.length) > 0 &&
QA.getAcquire(a, (cap - 1) & tail) == null));
}
/**
* Helps or blocks until timeout, closed, or space available.
*/
final void awaitSpace(long nanos) {
if (!isReleasable()) {
ForkJoinPool.helpAsyncBlocker(executor, this);
if (!isReleasable()) {
timeout = nanos;
try {
ForkJoinPool.managedBlock(this);
} catch (InterruptedException ie) {
timeout = INTERRUPTED;
}
if (timeout == INTERRUPTED)
Thread.currentThread().interrupt();
}
}
}
/**
* Blocks until closed, space available or timeout.
* For ManagedBlocker.
*/
public final boolean block() {
long nanos = timeout;
boolean timed = (nanos < Long.MAX_VALUE);
long deadline = timed ? System.nanoTime() + nanos : 0L;
while (!isReleasable()) {
if (Thread.interrupted()) {
timeout = INTERRUPTED;
if (timed)
break;
}
else if (timed && (nanos = deadline - System.nanoTime()) <= 0L)
break;
else if (waiter == null)
waiter = Thread.currentThread();
else if (waiting == 0)
waiting = 1;
else if (timed)
LockSupport.parkNanos(this, nanos);
else
LockSupport.park(this);
}
waiter = null;
waiting = 0;
return true;
}
// VarHandle mechanics
static final VarHandle CTL;
static final VarHandle DEMAND;
static final VarHandle QA;
static {
try {
MethodHandles.Lookup l = MethodHandles.lookup();
CTL = l.findVarHandle(BufferedSubscription.class, "ctl",
int.class);
DEMAND = l.findVarHandle(BufferedSubscription.class, "demand",
long.class);
QA = MethodHandles.arrayElementVarHandle(Object[].class);
} catch (ReflectiveOperationException e) {
throw new ExceptionInInitializerError(e);
}
// Reduce the risk of rare disastrous classloading in first call to
// LockSupport.park: https://bugs.openjdk.java.net/browse/JDK-8074773
Class ensureLoaded = LockSupport.class;
}
}
}