+++ /dev/null
-/*
- * Copyright 2017-present Open Networking Foundation
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package io.atomix.utils.concurrent;
-
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.CompletionStage;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.Executor;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-import java.util.function.BiConsumer;
-import java.util.function.BiFunction;
-import java.util.function.Consumer;
-import java.util.function.Function;
-
-/**
- * A {@link CompletableFuture} that tracks whether the future or one of its descendants has been blocked on
- * a {@link CompletableFuture#get()} or {@link CompletableFuture#join()} call.
- */
-public class AtomixFuture<T> extends CompletableFuture<T> {
-
- /**
- * Wraps the given future in a new blockable future.
- *
- * @param future the future to wrap
- * @param <T> the future value type
- * @return a new blockable future
- */
- public static <T> AtomixFuture<T> wrap(CompletableFuture<T> future) {
- AtomixFuture<T> newFuture = new AtomixFuture<>();
- future.whenComplete((result, error) -> {
- if (error == null) {
- newFuture.complete(result);
- } else {
- newFuture.completeExceptionally(error);
- }
- });
- return newFuture;
- }
-
- /**
- * Returns a new completed Atomix future.
- *
- * @param result the future result
- * @param <T> the future result type
- * @return the completed future
- */
- public static <T> CompletableFuture<T> completedFuture(T result) {
- CompletableFuture<T> future = new AtomixFuture<>();
- future.complete(result);
- return future;
- }
-
- /**
- * Returns a new exceptionally completed Atomix future.
- *
- * @param t the future exception
- * @param <T> the future result type
- * @return the completed future
- */
- public static <T> CompletableFuture<T> exceptionalFuture(Throwable t) {
- CompletableFuture<T> future = new AtomixFuture<>();
- future.completeExceptionally(t);
- return future;
- }
-
- private static final ThreadContext NULL_CONTEXT = new NullThreadContext();
-
- private ThreadContext getThreadContext() {
- ThreadContext context = ThreadContext.currentContext();
- return context != null ? context : NULL_CONTEXT;
- }
-
- @Override
- public T get() throws InterruptedException, ExecutionException {
- ThreadContext context = getThreadContext();
- context.block();
- try {
- return super.get();
- } finally {
- context.unblock();
- }
- }
-
- @Override
- public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
- ThreadContext context = getThreadContext();
- context.block();
- try {
- return super.get(timeout, unit);
- } finally {
- context.unblock();
- }
- }
-
- @Override
- public synchronized T join() {
- ThreadContext context = getThreadContext();
- context.block();
- try {
- return super.join();
- } finally {
- context.unblock();
- }
- }
-
- @Override
- public <U> CompletableFuture<U> thenApply(Function<? super T, ? extends U> fn) {
- return wrap(super.thenApply(fn));
- }
-
- @Override
- public <U> CompletableFuture<U> thenApplyAsync(Function<? super T, ? extends U> fn) {
- return wrap(super.thenApplyAsync(fn));
- }
-
- @Override
- public <U> CompletableFuture<U> thenApplyAsync(Function<? super T, ? extends U> fn, Executor executor) {
- return wrap(super.thenApplyAsync(fn, executor));
- }
-
- @Override
- public CompletableFuture<Void> thenAccept(Consumer<? super T> action) {
- return wrap(super.thenAccept(action));
- }
-
- @Override
- public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action) {
- return wrap(super.thenAcceptAsync(action));
- }
-
- @Override
- public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action, Executor executor) {
- return wrap(super.thenAcceptAsync(action, executor));
- }
-
- @Override
- public CompletableFuture<Void> thenRun(Runnable action) {
- return wrap(super.thenRun(action));
- }
-
- @Override
- public CompletableFuture<Void> thenRunAsync(Runnable action) {
- return wrap(super.thenRunAsync(action));
- }
-
- @Override
- public CompletableFuture<Void> thenRunAsync(Runnable action, Executor executor) {
- return wrap(super.thenRunAsync(action, executor));
- }
-
- @Override
- public <U, V> CompletableFuture<V> thenCombine(
- CompletionStage<? extends U> other, BiFunction<? super T, ? super U, ? extends V> fn) {
- return wrap(super.thenCombine(other, fn));
- }
-
- @Override
- public <U, V> CompletableFuture<V> thenCombineAsync(
- CompletionStage<? extends U> other, BiFunction<? super T, ? super U, ? extends V> fn) {
- return wrap(super.thenCombineAsync(other, fn));
- }
-
- @Override
- public <U, V> CompletableFuture<V> thenCombineAsync(
- CompletionStage<? extends U> other, BiFunction<? super T, ? super U, ? extends V> fn, Executor executor) {
- return wrap(super.thenCombineAsync(other, fn, executor));
- }
-
- @Override
- public <U> CompletableFuture<Void> thenAcceptBoth(
- CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action) {
- return wrap(super.thenAcceptBoth(other, action));
- }
-
- @Override
- public <U> CompletableFuture<Void> thenAcceptBothAsync(
- CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action) {
- return wrap(super.thenAcceptBothAsync(other, action));
- }
-
- @Override
- public <U> CompletableFuture<Void> thenAcceptBothAsync(
- CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action, Executor executor) {
- return wrap(super.thenAcceptBothAsync(other, action, executor));
- }
-
- @Override
- public CompletableFuture<Void> runAfterBoth(CompletionStage<?> other, Runnable action) {
- return wrap(super.runAfterBoth(other, action));
- }
-
- @Override
- public CompletableFuture<Void> runAfterBothAsync(CompletionStage<?> other, Runnable action) {
- return wrap(super.runAfterBothAsync(other, action));
- }
-
- @Override
- public CompletableFuture<Void> runAfterBothAsync(CompletionStage<?> other, Runnable action, Executor executor) {
- return wrap(super.runAfterBothAsync(other, action, executor));
- }
-
- @Override
- public <U> CompletableFuture<U> applyToEither(CompletionStage<? extends T> other, Function<? super T, U> fn) {
- return wrap(super.applyToEither(other, fn));
- }
-
- @Override
- public <U> CompletableFuture<U> applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T, U> fn) {
- return wrap(super.applyToEitherAsync(other, fn));
- }
-
- @Override
- public <U> CompletableFuture<U> applyToEitherAsync(
- CompletionStage<? extends T> other, Function<? super T, U> fn, Executor executor) {
- return wrap(super.applyToEitherAsync(other, fn, executor));
- }
-
- @Override
- public CompletableFuture<Void> acceptEither(CompletionStage<? extends T> other, Consumer<? super T> action) {
- return wrap(super.acceptEither(other, action));
- }
-
- @Override
- public CompletableFuture<Void> acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action) {
- return wrap(super.acceptEitherAsync(other, action));
- }
-
- @Override
- public CompletableFuture<Void> acceptEitherAsync(
- CompletionStage<? extends T> other, Consumer<? super T> action, Executor executor) {
- return wrap(super.acceptEitherAsync(other, action, executor));
- }
-
- @Override
- public CompletableFuture<Void> runAfterEither(CompletionStage<?> other, Runnable action) {
- return wrap(super.runAfterEither(other, action));
- }
-
- @Override
- public CompletableFuture<Void> runAfterEitherAsync(CompletionStage<?> other, Runnable action) {
- return wrap(super.runAfterEitherAsync(other, action));
- }
-
- @Override
- public CompletableFuture<Void> runAfterEitherAsync(CompletionStage<?> other, Runnable action, Executor executor) {
- return wrap(super.runAfterEitherAsync(other, action, executor));
- }
-
- @Override
- public <U> CompletableFuture<U> thenCompose(Function<? super T, ? extends CompletionStage<U>> fn) {
- return wrap(super.thenCompose(fn));
- }
-
- @Override
- public <U> CompletableFuture<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn) {
- return wrap(super.thenComposeAsync(fn));
- }
-
- @Override
- public <U> CompletableFuture<U> thenComposeAsync(
- Function<? super T, ? extends CompletionStage<U>> fn, Executor executor) {
- return wrap(super.thenComposeAsync(fn, executor));
- }
-
- @Override
- public CompletableFuture<T> whenComplete(BiConsumer<? super T, ? super Throwable> action) {
- return wrap(super.whenComplete(action));
- }
-
- @Override
- public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action) {
- return wrap(super.whenCompleteAsync(action));
- }
-
- @Override
- public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action, Executor executor) {
- return wrap(super.whenCompleteAsync(action, executor));
- }
-
- @Override
- public <U> CompletableFuture<U> handle(BiFunction<? super T, Throwable, ? extends U> fn) {
- return wrap(super.handle(fn));
- }
-
- @Override
- public <U> CompletableFuture<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn) {
- return wrap(super.handleAsync(fn));
- }
-
- @Override
- public <U> CompletableFuture<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn, Executor executor) {
- return wrap(super.handleAsync(fn, executor));
- }
-}
+++ /dev/null
-/*
- * Copyright 2015-present Open Networking Foundation
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package io.atomix.utils.concurrent;
-
-import java.util.List;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.Executor;
-import java.util.concurrent.Future;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-import java.util.function.BinaryOperator;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
-
-/**
- * Utilities for creating completed and exceptional futures.
- *
- * @author <a href="http://github.com/kuujo">Jordan Halterman</a>
- */
-public final class Futures {
-
- /**
- * Gets a future result with a default timeout.
- *
- * @param future the future to block
- * @param <T> the future result type
- * @return the future result
- * @throws RuntimeException if a future exception occurs
- */
- public static <T> T get(Future<T> future) {
- return get(future, 30, TimeUnit.SECONDS);
- }
-
- /**
- * Gets a future result with a default timeout.
- *
- * @param future the future to block
- * @param timeout the future timeout
- * @param timeUnit the future timeout time unit
- * @param <T> the future result type
- * @return the future result
- * @throws RuntimeException if a future exception occurs
- */
- public static <T> T get(Future<T> future, long timeout, TimeUnit timeUnit) {
- try {
- return future.get(timeout, timeUnit);
- } catch (InterruptedException | ExecutionException | TimeoutException e) {
- throw new RuntimeException(e);
- }
- }
-
- /**
- * Creates a future that is synchronously completed.
- *
- * @param result The future result.
- * @return The completed future.
- */
- public static <T> CompletableFuture<T> completedFuture(T result) {
- return CompletableFuture.completedFuture(result);
- }
-
- /**
- * Creates a future that is asynchronously completed.
- *
- * @param result The future result.
- * @param executor The executor on which to complete the future.
- * @return The completed future.
- */
- public static <T> CompletableFuture<T> completedFutureAsync(T result, Executor executor) {
- CompletableFuture<T> future = new CompletableFuture<>();
- executor.execute(() -> future.complete(result));
- return future;
- }
-
- /**
- * Creates a future that is synchronously completed exceptionally.
- *
- * @param t The future exception.
- * @return The exceptionally completed future.
- */
- public static <T> CompletableFuture<T> exceptionalFuture(Throwable t) {
- CompletableFuture<T> future = new CompletableFuture<>();
- future.completeExceptionally(t);
- return future;
- }
-
- /**
- * Creates a future that is asynchronously completed exceptionally.
- *
- * @param t The future exception.
- * @param executor The executor on which to complete the future.
- * @return The exceptionally completed future.
- */
- public static <T> CompletableFuture<T> exceptionalFutureAsync(Throwable t, Executor executor) {
- CompletableFuture<T> future = new CompletableFuture<>();
- executor.execute(() -> {
- future.completeExceptionally(t);
- });
- return future;
- }
-
- /**
- * Returns a future that completes callbacks in add order.
- *
- * @param <T> future value type
- * @return a new completable future that will complete added callbacks in the order in which they were added
- */
- public static <T> CompletableFuture<T> orderedFuture() {
- return new OrderedFuture<>();
- }
-
- /**
- * Returns a future that completes callbacks in add order.
- *
- * @param <T> future value type
- * @return a new completable future that will complete added callbacks in the order in which they were added
- */
- public static <T> CompletableFuture<T> orderedFuture(CompletableFuture<T> future) {
- CompletableFuture<T> newFuture = new OrderedFuture<>();
- future.whenComplete((r, e) -> {
- if (e == null) {
- newFuture.complete(r);
- } else {
- newFuture.completeExceptionally(e);
- }
- });
- return newFuture;
- }
-
- /**
- * Returns a wrapped future that will be completed on the given executor.
- *
- * @param future the future to be completed on the given executor
- * @param executor the executor with which to complete the future
- * @param <T> the future value type
- * @return a wrapped future to be completed on the given executor
- */
- public static <T> CompletableFuture<T> asyncFuture(CompletableFuture<T> future, Executor executor) {
- CompletableFuture<T> newFuture = new AtomixFuture<>();
- future.whenComplete((result, error) -> {
- executor.execute(() -> {
- if (error == null) {
- newFuture.complete(result);
- } else {
- newFuture.completeExceptionally(error);
- }
- });
- });
- return newFuture;
- }
-
- /**
- * Returns a new CompletableFuture completed with a list of computed values
- * when all of the given CompletableFuture complete.
- *
- * @param futures the CompletableFutures
- * @param <T> value type of CompletableFuture
- * @return a new CompletableFuture that is completed when all of the given CompletableFutures complete
- */
- @SuppressWarnings("unchecked")
- public static <T> CompletableFuture<Stream<T>> allOf(Stream<CompletableFuture<T>> futures) {
- CompletableFuture<T>[] futuresArray = futures.toArray(CompletableFuture[]::new);
- return AtomixFuture.wrap(CompletableFuture.allOf(futuresArray)
- .thenApply(v -> Stream.of(futuresArray).map(CompletableFuture::join)));
- }
-
- /**
- * Returns a new CompletableFuture completed with a list of computed values
- * when all of the given CompletableFuture complete.
- *
- * @param futures the CompletableFutures
- * @param <T> value type of CompletableFuture
- * @return a new CompletableFuture that is completed when all of the given CompletableFutures complete
- */
- public static <T> CompletableFuture<List<T>> allOf(List<CompletableFuture<T>> futures) {
- return AtomixFuture.wrap(CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()]))
- .thenApply(v -> futures.stream()
- .map(CompletableFuture::join)
- .collect(Collectors.toList())));
- }
-
- /**
- * Returns a new CompletableFuture completed by reducing a list of computed values
- * when all of the given CompletableFuture complete.
- *
- * @param futures the CompletableFutures
- * @param reducer reducer for computing the result
- * @param emptyValue zero value to be returned if the input future list is empty
- * @param <T> value type of CompletableFuture
- * @return a new CompletableFuture that is completed when all of the given CompletableFutures complete
- */
- public static <T> CompletableFuture<T> allOf(
- List<CompletableFuture<T>> futures, BinaryOperator<T> reducer, T emptyValue) {
- return allOf(futures).thenApply(resultList -> resultList.stream().reduce(reducer).orElse(emptyValue));
- }
-
-}
+++ /dev/null
-/*
- * Copyright 2017-present Open Networking Foundation
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package io.atomix.utils.concurrent;
-
-import java.util.LinkedList;
-import java.util.Queue;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.CompletionStage;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.Executor;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-import java.util.function.BiConsumer;
-import java.util.function.BiFunction;
-import java.util.function.Consumer;
-import java.util.function.Function;
-
-/**
- * A {@link CompletableFuture} that ensures callbacks are called in FIFO order.
- * <p>
- * The default {@link CompletableFuture} does not guarantee the ordering of callbacks, and indeed appears to
- * execute them in LIFO order.
- */
-public class OrderedFuture<T> extends CompletableFuture<T> {
-
- /**
- * Wraps the given future in a new blockable future.
- *
- * @param future the future to wrap
- * @param <T> the future value type
- * @return a new blockable future
- */
- public static <T> CompletableFuture<T> wrap(CompletableFuture<T> future) {
- CompletableFuture<T> newFuture = new OrderedFuture<>();
- future.whenComplete((result, error) -> {
- if (error == null) {
- newFuture.complete(result);
- } else {
- newFuture.completeExceptionally(error);
- }
- });
- return newFuture;
- }
-
- private static final ThreadContext NULL_CONTEXT = new NullThreadContext();
-
- private final Queue<CompletableFuture<T>> orderedFutures = new LinkedList<>();
- private volatile boolean complete;
- private volatile T result;
- private volatile Throwable error;
-
- public OrderedFuture() {
- super.whenComplete(this::complete);
- }
-
- private ThreadContext getThreadContext() {
- ThreadContext context = ThreadContext.currentContext();
- return context != null ? context : NULL_CONTEXT;
- }
-
- @Override
- public T get() throws InterruptedException, ExecutionException {
- ThreadContext context = getThreadContext();
- context.block();
- try {
- return super.get();
- } finally {
- context.unblock();
- }
- }
-
- @Override
- public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
- ThreadContext context = getThreadContext();
- context.block();
- try {
- return super.get(timeout, unit);
- } finally {
- context.unblock();
- }
- }
-
- @Override
- public synchronized T join() {
- ThreadContext context = getThreadContext();
- context.block();
- try {
- return super.join();
- } finally {
- context.unblock();
- }
- }
-
- /**
- * Adds a new ordered future.
- */
- private CompletableFuture<T> orderedFuture() {
- if (!complete) {
- synchronized (orderedFutures) {
- if (!complete) {
- CompletableFuture<T> future = new CompletableFuture<>();
- orderedFutures.add(future);
- return future;
- }
- }
- }
-
- // Completed
- if (error == null) {
- return CompletableFuture.completedFuture(result);
- } else {
- return Futures.exceptionalFuture(error);
- }
- }
-
- /**
- * Completes futures in FIFO order.
- */
- private void complete(T result, Throwable error) {
- synchronized (orderedFutures) {
- this.result = result;
- this.error = error;
- this.complete = true;
- if (error == null) {
- for (CompletableFuture<T> future : orderedFutures) {
- future.complete(result);
- }
- } else {
- for (CompletableFuture<T> future : orderedFutures) {
- future.completeExceptionally(error);
- }
- }
- orderedFutures.clear();
- }
- }
-
- @Override
- public <U> CompletableFuture<U> thenApply(Function<? super T, ? extends U> fn) {
- return wrap(orderedFuture().thenApply(fn));
- }
-
- @Override
- public <U> CompletableFuture<U> thenApplyAsync(Function<? super T, ? extends U> fn) {
- return wrap(orderedFuture().thenApplyAsync(fn));
- }
-
- @Override
- public <U> CompletableFuture<U> thenApplyAsync(Function<? super T, ? extends U> fn, Executor executor) {
- return wrap(orderedFuture().thenApplyAsync(fn, executor));
- }
-
- @Override
- public CompletableFuture<Void> thenAccept(Consumer<? super T> action) {
- return wrap(orderedFuture().thenAccept(action));
- }
-
- @Override
- public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action) {
- return wrap(orderedFuture().thenAcceptAsync(action));
- }
-
- @Override
- public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action, Executor executor) {
- return wrap(orderedFuture().thenAcceptAsync(action, executor));
- }
-
- @Override
- public CompletableFuture<Void> thenRun(Runnable action) {
- return wrap(orderedFuture().thenRun(action));
- }
-
- @Override
- public CompletableFuture<Void> thenRunAsync(Runnable action) {
- return wrap(orderedFuture().thenRunAsync(action));
- }
-
- @Override
- public CompletableFuture<Void> thenRunAsync(Runnable action, Executor executor) {
- return wrap(orderedFuture().thenRunAsync(action, executor));
- }
-
- @Override
- public <U, V> CompletableFuture<V> thenCombine(CompletionStage<? extends U> other, BiFunction<? super T, ? super U, ? extends V> fn) {
- return wrap(orderedFuture().thenCombine(other, fn));
- }
-
- @Override
- public <U, V> CompletableFuture<V> thenCombineAsync(CompletionStage<? extends U> other, BiFunction<? super T, ? super U, ? extends V> fn) {
- return wrap(orderedFuture().thenCombineAsync(other, fn));
- }
-
- @Override
- public <U, V> CompletableFuture<V> thenCombineAsync(CompletionStage<? extends U> other, BiFunction<? super T, ? super U, ? extends V> fn, Executor executor) {
- return wrap(orderedFuture().thenCombineAsync(other, fn, executor));
- }
-
- @Override
- public <U> CompletableFuture<Void> thenAcceptBoth(CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action) {
- return wrap(orderedFuture().thenAcceptBoth(other, action));
- }
-
- @Override
- public <U> CompletableFuture<Void> thenAcceptBothAsync(CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action) {
- return wrap(orderedFuture().thenAcceptBothAsync(other, action));
- }
-
- @Override
- public <U> CompletableFuture<Void> thenAcceptBothAsync(CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action, Executor executor) {
- return wrap(orderedFuture().thenAcceptBothAsync(other, action, executor));
- }
-
- @Override
- public CompletableFuture<Void> runAfterBoth(CompletionStage<?> other, Runnable action) {
- return wrap(orderedFuture().runAfterBoth(other, action));
- }
-
- @Override
- public CompletableFuture<Void> runAfterBothAsync(CompletionStage<?> other, Runnable action) {
- return wrap(orderedFuture().runAfterBothAsync(other, action));
- }
-
- @Override
- public CompletableFuture<Void> runAfterBothAsync(CompletionStage<?> other, Runnable action, Executor executor) {
- return wrap(orderedFuture().runAfterBothAsync(other, action, executor));
- }
-
- @Override
- public <U> CompletableFuture<U> applyToEither(CompletionStage<? extends T> other, Function<? super T, U> fn) {
- return wrap(orderedFuture().applyToEither(other, fn));
- }
-
- @Override
- public <U> CompletableFuture<U> applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T, U> fn) {
- return wrap(orderedFuture().applyToEitherAsync(other, fn));
- }
-
- @Override
- public <U> CompletableFuture<U> applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T, U> fn, Executor executor) {
- return wrap(orderedFuture().applyToEitherAsync(other, fn, executor));
- }
-
- @Override
- public CompletableFuture<Void> acceptEither(CompletionStage<? extends T> other, Consumer<? super T> action) {
- return wrap(orderedFuture().acceptEither(other, action));
- }
-
- @Override
- public CompletableFuture<Void> acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action) {
- return wrap(orderedFuture().acceptEitherAsync(other, action));
- }
-
- @Override
- public CompletableFuture<Void> acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action, Executor executor) {
- return wrap(orderedFuture().acceptEitherAsync(other, action, executor));
- }
-
- @Override
- public CompletableFuture<Void> runAfterEither(CompletionStage<?> other, Runnable action) {
- return wrap(orderedFuture().runAfterEither(other, action));
- }
-
- @Override
- public CompletableFuture<Void> runAfterEitherAsync(CompletionStage<?> other, Runnable action) {
- return wrap(orderedFuture().runAfterEitherAsync(other, action));
- }
-
- @Override
- public CompletableFuture<Void> runAfterEitherAsync(CompletionStage<?> other, Runnable action, Executor executor) {
- return wrap(orderedFuture().runAfterEitherAsync(other, action, executor));
- }
-
- @Override
- public <U> CompletableFuture<U> thenCompose(Function<? super T, ? extends CompletionStage<U>> fn) {
- return wrap(orderedFuture().thenCompose(fn));
- }
-
- @Override
- public <U> CompletableFuture<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn) {
- return wrap(orderedFuture().thenComposeAsync(fn));
- }
-
- @Override
- public <U> CompletableFuture<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn, Executor executor) {
- return wrap(orderedFuture().thenComposeAsync(fn, executor));
- }
-
- @Override
- public CompletableFuture<T> whenComplete(BiConsumer<? super T, ? super Throwable> action) {
- return wrap(orderedFuture().whenComplete(action));
- }
-
- @Override
- public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action) {
- return wrap(orderedFuture().whenCompleteAsync(action));
- }
-
- @Override
- public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action, Executor executor) {
- return wrap(orderedFuture().whenCompleteAsync(action, executor));
- }
-
- @Override
- public <U> CompletableFuture<U> handle(BiFunction<? super T, Throwable, ? extends U> fn) {
- return wrap(orderedFuture().handle(fn));
- }
-
- @Override
- public <U> CompletableFuture<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn) {
- return wrap(orderedFuture().handleAsync(fn));
- }
-
- @Override
- public <U> CompletableFuture<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn, Executor executor) {
- return wrap(orderedFuture().handleAsync(fn, executor));
- }
-
- @Override
- public CompletableFuture<T> exceptionally(Function<Throwable, ? extends T> fn) {
- return wrap(orderedFuture().exceptionally(fn));
- }
-
- @Override
- public CompletableFuture<T> toCompletableFuture() {
- return this;
- }
-}
+++ /dev/null
-/*
- * Copyright 2017-present Open Networking Foundation
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package io.atomix.utils.concurrent;
-
-import org.junit.Test;
-
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.atomic.AtomicInteger;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.fail;
-
-/**
- * Ordered completable future test.
- */
-public class OrderedFutureTest {
-
- /**
- * Tests ordered completion of future callbacks.
- */
- @Test
- public void testOrderedCompletion() throws Throwable {
- CompletableFuture<String> future = new OrderedFuture<>();
- AtomicInteger order = new AtomicInteger();
- future.whenComplete((r, e) -> assertEquals(1, order.incrementAndGet()));
- future.whenComplete((r, e) -> assertEquals(2, order.incrementAndGet()));
- future.handle((r, e) -> {
- assertEquals(3, order.incrementAndGet());
- assertEquals("foo", r);
- return "bar";
- });
- future.thenRun(() -> assertEquals(3, order.incrementAndGet()));
- future.thenAccept(r -> {
- assertEquals(5, order.incrementAndGet());
- assertEquals("foo", r);
- });
- future.thenApply(r -> {
- assertEquals(6, order.incrementAndGet());
- assertEquals("foo", r);
- return "bar";
- });
- future.whenComplete((r, e) -> {
- assertEquals(7, order.incrementAndGet());
- assertEquals("foo", r);
- });
- future.complete("foo");
- }
-
- /**
- * Tests ordered failure of future callbacks.
- */
- public void testOrderedFailure() throws Throwable {
- CompletableFuture<String> future = new OrderedFuture<>();
- AtomicInteger order = new AtomicInteger();
- future.whenComplete((r, e) -> assertEquals(1, order.incrementAndGet()));
- future.whenComplete((r, e) -> assertEquals(2, order.incrementAndGet()));
- future.handle((r, e) -> {
- assertEquals(3, order.incrementAndGet());
- return "bar";
- });
- future.thenRun(() -> fail());
- future.thenAccept(r -> fail());
- future.exceptionally(e -> {
- assertEquals(3, order.incrementAndGet());
- return "bar";
- });
- future.completeExceptionally(new RuntimeException("foo"));
- }
-
- /**
- * Tests calling callbacks that are added after completion.
- */
- public void testAfterComplete() throws Throwable {
- CompletableFuture<String> future = new OrderedFuture<>();
- future.whenComplete((result, error) -> assertEquals("foo", result));
- future.complete("foo");
- AtomicInteger count = new AtomicInteger();
- future.whenComplete((result, error) -> {
- assertEquals("foo", result);
- assertEquals(1, count.incrementAndGet());
- });
- future.thenAccept(result -> {
- assertEquals("foo", result);
- assertEquals(2, count.incrementAndGet());
- });
- assertEquals(2, count.get());
- }
-}
Code Review / controller.git / commitdiff