Update async functions

Author: katcharov

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<T> {
      * @param t the exception that describes the failure
      */
     void failed(Throwable t);
+
+    /**
+     * @return this handler as a callback
+     */
+    default SingleResultCallback<T> asCallback() {
+        return (r, t) -> {
+            if (t != null) {
+                failed(t);
+            } else {
+                completed(r);
+            }
+        };
+    }
 }

driver-core/src/main/com/mongodb/internal/async/AsyncFunction.java

@@ -16,8 +16,6 @@
 
 package com.mongodb.internal.async;
 
-import com.mongodb.lang.Nullable;
-
 /**
  * See tests for usage (AsyncFunctionsTest).
  * <p>
@@ -29,5 +27,5 @@ public interface AsyncFunction<T, R> {
      * This should not be called externally, but should be implemented as a
      * lambda. To "finish" an async chain, use one of the "finish" methods.
      */
-    void unsafeFinish(@Nullable T value, SingleResultCallback<R> callback);
+    void unsafeFinish(T value, SingleResultCallback<R> callback);
 }

driver-core/src/main/com/mongodb/internal/async/AsyncSupplier.java

@@ -16,8 +16,6 @@
 
 package com.mongodb.internal.async;
 
-import com.mongodb.lang.Nullable;
-
 import java.util.function.Predicate;
 
 
@@ -38,6 +36,7 @@ public interface AsyncSupplier<T> extends AsyncFunction<Void, T> {
     void unsafeFinish(SingleResultCallback<T> 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).
@@ -48,7 +47,7 @@ default void getAsync(final SingleResultCallback<T> callback) {
     }
 
     @Override
-    default void unsafeFinish(@Nullable final Void value, final SingleResultCallback<T> callback) {
+    default void unsafeFinish(final Void value, final SingleResultCallback<T> callback) {
         unsafeFinish(callback);
     }
 
@@ -108,13 +107,13 @@ default AsyncRunnable thenConsume(final AsyncConsumer<T> consumer) {
 
     /**
      * @param errorCheck A check, comparable to a catch-if/otherwise-rethrow
-     * @param supplier   The branch to execute if the error matches
+     * @param errorFunction   The branch to execute if the error matches
      * @return The composition of this, and the conditional branch
      */
     default AsyncSupplier<T> onErrorIf(
             final Predicate<Throwable> errorCheck,
-            final AsyncSupplier<T> supplier) {
-        return (callback) -> this.unsafeFinish((r, e) -> {
+            final AsyncFunction<Throwable, T> errorFunction) {
+        return (callback) -> this.finish((r, e) -> {
             if (e == null) {
                 callback.onResult(r, null);
                 return;
@@ -128,7 +127,7 @@ default AsyncSupplier<T> onErrorIf(
                 return;
             }
             if (errorMatched) {
-                supplier.unsafeFinish(callback);
+                errorFunction.unsafeFinish(e, callback);
             } else {
                 callback.onResult(null, e);
             }

driver-core/src/main/com/mongodb/internal/async/SingleResultCallback.java

@@ -16,6 +16,8 @@
 
 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;
 
@@ -34,4 +36,32 @@ public interface SingleResultCallback<T> {
      * @throws Error Never, on the best effort basis.
      */
     void onResult(@Nullable T result, @Nullable Throwable t);
+
+    /**
+     * @return this callback as a handler
+     */
+    default AsyncCompletionHandler<T> asHandler() {
+        return new AsyncCompletionHandler<T>() {
+            @Override
+            public void completed(@Nullable final T result) {
+                onResult(result, null);
+            }
+            @Override
+            public void failed(final Throwable t) {
+                onResult(null, t);
+            }
+        };
+    }
+
+    default void complete(final SingleResultCallback<Void> 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(final T result) {
+        this.onResult(result, null);
+    }
 }