-+ * This function is not MT-Safe. This function cannot be called with other read operations ({@link #peek()}, {@link #poll()},
-+ * {@link #clear()}, etc).
-+ * Write operations are safe to be called concurrently.
-+ *
-+ * @param consumer The consumer to accept the elements.
-+ * @return The total number of elements drained.
-+ */
-+ public int drain(final Consumer consumer) {
-+ return this.drain(consumer, false, ConcurrentUtil::rethrow);
-+ }
-+
-+ /**
-+ * Empties the queue into the specified consumer. This function is optimized for single-threaded reads, and should
-+ * be faster than a loop on {@link #poll()}.
-+ *
-+ * If {@code preventAdds} is {@code true}, then after this function returns the queue is guaranteed to be empty and
-+ * additions to the queue will fail.
-+ *
-+ *
-+ * This function is not MT-Safe. This function cannot be called with other read operations ({@link #peek()}, {@link #poll()},
-+ * {@link #clear()}, etc).
-+ * Write operations are safe to be called concurrently.
-+ *
-+ * @param consumer The consumer to accept the elements.
-+ * @param preventAdds Whether to prevent additions to this queue after draining.
-+ * @return The total number of elements drained.
-+ */
-+ public int drain(final Consumer consumer, final boolean preventAdds) {
-+ return this.drain(consumer, preventAdds, ConcurrentUtil::rethrow);
-+ }
-+
-+ /**
-+ * Empties the queue into the specified consumer. This function is optimized for single-threaded reads, and should
-+ * be faster than a loop on {@link #poll()}.
-+ *
-+ * If {@code preventAdds} is {@code true}, then after this function returns the queue is guaranteed to be empty and
-+ * additions to the queue will fail.
-+ *
-+ *
-+ * This function is not MT-Safe. This function cannot be called with other read operations ({@link #peek()}, {@link #poll()},
-+ * {@link #clear()}, {@link #remove(Object)} etc).
-+ * Only write operations are safe to be called concurrently.
-+ *
-+ * @param consumer The consumer to accept the elements.
-+ * @param preventAdds Whether to prevent additions to this queue after draining.
-+ * @param exceptionHandler Invoked when the consumer raises an exception.
-+ * @return The total number of elements drained.
-+ */
-+ public int drain(final Consumer consumer, final boolean preventAdds, final Consumer exceptionHandler) {
-+ Validate.notNull(consumer, "Null consumer");
-+ Validate.notNull(exceptionHandler, "Null exception handler");
-+
-+ /* This function assumes proper synchronization is made to ensure drain and no other read function are called concurrently */
-+ /* This allows plain write usages instead of opaque or higher */
-+ int total = 0;
-+
-+ final LinkedNode head = this.getHeadAcquire(); /* Required to synchronize with the write to the first element field */
-+ LinkedNode curr = head;
-+
-+ for (;;) {
-+ /* Volatile acquires with the write to the element field */
-+ final E currentVal = curr.getElementPlain();
-+ LinkedNode next = curr.getNextVolatile();
-+
-+ if (next == curr) {
-+ /* Add-locked nodes always have a null value */
-+ break;
-+ }
-+
-+ if (currentVal == null) {
-+ if (next == null) {
-+ if (preventAdds && (next = curr.compareExchangeNextVolatile(null, curr)) != null) {
-+ // failed to prevent adds, continue
-+ curr = next;
-+ continue;
-+ } else {
-+ // we're done here
-+ break;
-+ }
-+ }
-+ curr = next;
-+ continue;
-+ }
-+
-+ try {
-+ consumer.accept(currentVal);
-+ } catch (final Exception ex) {
-+ this.setHeadOpaque(next != null ? next : curr); /* Avoid perf penalty (of reiterating) if the exception handler decides to re-throw */
-+ curr.setElementOpaque(null); /* set here, we might re-throw */
-+
-+ exceptionHandler.accept(ex);
-+ }
-+
-+ curr.setElementOpaque(null);
-+
-+ ++total;
-+
-+ if (next == null) {
-+ if (preventAdds && (next = curr.compareExchangeNextVolatile(null, curr)) != null) {
-+ /* Retry with next value */
-+ curr = next;
-+ continue;
-+ }
-+ break;
-+ }
-+
-+ curr = next;
-+ }
-+ if (curr != head) {
-+ this.setHeadOpaque(curr); /* While this may be a plain write, eventually publish it for methods such as find. */
-+ }
-+ return total;
-+ }
-+
-+ @Override
-+ public Spliterator spliterator() { // TODO implement
-+ return Spliterators.spliterator(this, Spliterator.CONCURRENT |
-+ Spliterator.NONNULL | Spliterator.ORDERED);
-+ }
-+
-+ protected static final class LinkedNode {
-+
-+ protected volatile Object element;
-+ protected volatile LinkedNode next;
-+
-+ protected static final VarHandle ELEMENT_HANDLE = ConcurrentUtil.getVarHandle(LinkedNode.class, "element", Object.class);
-+ protected static final VarHandle NEXT_HANDLE = ConcurrentUtil.getVarHandle(LinkedNode.class, "next", LinkedNode.class);
-+
-+ protected LinkedNode(final Object element, final LinkedNode next) {
-+ ELEMENT_HANDLE.set(this, element);
-+ NEXT_HANDLE.set(this, next);
-+ }
-+
-+ /* element */
-+
-+ @SuppressWarnings("unchecked")
-+ protected final E getElementPlain() {
-+ return (E)ELEMENT_HANDLE.get(this);
-+ }
-+
-+ @SuppressWarnings("unchecked")
-+ protected final E getElementVolatile() {
-+ return (E)ELEMENT_HANDLE.getVolatile(this);
-+ }
-+
-+ protected final void setElementPlain(final E update) {
-+ ELEMENT_HANDLE.set(this, (Object)update);
-+ }
-+
-+ protected final void setElementOpaque(final E update) {
-+ ELEMENT_HANDLE.setOpaque(this, (Object)update);
-+ }
-+
-+ protected final void setElementVolatile(final E update) {
-+ ELEMENT_HANDLE.setVolatile(this, (Object)update);
-+ }
-+
-+ @SuppressWarnings("unchecked")
-+ protected final E getAndSetElementVolatile(final E update) {
-+ return (E)ELEMENT_HANDLE.getAndSet(this, update);
-+ }
-+
-+ @SuppressWarnings("unchecked")
-+ protected final E compareExchangeElementVolatile(final E expect, final E update) {
-+ return (E)ELEMENT_HANDLE.compareAndExchange(this, expect, update);
-+ }
-+
-+ /* next */
-+
-+ @SuppressWarnings("unchecked")
-+ protected final LinkedNode getNextPlain() {
-+ return (LinkedNode)NEXT_HANDLE.get(this);
-+ }
-+
-+ @SuppressWarnings("unchecked")
-+ protected final LinkedNode getNextOpaque() {
-+ return (LinkedNode)NEXT_HANDLE.getOpaque(this);
-+ }
-+
-+ @SuppressWarnings("unchecked")
-+ protected final LinkedNode getNextAcquire() {
-+ return (LinkedNode)NEXT_HANDLE.getAcquire(this);
-+ }
-+
-+ @SuppressWarnings("unchecked")
-+ protected final LinkedNode getNextVolatile() {
-+ return (LinkedNode)NEXT_HANDLE.getVolatile(this);
-+ }
-+
-+ protected final void setNextPlain(final LinkedNode next) {
-+ NEXT_HANDLE.set(this, next);
-+ }
-+
-+ protected final void setNextVolatile(final LinkedNode next) {
-+ NEXT_HANDLE.setVolatile(this, next);
-+ }
-+
-+ @SuppressWarnings("unchecked")
-+ protected final LinkedNode compareExchangeNextVolatile(final LinkedNode expect, final LinkedNode set) {
-+ return (LinkedNode)NEXT_HANDLE.compareAndExchange(this, expect, set);
-+ }
-+ }
-+
-+ protected static final class LinkedIterator implements Iterator {
-+
-+ protected LinkedNode curr; /* last returned by next() */
-+ protected LinkedNode next; /* next to return from next() */
-+ protected E nextElement; /* cached to avoid a race condition with removing or polling */
-+
-+ protected LinkedIterator(final LinkedNode start) {
-+ /* setup nextElement and next */
-+ for (LinkedNode curr = start;;) {
-+ final LinkedNode next = curr.getNextVolatile();
-+
-+ final E element = curr.getElementPlain();
-+
-+ if (element != null) {
-+ this.nextElement = element;
-+ this.next = curr;
-+ break;
-+ }
-+
-+ if (next == null || next == curr) {
-+ break;
-+ }
-+ curr = next;
-+ }
-+ }
-+
-+ protected final void findNext() {
-+ /* only called if this.nextElement != null, which means this.next != null */
-+ for (LinkedNode curr = this.next;;) {
-+ final LinkedNode next = curr.getNextVolatile();
-+
-+ if (next == null || next == curr) {
-+ break;
-+ }
-+
-+ final E element = next.getElementPlain();
-+
-+ if (element != null) {
-+ this.nextElement = element;
-+ this.curr = this.next; /* this.next will be the value returned from next(), set this.curr for remove() */
-+ this.next = next;
-+ return;
-+ }
-+ curr = next;
-+ }
-+
-+ /* out of nodes to iterate */
-+ /* keep curr for remove() calls */
-+ this.next = null;
-+ this.nextElement = null;
-+ }
-+
-+ /**
-+ * {@inheritDoc}
-+ */
-+ @Override
-+ public boolean hasNext() {
-+ return this.nextElement != null;
-+ }
-+
-+ /**
-+ * {@inheritDoc}
-+ */
-+ @Override
-+ public E next() {
-+ final E element = this.nextElement;
-+
-+ if (element == null) {
-+ throw new NoSuchElementException();
-+ }
-+
-+ this.findNext();
-+
-+ return element;
-+ }
-+
-+ /**
-+ * {@inheritDoc}
-+ */
-+ @Override
-+ public void remove() {
-+ if (this.curr == null) {
-+ throw new IllegalStateException();
-+ }
-+
-+ this.curr.setElementVolatile(null);
-+ this.curr = null;
-+ }
-+ }
-+}
-diff --git a/src/main/java/ca/spottedleaf/concurrentutil/completable/CallbackCompletable.java b/src/main/java/ca/spottedleaf/concurrentutil/completable/CallbackCompletable.java
-new file mode 100644
-index 0000000000000000000000000000000000000000..6bad6f8ecc0944d2f406924c7de7e227ff1e70fa
---- /dev/null
-+++ b/src/main/java/ca/spottedleaf/concurrentutil/completable/CallbackCompletable.java
-@@ -0,0 +1,110 @@
-+package ca.spottedleaf.concurrentutil.completable;
-+
-+import ca.spottedleaf.concurrentutil.collection.MultiThreadedQueue;
-+import ca.spottedleaf.concurrentutil.executor.Cancellable;
-+import ca.spottedleaf.concurrentutil.util.ConcurrentUtil;
-+import org.slf4j.Logger;
-+import org.slf4j.LoggerFactory;
-+import java.util.function.BiConsumer;
-+
-+public final class CallbackCompletable {
-+
-+ private static final Logger LOGGER = LoggerFactory.getLogger(CallbackCompletable.class);
-+
-+ private final MultiThreadedQueue> waiters = new MultiThreadedQueue<>();
-+ private T result;
-+ private Throwable throwable;
-+ private volatile boolean completed;
-+
-+ public boolean isCompleted() {
-+ return this.completed;
-+ }
-+
-+ /**
-+ * Note: Can only use after calling {@link #addAsynchronousWaiter(BiConsumer)}, as this function performs zero
-+ * synchronisation
-+ */
-+ public T getResult() {
-+ return this.result;
-+ }
-+
-+ /**
-+ * Note: Can only use after calling {@link #addAsynchronousWaiter(BiConsumer)}, as this function performs zero
-+ * synchronisation
-+ */
-+ public Throwable getThrowable() {
-+ return this.throwable;
-+ }
-+
-+ /**
-+ * Adds a waiter that should only be completed asynchronously by the complete() calls. If complete()
-+ * has already been called, returns {@code null} and does not invoke the specified consumer.
-+ * @param consumer Consumer to be executed on completion
-+ * @throws NullPointerException If consumer is null
-+ * @return A cancellable which will control the execution of the specified consumer
-+ */
-+ public Cancellable addAsynchronousWaiter(final BiConsumer consumer) {
-+ if (this.waiters.add(consumer)) {
-+ return new CancellableImpl(consumer);
-+ }
-+ return null;
-+ }
-+
-+ private void completeAllWaiters(final T result, final Throwable throwable) {
-+ this.completed = true;
-+ BiConsumer waiter;
-+ while ((waiter = this.waiters.pollOrBlockAdds()) != null) {
-+ this.completeWaiter(waiter, result, throwable);
-+ }
-+ }
-+
-+ private void completeWaiter(final BiConsumer consumer, final T result, final Throwable throwable) {
-+ try {
-+ consumer.accept(result, throwable);
-+ } catch (final Throwable throwable2) {
-+ LOGGER.error("Failed to complete callback " + ConcurrentUtil.genericToString(consumer), throwable2);
-+ }
-+ }
-+
-+ /**
-+ * Adds a waiter that will be completed asynchronously by the complete() calls. If complete()
-+ * has already been called, then invokes the consumer synchronously with the completed result.
-+ * @param consumer Consumer to be executed on completion
-+ * @throws NullPointerException If consumer is null
-+ * @return A cancellable which will control the execution of the specified consumer
-+ */
-+ public Cancellable addWaiter(final BiConsumer consumer) {
-+ if (this.waiters.add(consumer)) {
-+ return new CancellableImpl(consumer);
-+ }
-+ this.completeWaiter(consumer, this.result, this.throwable);
-+ return new CancellableImpl(consumer);
-+ }
-+
-+ public void complete(final T result) {
-+ this.result = result;
-+ this.completeAllWaiters(result, null);
-+ }
-+
-+ public void completeWithThrowable(final Throwable throwable) {
-+ if (throwable == null) {
-+ throw new NullPointerException("Throwable cannot be null");
-+ }
-+ this.throwable = throwable;
-+ this.completeAllWaiters(null, throwable);
-+ }
-+
-+ private final class CancellableImpl implements Cancellable {
-+
-+ private final BiConsumer waiter;
-+
-+ private CancellableImpl(final BiConsumer waiter) {
-+ this.waiter = waiter;
-+ }
-+
-+ @Override
-+ public boolean cancel() {
-+ return CallbackCompletable.this.waiters.remove(this.waiter);
-+ }
-+ }
-+}
-\ No newline at end of file
-diff --git a/src/main/java/ca/spottedleaf/concurrentutil/completable/Completable.java b/src/main/java/ca/spottedleaf/concurrentutil/completable/Completable.java
-new file mode 100644
-index 0000000000000000000000000000000000000000..365616439fa079017d648ed7f6ddf6950a691adf
---- /dev/null
-+++ b/src/main/java/ca/spottedleaf/concurrentutil/completable/Completable.java
-@@ -0,0 +1,737 @@
-+package ca.spottedleaf.concurrentutil.completable;
-+
-+import ca.spottedleaf.concurrentutil.util.ConcurrentUtil;
-+import ca.spottedleaf.concurrentutil.util.Validate;
-+import org.slf4j.Logger;
-+import org.slf4j.LoggerFactory;
-+import java.lang.invoke.VarHandle;
-+import java.util.concurrent.CompletableFuture;
-+import java.util.concurrent.CompletionException;
-+import java.util.concurrent.CompletionStage;
-+import java.util.concurrent.Executor;
-+import java.util.concurrent.ForkJoinPool;
-+import java.util.concurrent.locks.LockSupport;
-+import java.util.function.BiConsumer;
-+import java.util.function.BiFunction;
-+import java.util.function.Consumer;
-+import java.util.function.Function;
-+import java.util.function.Supplier;
-+
-+public final class Completable {
-+
-+ private static final Logger LOGGER = LoggerFactory.getLogger(Completable.class);
-+ private static final Function DEFAULT_EXCEPTION_HANDLER = (final Throwable thr) -> {
-+ LOGGER.error("Unhandled exception during Completable operation", thr);
-+ return thr;
-+ };
-+
-+ public static Executor getDefaultExecutor() {
-+ return ForkJoinPool.commonPool();
-+ }
-+
-+ private static final Transform COMPLETED_STACK = new Transform<>(null, null, null, null) {
-+ @Override
-+ public void run() {}
-+ };
-+ private volatile Transform completeStack;
-+ private static final VarHandle COMPLETE_STACK_HANDLE = ConcurrentUtil.getVarHandle(Completable.class, "completeStack", Transform.class);
-+
-+ private static final Object NULL_MASK = new Object();
-+ private volatile Object result;
-+ private static final VarHandle RESULT_HANDLE = ConcurrentUtil.getVarHandle(Completable.class, "result", Object.class);
-+
-+ private Object getResultPlain() {
-+ return (Object)RESULT_HANDLE.get(this);
-+ }
-+
-+ private Object getResultVolatile() {
-+ return (Object)RESULT_HANDLE.getVolatile(this);
-+ }
-+
-+ private void pushStackOrRun(final Transform push) {
-+ int failures = 0;
-+ for (Transform curr = (Transform)COMPLETE_STACK_HANDLE.getVolatile(this);;) {
-+ if (curr == COMPLETED_STACK) {
-+ push.execute();
-+ return;
-+ }
-+
-+ push.next = curr;
-+
-+ for (int i = 0; i < failures; ++i) {
-+ ConcurrentUtil.backoff();
-+ }
-+
-+ if (curr == (curr = (Transform)COMPLETE_STACK_HANDLE.compareAndExchange(this, curr, push))) {
-+ return;
-+ }
-+ push.next = null;
-+ ++failures;
-+ }
-+ }
-+
-+ private void propagateStack() {
-+ Transform topStack = (Transform)COMPLETE_STACK_HANDLE.getAndSet(this, COMPLETED_STACK);
-+ while (topStack != null) {
-+ topStack.execute();
-+ topStack = topStack.next;
-+ }
-+ }
-+
-+ private static Object maskNull(final Object res) {
-+ return res == null ? NULL_MASK : res;
-+ }
-+
-+ private static Object unmaskNull(final Object res) {
-+ return res == NULL_MASK ? null : res;
-+ }
-+
-+ private static Executor checkExecutor(final Executor executor) {
-+ return Validate.notNull(executor, "Executor may not be null");
-+ }
-+
-+ public Completable() {}
-+
-+ private Completable(final Object complete) {
-+ COMPLETE_STACK_HANDLE.set(this, COMPLETED_STACK);
-+ RESULT_HANDLE.setRelease(this, complete);
-+ }
-+
-+ public static Completable completed(final T value) {
-+ return new Completable<>(maskNull(value));
-+ }
-+
-+ public static Completable failed(final Throwable ex) {
-+ Validate.notNull(ex, "Exception may not be null");
-+
-+ return new Completable<>(new ExceptionResult(ex));
-+ }
-+
-+ public static Completable supplied(final Supplier supplier) {
-+ return supplied(supplier, DEFAULT_EXCEPTION_HANDLER);
-+ }
-+
-+ public static Completable supplied(final Supplier supplier, final Function exceptionHandler) {
-+ try {
-+ return completed(supplier.get());
-+ } catch (final Throwable throwable) {
-+ Throwable complete;
-+ try {
-+ complete = exceptionHandler.apply(throwable);
-+ } catch (final Throwable thr2) {
-+ throwable.addSuppressed(thr2);
-+ complete = throwable;
-+ }
-+ return failed(complete);
-+ }
-+ }
-+
-+ public static Completable suppliedAsync(final Supplier supplier, final Executor executor) {
-+ return suppliedAsync(supplier, executor, DEFAULT_EXCEPTION_HANDLER);
-+ }
-+
-+ public static Completable suppliedAsync(final Supplier supplier, final Executor executor, final Function exceptionHandler) {
-+ final Completable ret = new Completable<>();
-+
-+ class AsyncSuppliedCompletable implements Runnable, CompletableFuture.AsynchronousCompletionTask {
-+ @Override
-+ public void run() {
-+ try {
-+ ret.complete(supplier.get());
-+ } catch (final Throwable throwable) {
-+ Throwable complete;
-+ try {
-+ complete = exceptionHandler.apply(throwable);
-+ } catch (final Throwable thr2) {
-+ throwable.addSuppressed(thr2);
-+ complete = throwable;
-+ }
-+ ret.completeExceptionally(complete);
-+ }
-+ }
-+ }
-+
-+ try {
-+ executor.execute(new AsyncSuppliedCompletable());
-+ } catch (final Throwable throwable) {
-+ Throwable complete;
-+ try {
-+ complete = exceptionHandler.apply(throwable);
-+ } catch (final Throwable thr2) {
-+ throwable.addSuppressed(thr2);
-+ complete = throwable;
-+ }
-+ ret.completeExceptionally(complete);
-+ }
-+
-+ return ret;
-+ }
-+
-+ private boolean completeRaw(final Object value) {
-+ if ((Object)RESULT_HANDLE.getVolatile(this) != null || !(boolean)RESULT_HANDLE.compareAndSet(this, (Object)null, value)) {
-+ return false;
-+ }
-+
-+ this.propagateStack();
-+ return true;
-+ }
-+
-+ public boolean complete(final T result) {
-+ return this.completeRaw(maskNull(result));
-+ }
-+
-+ public boolean completeExceptionally(final Throwable exception) {
-+ Validate.notNull(exception, "Exception may not be null");
-+
-+ return this.completeRaw(new ExceptionResult(exception));
-+ }
-+
-+ public boolean isDone() {
-+ return this.getResultVolatile() != null;
-+ }
-+
-+ public boolean isNormallyComplete() {
-+ return this.getResultVolatile() != null && !(this.getResultVolatile() instanceof ExceptionResult);
-+ }
-+
-+ public boolean isExceptionallyComplete() {
-+ return this.getResultVolatile() instanceof ExceptionResult;
-+ }
-+
-+ public Throwable getException() {
-+ final Object res = this.getResultVolatile();
-+ if (res == null) {
-+ return null;
-+ }
-+
-+ if (!(res instanceof ExceptionResult exRes)) {
-+ throw new IllegalStateException("Not completed exceptionally");
-+ }
-+
-+ return exRes.ex;
-+ }
-+
-+ public T getNow(final T dfl) throws CompletionException {
-+ final Object res = this.getResultVolatile();
-+ if (res == null) {
-+ return dfl;
-+ }
-+
-+ if (res instanceof ExceptionResult exRes) {
-+ throw new CompletionException(exRes.ex);
-+ }
-+
-+ return (T)unmaskNull(res);
-+ }
-+
-+ public T join() throws CompletionException {
-+ if (this.isDone()) {
-+ return this.getNow(null);
-+ }
-+
-+ final UnparkTransform unparkTransform = new UnparkTransform<>(this, Thread.currentThread());
-+
-+ this.pushStackOrRun(unparkTransform);
-+
-+ boolean interuptted = false;
-+ while (!unparkTransform.isReleasable()) {
-+ try {
-+ ForkJoinPool.managedBlock(unparkTransform);
-+ } catch (final InterruptedException ex) {
-+ interuptted = true;
-+ }
-+ }
-+
-+ if (interuptted) {
-+ Thread.currentThread().interrupt();
-+ }
-+
-+ return this.getNow(null);
-+ }
-+
-+ public CompletableFuture toFuture() {
-+ final Object rawResult = this.getResultVolatile();
-+ if (rawResult != null) {
-+ if (rawResult instanceof ExceptionResult exRes) {
-+ return CompletableFuture.failedFuture(exRes.ex);
-+ } else {
-+ return CompletableFuture.completedFuture((T)unmaskNull(rawResult));
-+ }
-+ }
-+
-+ final CompletableFuture ret = new CompletableFuture<>();
-+
-+ class ToFuture implements BiConsumer {
-+
-+ @Override
-+ public void accept(final T res, final Throwable ex) {
-+ if (ex != null) {
-+ ret.completeExceptionally(ex);
-+ } else {
-+ ret.complete(res);
-+ }
-+ }
-+ }
-+
-+ this.whenComplete(new ToFuture());
-+
-+ return ret;
-+ }
-+
-+ public static Completable fromFuture(final CompletionStage stage) {
-+ final Completable ret = new Completable<>();
-+
-+ class FromFuture implements BiConsumer {
-+ @Override
-+ public void accept(final T res, final Throwable ex) {
-+ if (ex != null) {
-+ ret.completeExceptionally(ex);
-+ } else {
-+ ret.complete(res);
-+ }
-+ }
-+ }
-+
-+ stage.whenComplete(new FromFuture());
-+
-+ return ret;
-+ }
-+
-+
-+ public Completable thenApply(final Function function) {
-+ return this.thenApply(function, DEFAULT_EXCEPTION_HANDLER);
-+ }
-+
-+ public Completable thenApply(final Function function, final Function exceptionHandler) {
-+ Validate.notNull(function, "Function may not be null");
-+ Validate.notNull(exceptionHandler, "Exception handler may not be null");
-+
-+ final Completable ret = new Completable<>();
-+ this.pushStackOrRun(new ApplyTransform<>(null, this, ret, exceptionHandler, function));
-+ return ret;
-+ }
-+
-+ public Completable thenApplyAsync(final Function function) {
-+ return this.thenApplyAsync(function, getDefaultExecutor(), DEFAULT_EXCEPTION_HANDLER);
-+ }
-+
-+ public Completable thenApplyAsync(final Function function, final Executor executor) {
-+ return this.thenApplyAsync(function, executor, DEFAULT_EXCEPTION_HANDLER);
-+ }
-+
-+ public Completable thenApplyAsync(final Function function, final Executor executor, final Function exceptionHandler) {
-+ Validate.notNull(function, "Function may not be null");
-+ Validate.notNull(exceptionHandler, "Exception handler may not be null");
-+
-+ final Completable ret = new Completable<>();
-+ this.pushStackOrRun(new ApplyTransform<>(checkExecutor(executor), this, ret, exceptionHandler, function));
-+ return ret;
-+ }
-+
-+
-+ public Completable thenAccept(final Consumer consumer) {
-+ return this.thenAccept(consumer, DEFAULT_EXCEPTION_HANDLER);
-+ }
-+
-+ public Completable thenAccept(final Consumer consumer, final Function exceptionHandler) {
-+ Validate.notNull(consumer, "Consumer may not be null");
-+ Validate.notNull(exceptionHandler, "Exception handler may not be null");
-+
-+ final Completable ret = new Completable<>();
-+ this.pushStackOrRun(new AcceptTransform<>(null, this, ret, exceptionHandler, consumer));
-+ return ret;
-+ }
-+
-+ public Completable thenAcceptAsync(final Consumer consumer) {
-+ return this.thenAcceptAsync(consumer, getDefaultExecutor(), DEFAULT_EXCEPTION_HANDLER);
-+ }
-+
-+ public Completable thenAcceptAsync(final Consumer consumer, final Executor executor) {
-+ return this.thenAcceptAsync(consumer, executor, DEFAULT_EXCEPTION_HANDLER);
-+ }
-+
-+ public Completable thenAcceptAsync(final Consumer consumer, final Executor executor, final Function exceptionHandler) {
-+ Validate.notNull(consumer, "Consumer may not be null");
-+ Validate.notNull(exceptionHandler, "Exception handler may not be null");
-+
-+ final Completable ret = new Completable<>();
-+ this.pushStackOrRun(new AcceptTransform<>(checkExecutor(executor), this, ret, exceptionHandler, consumer));
-+ return ret;
-+ }
-+
-+
-+ public Completable thenRun(final Runnable run) {
-+ return this.thenRun(run, DEFAULT_EXCEPTION_HANDLER);
-+ }
-+
-+ public Completable thenRun(final Runnable run, final Function exceptionHandler) {
-+ Validate.notNull(run, "Run may not be null");
-+ Validate.notNull(exceptionHandler, "Exception handler may not be null");
-+
-+ final Completable ret = new Completable<>();
-+ this.pushStackOrRun(new RunTransform<>(null, this, ret, exceptionHandler, run));
-+ return ret;
-+ }
-+
-+ public Completable thenRunAsync(final Runnable run) {
-+ return this.thenRunAsync(run, getDefaultExecutor(), DEFAULT_EXCEPTION_HANDLER);
-+ }
-+
-+ public Completable thenRunAsync(final Runnable run, final Executor executor) {
-+ return this.thenRunAsync(run, executor, DEFAULT_EXCEPTION_HANDLER);
-+ }
-+
-+ public Completable thenRunAsync(final Runnable run, final Executor executor, final Function exceptionHandler) {
-+ Validate.notNull(run, "Run may not be null");
-+ Validate.notNull(exceptionHandler, "Exception handler may not be null");
-+
-+ final Completable ret = new Completable<>();
-+ this.pushStackOrRun(new RunTransform<>(checkExecutor(executor), this, ret, exceptionHandler, run));
-+ return ret;
-+ }
-+
-+
-+ public Completable handle(final BiFunction function) {
-+ return this.handle(function, DEFAULT_EXCEPTION_HANDLER);
-+ }
-+
-+ public Completable handle(final BiFunction function,
-+ final Function exceptionHandler) {
-+ Validate.notNull(function, "Function may not be null");
-+ Validate.notNull(exceptionHandler, "Exception handler may not be null");
-+
-+ final Completable ret = new Completable<>();
-+ this.pushStackOrRun(new HandleTransform<>(null, this, ret, exceptionHandler, function));
-+ return ret;
-+ }
-+
-+ public Completable handleAsync(final BiFunction function) {
-+ return this.handleAsync(function, getDefaultExecutor(), DEFAULT_EXCEPTION_HANDLER);
-+ }
-+
-+ public Completable handleAsync(final BiFunction function,
-+ final Executor executor) {
-+ return this.handleAsync(function, executor, DEFAULT_EXCEPTION_HANDLER);
-+ }
-+
-+ public Completable handleAsync(final BiFunction function,
-+ final Executor executor,
-+ final Function exceptionHandler) {
-+ Validate.notNull(function, "Function may not be null");
-+ Validate.notNull(exceptionHandler, "Exception handler may not be null");
-+
-+ final Completable ret = new Completable<>();
-+ this.pushStackOrRun(new HandleTransform<>(checkExecutor(executor), this, ret, exceptionHandler, function));
-+ return ret;
-+ }
-+
-+
-+ public Completable whenComplete(final BiConsumer consumer) {
-+ return this.whenComplete(consumer, DEFAULT_EXCEPTION_HANDLER);
-+ }
-+
-+ public Completable whenComplete(final BiConsumer consumer, final Function exceptionHandler) {
-+ Validate.notNull(consumer, "Consumer may not be null");
-+ Validate.notNull(exceptionHandler, "Exception handler may not be null");
-+
-+ final Completable ret = new Completable<>();
-+ this.pushStackOrRun(new WhenTransform<>(null, this, ret, exceptionHandler, consumer));
-+ return ret;
-+ }
-+
-+ public Completable whenCompleteAsync(final BiConsumer consumer) {
-+ return this.whenCompleteAsync(consumer, getDefaultExecutor(), DEFAULT_EXCEPTION_HANDLER);
-+ }
-+
-+ public Completable whenCompleteAsync(final BiConsumer consumer, final Executor executor) {
-+ return this.whenCompleteAsync(consumer, executor, DEFAULT_EXCEPTION_HANDLER);
-+ }
-+
-+ public Completable whenCompleteAsync(final BiConsumer consumer, final Executor executor,
-+ final Function exceptionHandler) {
-+ Validate.notNull(consumer, "Consumer may not be null");
-+ Validate.notNull(exceptionHandler, "Exception handler may not be null");
-+
-+ final Completable ret = new Completable<>();
-+ this.pushStackOrRun(new WhenTransform<>(checkExecutor(executor), this, ret, exceptionHandler, consumer));
-+ return ret;
-+ }
-+
-+
-+ public Completable exceptionally(final Function function) {
-+ return this.exceptionally(function, DEFAULT_EXCEPTION_HANDLER);
-+ }
-+
-+ public Completable exceptionally(final Function function, final Function exceptionHandler) {
-+ Validate.notNull(function, "Function may not be null");
-+ Validate.notNull(exceptionHandler, "Exception handler may not be null");
-+
-+ final Completable ret = new Completable<>();
-+ this.pushStackOrRun(new ExceptionallyTransform<>(null, this, ret, exceptionHandler, function));
-+ return ret;
-+ }
-+
-+ public Completable exceptionallyAsync(final Function function) {
-+ return this.exceptionallyAsync(function, getDefaultExecutor(), DEFAULT_EXCEPTION_HANDLER);
-+ }
-+
-+ public Completable exceptionallyAsync(final Function function, final Executor executor) {
-+ return this.exceptionallyAsync(function, executor, DEFAULT_EXCEPTION_HANDLER);
-+ }
-+
-+ public Completable exceptionallyAsync(final Function function, final Executor executor,
-+ final Function exceptionHandler) {
-+ Validate.notNull(function, "Function may not be null");
-+ Validate.notNull(exceptionHandler, "Exception handler may not be null");
-+
-+ final Completable ret = new Completable<>();
-+ this.pushStackOrRun(new ExceptionallyTransform<>(checkExecutor(executor), this, ret, exceptionHandler, function));
-+ return ret;
-+ }
-+
-+ private static final class ExceptionResult {
-+ public final Throwable ex;
-+
-+ public ExceptionResult(final Throwable ex) {
-+ this.ex = ex;
-+ }
-+ }
-+
-+ private static abstract class Transform implements Runnable, CompletableFuture.AsynchronousCompletionTask {
-+
-+ private Transform next;
-+
-+ private final Executor executor;
-+ protected final Completable from;
-+ protected final Completable to;
-+ protected final Function exceptionHandler;
-+
-+ protected Transform(final Executor executor, final Completable from, final Completable to,
-+ final Function exceptionHandler) {
-+ this.executor = executor;
-+ this.from = from;
-+ this.to = to;
-+ this.exceptionHandler = exceptionHandler;
-+ }
-+
-+ // force interface call to become virtual call
-+ @Override
-+ public abstract void run();
-+
-+ protected void failed(final Throwable throwable) {
-+ Throwable complete;
-+ try {
-+ complete = this.exceptionHandler.apply(throwable);
-+ } catch (final Throwable thr2) {
-+ throwable.addSuppressed(thr2);
-+ complete = throwable;
-+ }
-+ this.to.completeExceptionally(complete);
-+ }
-+
-+ public void execute() {
-+ if (this.executor == null) {
-+ this.run();
-+ return;
-+ }
-+
-+ try {
-+ this.executor.execute(this);
-+ } catch (final Throwable throwable) {
-+ this.failed(throwable);
-+ }
-+ }
-+ }
-+
-+ private static final class ApplyTransform extends Transform {
-+
-+ private final Function function;
-+
-+ public ApplyTransform(final Executor executor, final Completable from, final Completable to,
-+ final Function exceptionHandler,
-+ final Function function) {
-+ super(executor, from, to, exceptionHandler);
-+ this.function = function;
-+ }
-+
-+ @Override
-+ public void run() {
-+ final Object result = this.from.getResultPlain();
-+ try {
-+ if (result instanceof ExceptionResult exRes) {
-+ this.to.completeExceptionally(exRes.ex);
-+ } else {
-+ this.to.complete(this.function.apply((T)unmaskNull(result)));
-+ }
-+ } catch (final Throwable throwable) {
-+ this.failed(throwable);
-+ }
-+ }
-+ }
-+
-+ private static final class AcceptTransform extends Transform {
-+ private final Consumer consumer;
-+
-+ public AcceptTransform(final Executor executor, final Completable from, final Completable to,
-+ final Function exceptionHandler,
-+ final Consumer consumer) {
-+ super(executor, from, to, exceptionHandler);
-+ this.consumer = consumer;
-+ }
-+
-+ @Override
-+ public void run() {
-+ final Object result = this.from.getResultPlain();
-+ try {
-+ if (result instanceof ExceptionResult exRes) {
-+ this.to.completeExceptionally(exRes.ex);
-+ } else {
-+ this.consumer.accept((T)unmaskNull(result));
-+ this.to.complete(null);
-+ }
-+ } catch (final Throwable throwable) {
-+ this.failed(throwable);
-+ }
-+ }
-+ }
-+
-+ private static final class RunTransform extends Transform {
-+ private final Runnable run;
-+
-+ public RunTransform(final Executor executor, final Completable from, final Completable to,
-+ final Function exceptionHandler,
-+ final Runnable run) {
-+ super(executor, from, to, exceptionHandler);
-+ this.run = run;
-+ }
-+
-+ @Override
-+ public void run() {
-+ final Object result = this.from.getResultPlain();
-+ try {
-+ if (result instanceof ExceptionResult exRes) {
-+ this.to.completeExceptionally(exRes.ex);
-+ } else {
-+ this.run.run();
-+ this.to.complete(null);
-+ }
-+ } catch (final Throwable throwable) {
-+ this.failed(throwable);
-+ }
-+ }
-+ }
-+
-+ private static final class HandleTransform extends Transform {
-+
-+ private final BiFunction function;
-+
-+ public HandleTransform(final Executor executor, final Completable from, final Completable to,
-+ final Function exceptionHandler,
-+ final BiFunction function) {
-+ super(executor, from, to, exceptionHandler);
-+ this.function = function;
-+ }
-+
-+ @Override
-+ public void run() {
-+ final Object result = this.from.getResultPlain();
-+ try {
-+ if (result instanceof ExceptionResult exRes) {
-+ this.to.complete(this.function.apply(null, exRes.ex));
-+ } else {
-+ this.to.complete(this.function.apply((T)unmaskNull(result), null));
-+ }
-+ } catch (final Throwable throwable) {
-+ this.failed(throwable);
-+ }
-+ }
-+ }
-+
-+ private static final class WhenTransform extends Transform {
-+
-+ private final BiConsumer consumer;
-+
-+ public WhenTransform(final Executor executor, final Completable from, final Completable to,
-+ final Function exceptionHandler,
-+ final BiConsumer consumer) {
-+ super(executor, from, to, exceptionHandler);
-+ this.consumer = consumer;
-+ }
-+
-+ @Override
-+ public void run() {
-+ final Object result = this.from.getResultPlain();
-+ try {
-+ if (result instanceof ExceptionResult exRes) {
-+ this.consumer.accept(null, exRes.ex);
-+ this.to.completeExceptionally(exRes.ex);
-+ } else {
-+ final T unmasked = (T)unmaskNull(result);
-+ this.consumer.accept(unmasked, null);
-+ this.to.complete(unmasked);
-+ }
-+ } catch (final Throwable throwable) {
-+ this.failed(throwable);
-+ }
-+ }
-+ }
-+
-+ private static final class ExceptionallyTransform extends Transform {
-+ private final Function function;
-+
-+ public ExceptionallyTransform(final Executor executor, final Completable from, final Completable to,
-+ final Function exceptionHandler,
-+ final Function function) {
-+ super(executor, from, to, exceptionHandler);
-+ this.function = function;
-+ }
-+
-+ @Override
-+ public void run() {
-+ final Object result = this.from.getResultPlain();
-+ try {
-+ if (result instanceof ExceptionResult exRes) {
-+ this.to.complete(this.function.apply(exRes.ex));
-+ } else {
-+ this.to.complete((T)unmaskNull(result));
-+ }
-+ } catch (final Throwable throwable) {
-+ this.failed(throwable);
-+ }
-+ }
-+ }
-+
-+ private static final class UnparkTransform extends Transform implements ForkJoinPool.ManagedBlocker {
-+
-+ private volatile Thread thread;
-+
-+ public UnparkTransform(final Completable