diff --git a/driver-core/src/main/com/mongodb/assertions/Assertions.java b/driver-core/src/main/com/mongodb/assertions/Assertions.java index 98100f65b45..ae30c179e85 100644 --- a/driver-core/src/main/com/mongodb/assertions/Assertions.java +++ b/driver-core/src/main/com/mongodb/assertions/Assertions.java @@ -92,7 +92,7 @@ public static Iterable notNullElements(final String name, final Iterable< public static T notNull(final String name, final T value, final SingleResultCallback callback) { if (value == null) { IllegalArgumentException exception = new IllegalArgumentException(name + " can not be null"); - callback.onResult(null, exception); + callback.completeExceptionally(exception); throw exception; } return value; @@ -122,7 +122,7 @@ public static void isTrue(final String name, final boolean condition) { public static void isTrue(final String name, final boolean condition, final SingleResultCallback callback) { if (!condition) { IllegalStateException exception = new IllegalStateException("state should be: " + name); - callback.onResult(null, exception); + callback.completeExceptionally(exception); throw exception; } } diff --git a/driver-core/src/main/com/mongodb/connection/AsyncCompletionHandler.java b/driver-core/src/main/com/mongodb/connection/AsyncCompletionHandler.java index 893c5f0eedf..a286f346427 100644 --- a/driver-core/src/main/com/mongodb/connection/AsyncCompletionHandler.java +++ b/driver-core/src/main/com/mongodb/connection/AsyncCompletionHandler.java @@ -16,6 +16,7 @@ package com.mongodb.connection; +import com.mongodb.internal.async.SingleResultCallback; import com.mongodb.lang.Nullable; /** @@ -38,4 +39,17 @@ public interface AsyncCompletionHandler { * @param t the exception that describes the failure */ void failed(Throwable t); + + /** + * @return this handler as a callback + */ + default SingleResultCallback asCallback() { + return (r, t) -> { + if (t != null) { + failed(t); + } else { + completed(r); + } + }; + } } diff --git a/driver-core/src/main/com/mongodb/internal/async/AsyncConsumer.java b/driver-core/src/main/com/mongodb/internal/async/AsyncConsumer.java new file mode 100644 index 00000000000..93a10c9cd2d --- /dev/null +++ b/driver-core/src/main/com/mongodb/internal/async/AsyncConsumer.java @@ -0,0 +1,26 @@ +/* + * Copyright 2008-present MongoDB, Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.mongodb.internal.async; + +/** + * See {@link AsyncRunnable}. + *

+ * This class is not part of the public API and may be removed or changed at any time + */ +@FunctionalInterface +public interface AsyncConsumer extends AsyncFunction { +} diff --git a/driver-core/src/main/com/mongodb/internal/async/AsyncFunction.java b/driver-core/src/main/com/mongodb/internal/async/AsyncFunction.java new file mode 100644 index 00000000000..5be92558ee0 --- /dev/null +++ b/driver-core/src/main/com/mongodb/internal/async/AsyncFunction.java @@ -0,0 +1,36 @@ +/* + * Copyright 2008-present MongoDB, Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.mongodb.internal.async; + +import com.mongodb.lang.Nullable; + +/** + * See {@link AsyncRunnable} + *

+ * This class is not part of the public API and may be removed or changed at any time + */ +@FunctionalInterface +public interface AsyncFunction { + /** + * This should not be called externally, but should be implemented as a + * lambda. To "finish" an async chain, use one of the "finish" methods. + * + * @param value A {@code @}{@link Nullable} argument of the asynchronous function. + * @param callback the callback + */ + void unsafeFinish(T value, SingleResultCallback callback); +} diff --git a/driver-core/src/main/com/mongodb/internal/async/AsyncRunnable.java b/driver-core/src/main/com/mongodb/internal/async/AsyncRunnable.java new file mode 100644 index 00000000000..fcf8d61387d --- /dev/null +++ b/driver-core/src/main/com/mongodb/internal/async/AsyncRunnable.java @@ -0,0 +1,246 @@ +/* + * Copyright 2008-present MongoDB, Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.mongodb.internal.async; + +import com.mongodb.internal.async.function.RetryState; +import com.mongodb.internal.async.function.RetryingAsyncCallbackSupplier; + +import java.util.function.Predicate; +import java.util.function.Supplier; + +/** + *

See the test code (AsyncFunctionsTest) for API usage. + * + *

This API is used to write "Async" methods. These must exhibit the + * same behaviour as their sync counterparts, except asynchronously, + * and will make use of a {@link SingleResultCallback} parameter. + * + *

This API makes it easy to compare and verify async code against + * corresponding sync code, since the "shape" and ordering of the + * async code matches that of the sync code. For example, given the + * following "sync" method: + * + * 

+ * public T myMethod()
+ *     method1();
+ *     method2();
+ * }
+ * + *

The async counterpart would be: + * + * 

+ * public void myMethodAsync(SingleResultCallback<T> callback)
+ *     beginAsync().thenRun(c -> {
+ *         method1Async(c);
+ *     }).thenRun(c -> {
+ *         method2Async(c);
+ *     }).finish(callback);
+ * }
+ *
+ * + *

The usage of this API is defined in its tests (AsyncFunctionsTest). + * Each test specifies the Async API code that must be used to formally + * replace a particular pattern of sync code. These tests, in a sense, + * define formal rules of replacement. + * + *

Requirements and conventions: + * + *

Each async method SHOULD start with {@link #beginAsync()}, which begins + * a chain of lambdas. Each lambda provides a callback "c" that MUST be passed + * or completed at the lambda's end of execution. The async method's "callback" + * parameter MUST be passed to {@link #finish(SingleResultCallback)}, and MUST + * NOT be used otherwise. + * + *

Consider refactoring corresponding sync code to reduce nesting or to + * otherwise improve clarity, since minor issues will often be amplified in + * the async code. + * + *

Each async lambda MUST invoke its async method with "c", and MUST return + * immediately after invoking that method. It MUST NOT, for example, have + * a catch or finally (including close on try-with-resources) after the + * invocation of the async method. + * + *

In cases where the async method has "mixed" returns (some of which are + * plain sync, some async), the "c" callback MUST be completed on the + * plain sync path, using {@link SingleResultCallback#complete(Object)} or + * {@link SingleResultCallback#complete(SingleResultCallback)}, followed by a + * return or end of method. + * + *

Chains starting with {@link #beginAsync()} correspond roughly to code + * blocks. This includes method bodies, blocks used in if/try/catch/while/etc. + * statements, and places where anonymous code blocks might be used. For + * clarity, such nested/indented chains might be omitted (where possible, + * as demonstrated in tests). + * + *

Plain sync code MAY throw exceptions, and SHOULD NOT attempt to handle + * them asynchronously. The exceptions will be caught and handled by the API. + * + *

All code, including "plain" code (parameter checks) SHOULD be placed + * within the API's async lambdas. This ensures that exceptions are handled, + * and facilitates comparison/review. This excludes code that must be + * "shared", such as lambda and variable declarations. + * + *

For consistency, and ease of comparison/review, async chains SHOULD be + * formatted as in the tests; that is, with line-breaks at the curly-braces of + * lambda bodies, with no linebreak before the "." of any Async API method. + * + *

Code review checklist, for common mistakes: + * + *

    + *
  1. Is everything (that can be) inside the async lambdas?
    2. + *
  2. Is "callback" supplied to "finish"?
    3. + *
  3. In each block and nested block, is that same block's "c" always + * passed/completed at the end of execution?
    4. + *
  4. Is every c.complete followed by a return, to end execution?
    5. + *
  5. Have all sync method calls been converted to async, where needed?
    6. + *
+ * + *

This class is not part of the public API and may be removed or changed + * at any time + */ +@FunctionalInterface +public interface AsyncRunnable extends AsyncSupplier, AsyncConsumer { + + static AsyncRunnable beginAsync() { + return (c) -> c.complete(c); + } + + /** + * Must be invoked at end of async chain + * @param runnable the sync code to invoke (under non-exceptional flow) + * prior to the callback + * @param callback the callback provided by the method the chain is used in + */ + default void thenRunAndFinish(final Runnable runnable, final SingleResultCallback callback) { + this.finish((r, e) -> { + if (e != null) { + callback.completeExceptionally(e); + return; + } + try { + runnable.run(); + } catch (Throwable t) { + callback.completeExceptionally(t); + return; + } + callback.complete(callback); + }); + } + + /** + * See {@link #thenRunAndFinish(Runnable, SingleResultCallback)}, but the runnable + * will always be executed, including on the exceptional path. + * @param runnable the runnable + * @param callback the callback + */ + default void thenAlwaysRunAndFinish(final Runnable runnable, final SingleResultCallback callback) { + this.finish((r, e) -> { + try { + runnable.run(); + } catch (Throwable t) { + if (e != null) { + t.addSuppressed(e); + } + callback.completeExceptionally(t); + return; + } + callback.onResult(r, e); + }); + } + + /** + * @param runnable The async runnable to run after this runnable + * @return the composition of this runnable and the runnable, a runnable + */ + default AsyncRunnable thenRun(final AsyncRunnable runnable) { + return (c) -> { + this.unsafeFinish((r, e) -> { + if (e == null) { + runnable.unsafeFinish(c); + } else { + c.completeExceptionally(e); + } + }); + }; + } + + /** + * @param condition the condition to check + * @param runnable The async runnable to run after this runnable, + * if and only if the condition is met + * @return the composition of this runnable and the runnable, a runnable + */ + default AsyncRunnable thenRunIf(final Supplier condition, final AsyncRunnable runnable) { + return (callback) -> { + this.unsafeFinish((r, e) -> { + if (e != null) { + callback.completeExceptionally(e); + return; + } + boolean matched; + try { + matched = condition.get(); + } catch (Throwable t) { + callback.completeExceptionally(t); + return; + } + if (matched) { + runnable.unsafeFinish(callback); + } else { + callback.complete(callback); + } + }); + }; + } + + /** + * @param supplier The supplier to supply using after this runnable + * @return the composition of this runnable and the supplier, a supplier + * @param The return type of the resulting supplier + */ + default AsyncSupplier thenSupply(final AsyncSupplier supplier) { + return (c) -> { + this.unsafeFinish((r, e) -> { + if (e == null) { + supplier.unsafeFinish(c); + } else { + c.completeExceptionally(e); + } + }); + }; + } + + /** + * @param runnable the runnable to loop + * @param shouldRetry condition under which to retry + * @return the composition of this, and the looping branch + * @see RetryingAsyncCallbackSupplier + */ + default AsyncRunnable thenRunRetryingWhile( + final AsyncRunnable runnable, final Predicate shouldRetry) { + return thenRun(callback -> { + new RetryingAsyncCallbackSupplier( + new RetryState(), + (rs, lastAttemptFailure) -> shouldRetry.test(lastAttemptFailure), + // `finish` is required here instead of `unsafeFinish` + // because only `finish` meets the contract of + // `AsyncCallbackSupplier.get`, which we implement here + cb -> runnable.finish(cb) + ).get(callback); + }); + } +} diff --git a/driver-core/src/main/com/mongodb/internal/async/AsyncSupplier.java b/driver-core/src/main/com/mongodb/internal/async/AsyncSupplier.java new file mode 100644 index 00000000000..b7d24dd3df5 --- /dev/null +++ b/driver-core/src/main/com/mongodb/internal/async/AsyncSupplier.java @@ -0,0 +1,141 @@ +/* + * Copyright 2008-present MongoDB, Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.mongodb.internal.async; + +import com.mongodb.lang.Nullable; + +import java.util.function.Predicate; + + +/** + * See {@link AsyncRunnable} + *

+ * This class is not part of the public API and may be removed or changed at any time + */ +@FunctionalInterface +public interface AsyncSupplier extends AsyncFunction { + /** + * This should not be called externally to this API. It should be + * implemented as a lambda. To "finish" an async chain, use one of + * the "finish" methods. + * + * @see #finish(SingleResultCallback) + */ + void unsafeFinish(SingleResultCallback callback); + + /** + * This is the async variant of a supplier's get method. + * This method must only be used when this AsyncSupplier corresponds + * to a {@link java.util.function.Supplier} (and is therefore being + * used within an async chain method lambda). + * @param callback the callback + */ + default void getAsync(final SingleResultCallback callback) { + finish(callback); + } + + @Override + default void unsafeFinish(@Nullable final Void value, final SingleResultCallback callback) { + unsafeFinish(callback); + } + + /** + * Must be invoked at end of async chain. + * @param callback the callback provided by the method the chain is used in + */ + default void finish(final SingleResultCallback callback) { + final boolean[] callbackInvoked = {false}; + try { + this.unsafeFinish((v, e) -> { + callbackInvoked[0] = true; + callback.onResult(v, e); + }); + } catch (Throwable t) { + if (callbackInvoked[0]) { + throw t; + } else { + callback.completeExceptionally(t); + } + } + } + + /** + * @param function The async function to run after this supplier + * @return the composition of this supplier and the function, a supplier + * @param The return type of the resulting supplier + */ + default AsyncSupplier thenApply(final AsyncFunction function) { + return (c) -> { + this.unsafeFinish((v, e) -> { + if (e == null) { + function.unsafeFinish(v, c); + } else { + c.completeExceptionally(e); + } + }); + }; + } + + + /** + * @param consumer The async consumer to run after this supplier + * @return the composition of this supplier and the consumer, a runnable + */ + default AsyncRunnable thenConsume(final AsyncConsumer consumer) { + return (c) -> { + this.unsafeFinish((v, e) -> { + if (e == null) { + consumer.unsafeFinish(v, c); + } else { + c.completeExceptionally(e); + } + }); + }; + } + + /** + * @param errorCheck A check, comparable to a catch-if/otherwise-rethrow + * @param errorFunction The branch to execute if the error matches + * @return The composition of this, and the conditional branch + */ + default AsyncSupplier onErrorIf( + final Predicate errorCheck, + final AsyncFunction errorFunction) { + // finish is used here instead of unsafeFinish to ensure that + // exceptions thrown from the callback are properly handled + return (callback) -> this.finish((r, e) -> { + if (e == null) { + callback.complete(r); + return; + } + boolean errorMatched; + try { + errorMatched = errorCheck.test(e); + } catch (Throwable t) { + t.addSuppressed(e); + callback.completeExceptionally(t); + return; + } + if (errorMatched) { + errorFunction.unsafeFinish(e, callback); + } else { + callback.completeExceptionally(e); + } + }); + } + +} diff --git a/driver-core/src/main/com/mongodb/internal/async/SingleResultCallback.java b/driver-core/src/main/com/mongodb/internal/async/SingleResultCallback.java index 573c1ba423c..632e453d0c0 100644 --- a/driver-core/src/main/com/mongodb/internal/async/SingleResultCallback.java +++ b/driver-core/src/main/com/mongodb/internal/async/SingleResultCallback.java @@ -16,9 +16,13 @@ package com.mongodb.internal.async; +import com.mongodb.assertions.Assertions; +import com.mongodb.connection.AsyncCompletionHandler; import com.mongodb.internal.async.function.AsyncCallbackFunction; import com.mongodb.lang.Nullable; +import static com.mongodb.assertions.Assertions.assertNotNull; + /** * An interface to describe the completion of an asynchronous function, which may be represented as {@link AsyncCallbackFunction}. * @@ -34,4 +38,36 @@ public interface SingleResultCallback { * @throws Error Never, on the best effort basis. */ void onResult(@Nullable T result, @Nullable Throwable t); + + /** + * @return this callback as a handler + */ + default AsyncCompletionHandler asHandler() { + return new AsyncCompletionHandler() { + @Override + public void completed(@Nullable final T result) { + onResult(result, null); + } + @Override + public void failed(final Throwable t) { + completeExceptionally(t); + } + }; + } + + default void complete(final SingleResultCallback callback) { + // takes a void callback (itself) to help ensure that this method + // is not accidentally used when "complete(T)" should have been used + // instead, since results are not marked nullable. + Assertions.assertTrue(callback == this); + this.onResult(null, null); + } + + default void complete(@Nullable final T result) { + this.onResult(result, null); + } + + default void completeExceptionally(final Throwable t) { + this.onResult(null, assertNotNull(t)); + } } diff --git a/driver-core/src/main/com/mongodb/internal/async/function/AsyncCallbackRunnable.java b/driver-core/src/main/com/mongodb/internal/async/function/AsyncCallbackRunnable.java index 7304a9ef9b5..02fdbdf9699 100644 --- a/driver-core/src/main/com/mongodb/internal/async/function/AsyncCallbackRunnable.java +++ b/driver-core/src/main/com/mongodb/internal/async/function/AsyncCallbackRunnable.java @@ -32,17 +32,4 @@ public interface AsyncCallbackRunnable { */ void run(SingleResultCallback callback); - /** - * Converts this {@link AsyncCallbackSupplier} to {@link AsyncCallbackSupplier}{@code }. - */ - default AsyncCallbackSupplier asSupplier() { - return this::run; - } - - /** - * @see AsyncCallbackSupplier#whenComplete(Runnable) - */ - default AsyncCallbackRunnable whenComplete(final Runnable after) { - return callback -> asSupplier().whenComplete(after).get(callback); - } } diff --git a/driver-core/src/main/com/mongodb/internal/async/function/RetryingAsyncCallbackSupplier.java b/driver-core/src/main/com/mongodb/internal/async/function/RetryingAsyncCallbackSupplier.java index 9ebe02f5aa7..92233a072be 100644 --- a/driver-core/src/main/com/mongodb/internal/async/function/RetryingAsyncCallbackSupplier.java +++ b/driver-core/src/main/com/mongodb/internal/async/function/RetryingAsyncCallbackSupplier.java @@ -84,6 +84,13 @@ public RetryingAsyncCallbackSupplier( this.asyncFunction = asyncFunction; } + public RetryingAsyncCallbackSupplier( + final RetryState state, + final BiPredicate retryPredicate, + final AsyncCallbackSupplier asyncFunction) { + this(state, (previouslyChosenFailure, lastAttemptFailure) -> lastAttemptFailure, retryPredicate, asyncFunction); + } + @Override public void get(final SingleResultCallback callback) { /* `asyncFunction` and `callback` are the only externally provided pieces of code for which we do not need to care about diff --git a/driver-core/src/test/functional/com/mongodb/client/TestListener.java b/driver-core/src/test/functional/com/mongodb/client/TestListener.java new file mode 100644 index 00000000000..6b968f31f1b --- /dev/null +++ b/driver-core/src/test/functional/com/mongodb/client/TestListener.java @@ -0,0 +1,45 @@ +/* + * Copyright 2008-present MongoDB, Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.mongodb.client; + +import com.mongodb.annotations.ThreadSafe; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; + +/** + * A simple listener that consumes string events, which can be checked in tests. + */ +@ThreadSafe +public final class TestListener { + private final List events = Collections.synchronizedList(new ArrayList<>()); + + public void add(final String s) { + events.add(s); + } + + public List getEventStrings() { + synchronized (events) { + return new ArrayList<>(events); + } + } + + public void clear() { + events.clear(); + } +} diff --git a/driver-core/src/test/unit/com/mongodb/internal/async/AsyncFunctionsTest.java b/driver-core/src/test/unit/com/mongodb/internal/async/AsyncFunctionsTest.java new file mode 100644 index 00000000000..b783b3de93b --- /dev/null +++ b/driver-core/src/test/unit/com/mongodb/internal/async/AsyncFunctionsTest.java @@ -0,0 +1,1101 @@ +/* + * Copyright 2008-present MongoDB, Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.mongodb.internal.async; + +import com.mongodb.client.TestListener; +import org.junit.jupiter.api.Test; +import org.opentest4j.AssertionFailedError; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.BiConsumer; +import java.util.function.Consumer; +import java.util.function.Supplier; + +import static com.mongodb.assertions.Assertions.assertNotNull; +import static com.mongodb.internal.async.AsyncRunnable.beginAsync; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assertions.fail; + +final class AsyncFunctionsTest { + private final TestListener listener = new TestListener(); + private final InvocationTracker invocationTracker = new InvocationTracker(); + private boolean isTestingAbruptCompletion = false; + + @Test + void test1Method() { + // the number of expected variations is often: 1 + N methods invoked + // 1 variation with no exceptions, and N per an exception in each method + assertBehavesSameVariations(2, + () -> { + // single sync method invocations... + sync(1); + }, + (callback) -> { + // ...become a single async invocation, wrapped in begin-thenRun/finish: + beginAsync().thenRun(c -> { + async(1, c); + }).finish(callback); + }); + } + + @Test + void test2Methods() { + // tests pairs, converting: plain-sync, sync-plain, sync-sync + // (plain-plain does not need an async chain) + + assertBehavesSameVariations(3, + () -> { + // plain (unaffected) invocations... + plain(1); + sync(2); + }, + (callback) -> { + beginAsync().thenRun(c -> { + // ...are preserved above affected methods + plain(1); + async(2, c); + }).finish(callback); + }); + + assertBehavesSameVariations(3, + () -> { + // when a plain invocation follows an affected method... + sync(1); + plain(2); + }, + (callback) -> { + // ...it is moved to its own block, and must be completed: + beginAsync().thenRun(c -> { + async(1, c); + }).thenRun(c -> { + plain(2); + c.complete(c); + }).finish(callback); + }); + + assertBehavesSameVariations(3, + () -> { + // when an affected method follows an affected method + sync(1); + sync(2); + }, + (callback) -> { + // ...it is moved to its own block + beginAsync().thenRun(c -> { + async(1, c); + }).thenRun(c -> { + async(2, c); + }).finish(callback); + }); + } + + @Test + void test4Methods() { + // tests the sync-sync pair with preceding and ensuing plain methods. + + assertBehavesSameVariations(5, + () -> { + plain(11); + sync(1); + plain(22); + sync(2); + }, + (callback) -> { + beginAsync().thenRun(c -> { + plain(11); + async(1, c); + }).thenRun(c -> { + plain(22); + async(2, c); + }).finish(callback); + }); + + assertBehavesSameVariations(5, + () -> { + sync(1); + plain(11); + sync(2); + plain(22); + }, + (callback) -> { + beginAsync().thenRun(c -> { + async(1, c); + }).thenRun(c -> { + plain(11); + async(2, c); + }).thenRunAndFinish(() ->{ + plain(22); + }, callback); + }); + } + + @Test + void testSupply() { + assertBehavesSameVariations(4, + () -> { + sync(0); + plain(1); + return syncReturns(2); + }, + (callback) -> { + beginAsync().thenRun(c -> { + async(0, c); + }).thenSupply(c -> { + plain(1); + asyncReturns(2, c); + }).finish(callback); + }); + } + + @Test + void testSupplyWithMixedReturns() { + assertBehavesSameVariations(5, + () -> { + if (plainTest(1)) { + return syncReturns(11); + } else { + return plainReturns(22); + } + }, + (callback) -> { + beginAsync().thenSupply(c -> { + if (plainTest(1)) { + asyncReturns(11, c); + } else { + int r = plainReturns(22); + c.complete(r); // corresponds to a return, and + // must be followed by a return or end of method + } + }).finish(callback); + }); + } + + @Test + void testFullChain() { + // tests a chain with: runnable, producer, function, function, consumer + assertBehavesSameVariations(14, + () -> { + plain(90); + sync(0); + plain(91); + sync(1); + plain(92); + int v = syncReturns(2); + plain(93); + v = syncReturns(v + 1); + plain(94); + v = syncReturns(v + 10); + plain(95); + sync(v + 100); + plain(96); + }, + (callback) -> { + beginAsync().thenRun(c -> { + plain(90); + async(0, c); + }).thenRun(c -> { + plain(91); + async(1, c); + }).thenSupply(c -> { + plain(92); + asyncReturns(2, c); + }).thenApply((v, c) -> { + plain(93); + asyncReturns(v + 1, c); + }).thenApply((v, c) -> { + plain(94); + asyncReturns(v + 10, c); + }).thenConsume((v, c) -> { + plain(95); + async(v + 100, c); + }).thenRunAndFinish(() -> { + plain(96); + }, callback); + }); + } + + @Test + void testConditionals() { + assertBehavesSameVariations(5, + () -> { + if (plainTest(1)) { + sync(2); + } else { + sync(3); + } + }, + (callback) -> { + beginAsync().thenRun(c -> { + if (plainTest(1)) { + async(2, c); + } else { + async(3, c); + } + }).finish(callback); + }); + + // 2 : fail on first sync, fail on test + // 3 : true test, sync2, sync3 + // 2 : false test, sync3 + // 7 total + assertBehavesSameVariations(7, + () -> { + sync(0); + if (plainTest(1)) { + sync(2); + } + sync(3); + }, + (callback) -> { + beginAsync().thenRun(c -> { + async(0, c); + }).thenRunIf(() -> plainTest(1), c -> { + async(2, c); + }).thenRun(c -> { + async(3, c); + }).finish(callback); + }); + + // an additional affected method within the "if" branch + assertBehavesSameVariations(8, + () -> { + sync(0); + if (plainTest(1)) { + sync(21); + sync(22); + } + sync(3); + }, + (callback) -> { + beginAsync().thenRun(c -> { + async(0, c); + }).thenRunIf(() -> plainTest(1), + beginAsync().thenRun(c -> { + async(21, c); + }).thenRun((c) -> { + async(22, c); + }) + ).thenRun(c -> { + async(3, c); + }).finish(callback); + }); + } + + @Test + void testMixedConditionalCascade() { + assertBehavesSameVariations(9, + () -> { + boolean test1 = plainTest(1); + if (test1) { + return syncReturns(11); + } + boolean test2 = plainTest(2); + if (test2) { + return 22; + } + int x = syncReturns(33); + plain(x + 100); + return syncReturns(44); + }, + (callback) -> { + beginAsync().thenSupply(c -> { + boolean test1 = plainTest(1); + if (test1) { + asyncReturns(11, c); + return; + } + boolean test2 = plainTest(2); + if (test2) { + c.complete(22); + return; + } + beginAsync().thenSupply(c2 -> { + asyncReturns(33, c2); + }).thenApply((x, c2) -> { + plain(assertNotNull(x) + 100); + asyncReturns(44, c2); + }).finish(c); + }).finish(callback); + }); + } + + @Test + void testPlain() { + // For completeness. This should not be used, since there is no async. + assertBehavesSameVariations(2, + () -> { + plain(1); + }, + (callback) -> { + beginAsync().thenRun(c -> { + plain(1); + c.complete(c); + }).finish(callback); + }); + } + + @Test + void testTryCatch() { + // single method in both try and catch + assertBehavesSameVariations(3, + () -> { + try { + sync(1); + } catch (Throwable t) { + sync(2); + } + }, + (callback) -> { + beginAsync().thenRun(c -> { + async(1, c); + }).onErrorIf(t -> true, (t, c) -> { + async(2, c); + }).finish(callback); + }); + + // mixed sync/plain + assertBehavesSameVariations(3, + () -> { + try { + sync(1); + } catch (Throwable t) { + plain(2); + } + }, + (callback) -> { + beginAsync().thenRun(c -> { + async(1, c); + }).onErrorIf(t -> true, (t, c) -> { + plain(2); + c.complete(c); + }).finish(callback); + }); + + // chain of 2 in try. + // WARNING: "onErrorIf" will consider everything in + // the preceding chain to be part of the try. + // Use nested async chains to define the beginning + // of the "try". + assertBehavesSameVariations(5, + () -> { + try { + sync(1); + sync(2); + } catch (Throwable t) { + sync(9); + } + }, + (callback) -> { + beginAsync().thenRun(c -> { + async(1, c); + }).thenRun(c -> { + async(2, c); + }).onErrorIf(t -> true, (t, c) -> { + async(9, c); + }).finish(callback); + }); + + // chain of 2 in catch + assertBehavesSameVariations(4, + () -> { + try { + sync(1); + } catch (Throwable t) { + sync(8); + sync(9); + } + }, + (callback) -> { + beginAsync().thenRun(c -> { + async(1, c); + }).onErrorIf(t -> true, (t, callback2) -> { + beginAsync().thenRun(c -> { + async(8, c); + }).thenRun(c -> { + async(9, c); + }).finish(callback2); + }).finish(callback); + }); + + // method after the try-catch block + // here, the try-catch must be nested (as a code block) + assertBehavesSameVariations(5, + () -> { + try { + sync(1); + } catch (Throwable t) { + sync(2); + } + sync(3); + }, + (callback) -> { + beginAsync().thenRun(c2 -> { + beginAsync().thenRun(c -> { + async(1, c); + }).onErrorIf(t -> true, (t, c) -> { + async(2, c); + }).finish(c2); + }).thenRun(c -> { + async(3, c); + }).finish(callback); + }); + + // multiple catch blocks + // WARNING: these are not exclusive; if multiple "onErrorIf" blocks + // match, they will all be executed. + assertBehavesSameVariations(5, + () -> { + try { + if (plainTest(1)) { + throw new UnsupportedOperationException("A"); + } else { + throw new IllegalStateException("B"); + } + } catch (UnsupportedOperationException t) { + sync(8); + } catch (IllegalStateException t) { + sync(9); + } + }, + (callback) -> { + beginAsync().thenRun(c -> { + if (plainTest(1)) { + throw new UnsupportedOperationException("A"); + } else { + throw new IllegalStateException("B"); + } + }).onErrorIf(t -> t instanceof UnsupportedOperationException, (t, c) -> { + async(8, c); + }).onErrorIf(t -> t instanceof IllegalStateException, (t, c) -> { + async(9, c); + }).finish(callback); + }); + } + + @Test + void testTryCatchWithVariables() { + // using supply etc. + assertBehavesSameVariations(12, + () -> { + try { + int i = plainTest(0) ? 1 : 2; + i = syncReturns(i + 10); + sync(i + 100); + } catch (Throwable t) { + sync(3); + } + }, + (callback) -> { + beginAsync().thenRun( + beginAsync().thenSupply(c -> { + int i = plainTest(0) ? 1 : 2; + asyncReturns(i + 10, c); + }).thenConsume((i, c) -> { + async(assertNotNull(i) + 100, c); + }) + ).onErrorIf(t -> true, (t, c) -> { + async(3, c); + }).finish(callback); + }); + + // using an externally-declared variable + assertBehavesSameVariations(17, + () -> { + int i = plainTest(0) ? 1 : 2; + try { + i = syncReturns(i + 10); + sync(i + 100); + } catch (Throwable t) { + sync(3); + } + sync(i + 1000); + }, + (callback) -> { + final int[] i = new int[1]; + beginAsync().thenRun(c -> { + i[0] = plainTest(0) ? 1 : 2; + c.complete(c); + }).thenRun(c -> { + beginAsync().thenSupply(c2 -> { + asyncReturns(i[0] + 10, c2); + }).thenConsume((i2, c2) -> { + i[0] = assertNotNull(i2); + async(i2 + 100, c2); + }).onErrorIf(t -> true, (t, c2) -> { + async(3, c2); + }).finish(c); + }).thenRun(c -> { + async(i[0] + 1000, c); + }).finish(callback); + }); + } + + @Test + void testTryCatchWithConditionInCatch() { + assertBehavesSameVariations(12, + () -> { + try { + sync(plainTest(0) ? 1 : 2); + sync(3); + } catch (Throwable t) { + sync(5); + if (t.getMessage().equals("exception-1")) { + throw t; + } else { + throw new RuntimeException("wrapped-" + t.getMessage(), t); + } + } + }, + (callback) -> { + beginAsync().thenRun(c -> { + async(plainTest(0) ? 1 : 2, c); + }).thenRun(c -> { + async(3, c); + }).onErrorIf(t -> true, (t, c) -> { + beginAsync().thenRun(c2 -> { + async(5, c2); + }).thenRun(c2 -> { + if (assertNotNull(t).getMessage().equals("exception-1")) { + throw (RuntimeException) t; + } else { + throw new RuntimeException("wrapped-" + t.getMessage(), t); + } + }).finish(c); + }).finish(callback); + }); + } + + @Test + void testTryCatchTestAndRethrow() { + // thenSupply: + assertBehavesSameVariations(5, + () -> { + try { + return syncReturns(1); + } catch (Exception e) { + if (e.getMessage().equals(plainTest(1) ? "unexpected" : "exception-1")) { + return syncReturns(2); + } else { + throw e; + } + } + }, + (callback) -> { + beginAsync().thenSupply(c -> { + asyncReturns(1, c); + }).onErrorIf(e -> e.getMessage().equals(plainTest(1) ? "unexpected" : "exception-1"), (t, c) -> { + asyncReturns(2, c); + }).finish(callback); + }); + + // thenRun: + assertBehavesSameVariations(5, + () -> { + try { + sync(1); + } catch (Exception e) { + if (e.getMessage().equals(plainTest(1) ? "unexpected" : "exception-1")) { + sync(2); + } else { + throw e; + } + } + }, + (callback) -> { + beginAsync().thenRun(c -> { + async(1, c); + }).onErrorIf(e -> e.getMessage().equals(plainTest(1) ? "unexpected" : "exception-1"), (t, c) -> { + async(2, c); + }).finish(callback); + }); + } + + @Test + void testRetryLoop() { + assertBehavesSameVariations(InvocationTracker.DEPTH_LIMIT * 2 + 1, + () -> { + while (true) { + try { + sync(plainTest(0) ? 1 : 2); + } catch (RuntimeException e) { + if (e.getMessage().equals("exception-1")) { + continue; + } + throw e; + } + break; + } + }, + (callback) -> { + beginAsync().thenRunRetryingWhile( + c -> async(plainTest(0) ? 1 : 2, c), + e -> e.getMessage().equals("exception-1") + ).finish(callback); + }); + } + + @Test + void testFinally() { + // (in try: normal flow + exception + exception) * (in finally: normal + exception) = 6 + assertBehavesSameVariations(6, + () -> { + try { + plain(1); + sync(2); + } finally { + plain(3); + } + }, + (callback) -> { + beginAsync().thenRun(c -> { + plain(1); + async(2, c); + }).thenAlwaysRunAndFinish(() -> { + plain(3); + }, callback); + }); + } + + @Test + void testUsedAsLambda() { + assertBehavesSameVariations(4, + () -> { + Supplier s = () -> syncReturns(9); + sync(0); + plain(1); + return s.get(); + }, + (callback) -> { + AsyncSupplier s = (c) -> asyncReturns(9, c); + beginAsync().thenRun(c -> { + async(0, c); + }).thenSupply((c) -> { + plain(1); + s.getAsync(c); + }).finish(callback); + }); + } + + @Test + void testVariables() { + assertBehavesSameVariations(3, + () -> { + int something; + something = 90; + sync(something); + something = something + 10; + sync(something); + }, + (callback) -> { + // Certain variables may need to be shared; these can be + // declared (but not initialized) outside the async chain. + // Any container works (atomic allowed but not needed) + final int[] something = new int[1]; + beginAsync().thenRun(c -> { + something[0] = 90; + async(something[0], c); + }).thenRun((c) -> { + something[0] = something[0] + 10; + async(something[0], c); + }).finish(callback); + }); + } + + @Test + void testInvalid() { + isTestingAbruptCompletion = false; + invocationTracker.isAsyncStep = true; + assertThrows(IllegalStateException.class, () -> { + beginAsync().thenRun(c -> { + async(3, c); + throw new IllegalStateException("must not cause second callback invocation"); + }).finish((v, e) -> {}); + }); + assertThrows(IllegalStateException.class, () -> { + beginAsync().thenRun(c -> { + async(3, c); + }).finish((v, e) -> { + throw new IllegalStateException("must not cause second callback invocation"); + }); + }); + } + + @Test + void testDerivation() { + // Demonstrates the progression from nested async to the API. + + // Stand-ins for sync-async methods; these "happily" do not throw + // exceptions, to avoid complicating this demo async code. + Consumer happySync = (i) -> { + invocationTracker.getNextOption(1); + listener.add("affected-success-" + i); + }; + BiConsumer> happyAsync = (i, c) -> { + happySync.accept(i); + c.complete(c); + }; + + // Standard nested async, no error handling: + assertBehavesSameVariations(1, + () -> { + happySync.accept(1); + happySync.accept(2); + }, + (callback) -> { + happyAsync.accept(1, (v, e) -> { + happyAsync.accept(2, callback); + }); + }); + + // When both methods are naively extracted, they are out of order: + assertBehavesSameVariations(1, + () -> { + happySync.accept(1); + happySync.accept(2); + }, + (callback) -> { + SingleResultCallback second = (v, e) -> { + happyAsync.accept(2, callback); + }; + SingleResultCallback first = (v, e) -> { + happyAsync.accept(1, second); + }; + first.onResult(null, null); + }); + + // We create an "AsyncRunnable" that takes a callback, which + // decouples any async methods from each other, allowing them + // to be declared in a sync-like order, and without nesting: + assertBehavesSameVariations(1, + () -> { + happySync.accept(1); + happySync.accept(2); + }, + (callback) -> { + AsyncRunnable first = (SingleResultCallback c) -> { + happyAsync.accept(1, c); + }; + AsyncRunnable second = (SingleResultCallback c) -> { + happyAsync.accept(2, c); + }; + // This is a simplified variant of the "then" methods; + // it has no error handling. It takes methods A and B, + // and returns C, which is B(A()). + AsyncRunnable combined = (c) -> { + first.unsafeFinish((r, e) -> { + second.unsafeFinish(c); + }); + }; + combined.unsafeFinish(callback); + }); + + // This combining method is added as a default method on AsyncRunnable, + // and a "finish" method wraps the resulting methods. This also adds + // exception handling and monadic short-circuiting of ensuing methods + // when an exception arises (comparable to how thrown exceptions "skip" + // ensuing code). + assertBehavesSameVariations(3, + () -> { + sync(1); + sync(2); + }, + (callback) -> { + beginAsync().thenRun(c -> { + async(1, c); + }).thenRun(c -> { + async(2, c); + }).finish(callback); + }); + } + + // invoked methods: + + private void plain(final int i) { + int cur = invocationTracker.getNextOption(2); + if (cur == 0) { + listener.add("plain-exception-" + i); + throw new RuntimeException("affected method exception-" + i); + } else { + listener.add("plain-success-" + i); + } + } + + private int plainReturns(final int i) { + int cur = invocationTracker.getNextOption(2); + if (cur == 0) { + listener.add("plain-exception-" + i); + throw new RuntimeException("affected method exception-" + i); + } else { + listener.add("plain-success-" + i); + return i; + } + } + + private boolean plainTest(final int i) { + int cur = invocationTracker.getNextOption(3); + if (cur == 0) { + listener.add("plain-exception-" + i); + throw new RuntimeException("affected method exception-" + i); + } else if (cur == 1) { + listener.add("plain-false-" + i); + return false; + } else { + listener.add("plain-true-" + i); + return true; + } + } + + private void sync(final int i) { + assertFalse(invocationTracker.isAsyncStep); + affected(i); + } + + + private Integer syncReturns(final int i) { + assertFalse(invocationTracker.isAsyncStep); + return affectedReturns(i); + } + + private void async(final int i, final SingleResultCallback callback) { + assertTrue(invocationTracker.isAsyncStep); + if (isTestingAbruptCompletion) { + affected(i); + callback.complete(callback); + + } else { + try { + affected(i); + callback.complete(callback); + } catch (Throwable t) { + callback.onResult(null, t); + } + } + } + + private void asyncReturns(final int i, final SingleResultCallback callback) { + assertTrue(invocationTracker.isAsyncStep); + if (isTestingAbruptCompletion) { + callback.complete(affectedReturns(i)); + } else { + try { + callback.complete(affectedReturns(i)); + } catch (Throwable t) { + callback.onResult(null, t); + } + } + } + + private void affected(final int i) { + int cur = invocationTracker.getNextOption(2); + if (cur == 0) { + listener.add("affected-exception-" + i); + throw new RuntimeException("exception-" + i); + } else { + listener.add("affected-success-" + i); + } + } + + private int affectedReturns(final int i) { + int cur = invocationTracker.getNextOption(2); + if (cur == 0) { + listener.add("affected-exception-" + i); + throw new RuntimeException("exception-" + i); + } else { + listener.add("affected-success-" + i); + return i; + } + } + + // assert methods: + + private void assertBehavesSameVariations(final int expectedVariations, final Runnable sync, + final Consumer> async) { + assertBehavesSameVariations(expectedVariations, + () -> { + sync.run(); + return null; + }, + (c) -> { + async.accept((v, e) -> c.onResult(v, e)); + }); + } + + private void assertBehavesSameVariations(final int expectedVariations, final Supplier sync, + final Consumer> async) { + // run the variation-trying code twice, with direct/indirect exceptions + for (int i = 0; i < 2; i++) { + isTestingAbruptCompletion = i != 0; + + // the variation-trying code: + invocationTracker.reset(); + do { + invocationTracker.startInitialStep(); + assertBehavesSame( + sync, + () -> invocationTracker.startMatchStep(), + async); + } while (invocationTracker.countDown()); + assertEquals(expectedVariations, invocationTracker.getVariationCount(), + "number of variations did not match"); + } + + } + + private void assertBehavesSame(final Supplier sync, final Runnable between, + final Consumer> async) { + + T expectedValue = null; + Throwable expectedException = null; + try { + expectedValue = sync.get(); + } catch (Throwable e) { + expectedException = e; + } + List expectedEvents = listener.getEventStrings(); + + listener.clear(); + between.run(); + + AtomicReference actualValue = new AtomicReference<>(); + AtomicReference actualException = new AtomicReference<>(); + AtomicBoolean wasCalled = new AtomicBoolean(false); + try { + async.accept((v, e) -> { + actualValue.set(v); + actualException.set(e); + if (wasCalled.get()) { + fail(); + } + wasCalled.set(true); + }); + } catch (Throwable e) { + fail("async threw instead of using callback"); + } + + // The following code can be used to debug variations: +// System.out.println("===VARIATION START"); +// System.out.println("sync: " + expectedEvents); +// System.out.println("callback called?: " + wasCalled.get()); +// System.out.println("value -- sync: " + expectedValue + " -- async: " + actualValue.get()); +// System.out.println("excep -- sync: " + expectedException + " -- async: " + actualException.get()); +// System.out.println("exception mode: " + (isTestingAbruptCompletion +// ? "exceptions thrown directly (abrupt completion)" : "exceptions into callbacks")); +// System.out.println("===VARIATION END"); + + // show assertion failures arising in async tests + if (actualException.get() != null && actualException.get() instanceof AssertionFailedError) { + throw (AssertionFailedError) actualException.get(); + } + + assertTrue(wasCalled.get(), "callback should have been called"); + assertEquals(expectedEvents, listener.getEventStrings(), "steps should have matched"); + assertEquals(expectedValue, actualValue.get()); + assertEquals(expectedException == null, actualException.get() == null, + "both or neither should have produced an exception"); + if (expectedException != null) { + assertEquals(expectedException.getMessage(), actualException.get().getMessage()); + assertEquals(expectedException.getClass(), actualException.get().getClass()); + } + + listener.clear(); + } + + /** + * Tracks invocations: allows testing of all variations of a method calls + */ + private static class InvocationTracker { + public static final int DEPTH_LIMIT = 50; + private final List invocationOptionSequence = new ArrayList<>(); + private boolean isAsyncStep; // async = matching, vs initial step = populating + private int currentInvocationIndex; + private int variationCount; + + public void reset() { + variationCount = 0; + } + + public void startInitialStep() { + variationCount++; + isAsyncStep = false; + currentInvocationIndex = -1; + } + + public int getNextOption(final int myOptionsSize) { + /* + This method creates (or gets) the next invocation's option. Each + invoker of this method has the "option" to behave in various ways, + usually just success (option 1) and exceptional failure (option 0), + though some callers might have more options. A sequence of method + outcomes (options) is one "variation". Tests automatically test + all possible variations (up to a limit, to prevent infinite loops). + + Methods generally have labels, to ensure that corresponding + sync/async methods are called in the right order, but these labels + are unrelated to the "variation" logic here. There are two "modes" + (whether completion is abrupt, or not), which are also unrelated. + */ + + currentInvocationIndex++; // which invocation result we are dealing with + + if (currentInvocationIndex >= invocationOptionSequence.size()) { + if (isAsyncStep) { + fail("result should have been pre-initialized: steps may not match"); + } + if (isWithinDepthLimit()) { + invocationOptionSequence.add(myOptionsSize - 1); + } else { + invocationOptionSequence.add(0); // choose "0" option, should always be an exception + } + } + return invocationOptionSequence.get(currentInvocationIndex); + } + + public void startMatchStep() { + isAsyncStep = true; + currentInvocationIndex = -1; + } + + private boolean countDown() { + while (!invocationOptionSequence.isEmpty()) { + int lastItemIndex = invocationOptionSequence.size() - 1; + int lastItem = invocationOptionSequence.get(lastItemIndex); + if (lastItem > 0) { + // count current digit down by 1, until 0 + invocationOptionSequence.set(lastItemIndex, lastItem - 1); + return true; + } else { + // current digit completed, remove (move left) + invocationOptionSequence.remove(lastItemIndex); + } + } + return false; + } + + public int getVariationCount() { + return variationCount; + } + + public boolean isWithinDepthLimit() { + return invocationOptionSequence.size() < DEPTH_LIMIT; + } + } +}