Adds AsyncCloseable (Azure#21991)

* Adding AsyncCloseable with codesnippet.

* Implementing AsyncCloseable and deleting AsyncAutoCloseable.

* Add CHANGELOG entry.

* Removing azure-core as an explicit dependency.

* Fix use in AmqpReceiveLinkProcessor.

Author: conniey

sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/AmqpChannelProcessor.java

@@ -6,6 +6,7 @@
 import com.azure.core.amqp.AmqpEndpointState;
 import com.azure.core.amqp.AmqpRetryPolicy;
 import com.azure.core.amqp.exception.AmqpException;
+import com.azure.core.util.AsyncCloseable;
 import com.azure.core.util.logging.ClientLogger;
 import org.reactivestreams.Processor;
 import org.reactivestreams.Subscription;
@@ -298,8 +299,8 @@ private void setAndClearChannel() {
     }
 
     private void close(T channel) {
-        if (channel instanceof AsyncAutoCloseable) {
-            ((AsyncAutoCloseable) channel).closeAsync().subscribe();
+        if (channel instanceof AsyncCloseable) {
+            ((AsyncCloseable) channel).closeAsync().subscribe();
         } else if (channel instanceof AutoCloseable) {
             try {
                 ((AutoCloseable) channel).close();

sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/AsyncAutoCloseable.java

@@ -8,6 +8,7 @@
 import com.azure.core.amqp.AmqpRetryOptions;
 import com.azure.core.amqp.exception.AmqpErrorCondition;
 import com.azure.core.amqp.implementation.handler.ReceiveLinkHandler;
+import com.azure.core.util.AsyncCloseable;
 import com.azure.core.util.logging.ClientLogger;
 import org.apache.qpid.proton.Proton;
 import org.apache.qpid.proton.amqp.Symbol;
@@ -37,7 +38,7 @@
 /**
  * Handles receiving events from Event Hubs service and translating them to proton-j messages.
  */
-public class ReactorReceiver implements AmqpReceiveLink, AsyncAutoCloseable, AutoCloseable {
+public class ReactorReceiver implements AmqpReceiveLink, AsyncCloseable, AutoCloseable {
     private final String entityPath;
     private final Receiver receiver;
     private final ReceiveLinkHandler handler;

sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorReceiver.java

@@ -12,6 +12,7 @@
 import com.azure.core.amqp.exception.AmqpException;
 import com.azure.core.amqp.exception.OperationCancelledException;
 import com.azure.core.amqp.implementation.handler.SendLinkHandler;
+import com.azure.core.util.AsyncCloseable;
 import com.azure.core.util.logging.ClientLogger;
 import org.apache.qpid.proton.Proton;
 import org.apache.qpid.proton.amqp.Binary;
@@ -62,7 +63,7 @@
 /**
  * Handles scheduling and transmitting events through proton-j to Event Hubs service.
  */
-class ReactorSender implements AmqpSendLink, AsyncAutoCloseable, AutoCloseable {
+class ReactorSender implements AmqpSendLink, AsyncCloseable, AutoCloseable {
     private final String entityPath;
     private final Sender sender;
     private final SendLinkHandler handler;

sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorSender.java

@@ -9,6 +9,7 @@
 import com.azure.core.amqp.exception.AmqpErrorContext;
 import com.azure.core.amqp.implementation.handler.ReceiveLinkHandler;
 import com.azure.core.amqp.implementation.handler.SendLinkHandler;
+import com.azure.core.util.AsyncCloseable;
 import com.azure.core.util.logging.ClientLogger;
 import org.apache.qpid.proton.Proton;
 import org.apache.qpid.proton.amqp.UnsignedLong;
@@ -50,7 +51,7 @@
  * Represents a bidirectional link between the message broker and the client. Allows client to send a request to the
  * broker and receive the associated response.
  */
-public class RequestResponseChannel implements AsyncAutoCloseable {
+public class RequestResponseChannel implements AsyncCloseable {
     private final ConcurrentSkipListMap<UnsignedLong, MonoSink<Message>> unconfirmedSends =
         new ConcurrentSkipListMap<>();
     private final AtomicBoolean hasError = new AtomicBoolean();

sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/RequestResponseChannel.java

@@ -2,6 +2,9 @@
 
 ## 1.17.0-beta.1 (Unreleased)
 
+### Features Added
+
+- Added `AsyncCloseable`
 
 ## 1.16.0 (2021-05-07)
 

sdk/core/azure-core/CHANGELOG.md

@@ -0,0 +1,27 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.core.util;
+
+import reactor.core.publisher.Mono;
+
+/**
+ * Interface for close operations that are asynchronous.
+ *
+ * <p><strong>Asynchronously closing a class</strong></p>
+ * <p>In the snippet below, we have a long-lived {@code NetworkResource} class. There are some operations such
+ * as closing {@literal I/O}. Instead of returning a sync {@code close()}, we use {@code closeAsync()} so users'
+ * programs don't block waiting for this operation to complete.</p>
+ *
+ * {@codesnippet com.azure.core.util.AsyncCloseable.closeAsync}
+ */
+public interface AsyncCloseable {
+    /**
+     * Begins the close operation. If one is in progress, will return that existing close operation. If the close
+     * operation is unsuccessful, the Mono completes with an error.
+     *
+     * @return A Mono representing the close operation. If the close operation is unsuccessful, the Mono completes with
+     *     an error.
+     */
+    Mono<Void> closeAsync();
+}

sdk/core/azure-core/src/main/java/com/azure/core/util/AsyncCloseable.java

@@ -0,0 +1,79 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.core.util;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.core.publisher.Sinks;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.IntStream;
+
+/**
+ * Code snippets for {@link AsyncCloseable}.
+ */
+public class AsyncCloseableJavaDocCodeSnippet {
+    public void asyncResource() throws IOException {
+        // BEGIN: com.azure.core.util.AsyncCloseable.closeAsync
+        NetworkResource resource = new NetworkResource();
+        resource.longRunningDownload("https://longdownload.com")
+            .subscribe(
+                byteBuffer -> System.out.println("Buffer received: " + byteBuffer),
+                error -> System.err.printf("Error occurred while downloading: %s%n", error),
+                () -> System.out.println("Completed download operation."));
+
+        System.out.println("Press enter to stop downloading.");
+        System.in.read();
+
+        // We block here because it is the end of the main Program function. A real-life program may chain this
+        // with some other close operations like save download/program state, etc.
+        resource.closeAsync().block();
+        // END: com.azure.core.util.AsyncCloseable.closeAsync
+    }
+
+    /**
+     * A long lived network resource.
+     */
+    static class NetworkResource implements AsyncCloseable {
+        private final AtomicBoolean isClosed = new AtomicBoolean();
+        private final Sinks.Empty<Void> closeMono = Sinks.empty();
+
+        /**
+         * Downloads a resource.
+         *
+         * @param url URL for the download.
+         *
+         * @return A stream of bytes.
+         */
+        Flux<ByteBuffer> longRunningDownload(String url) {
+            final byte[] bytes = url.getBytes(StandardCharsets.UTF_8);
+
+            // Does nothing real but it represents taking from this possibly infinite Flux until
+            // the closeMono emits a signal.
+            return Flux.fromStream(IntStream.range(0, bytes.length)
+                .mapToObj(index -> ByteBuffer.wrap(bytes)))
+                .takeUntilOther(closeMono.asMono());
+        }
+
+        @Override
+        public Mono<Void> closeAsync() {
+            // If the close operation has started, then
+            if (isClosed.getAndSet(true)) {
+                return closeMono.asMono();
+            }
+
+            return startAsyncClose().then(closeMono.asMono());
+        }
+
+        private Mono<Void> startAsyncClose() {
+            return Mono.delay(Duration.ofSeconds(10)).then()
+                .doOnError(error -> closeMono.emitError(error, Sinks.EmitFailureHandler.FAIL_FAST))
+                .doOnSuccess(unused -> closeMono.emitEmpty(Sinks.EmitFailureHandler.FAIL_FAST));
+        }
+    }
+}

sdk/core/azure-core/src/samples/java/com/azure/core/util/AsyncCloseableJavaDocCodeSnippet.java

@@ -37,7 +37,7 @@
     <dependency>
       <groupId>com.azure</groupId>
       <artifactId>azure-core</artifactId>
-      <version>1.16.0</version> <!-- {x-version-update;com.azure:azure-core;dependency} -->
+      <version>1.17.0-beta.1</version> <!-- {x-version-update;unreleased_com.azure:azure-core;dependency} -->
     </dependency>
     <dependency>
       <groupId>com.azure</groupId>

sdk/eventhubs/azure-messaging-eventhubs/pom.xml

@@ -9,7 +9,7 @@
 import com.azure.core.amqp.exception.AmqpException;
 import com.azure.core.amqp.exception.LinkErrorContext;
 import com.azure.core.amqp.implementation.AmqpReceiveLink;
-import com.azure.core.amqp.implementation.AsyncAutoCloseable;
+import com.azure.core.util.AsyncCloseable;
 import com.azure.core.util.logging.ClientLogger;
 import org.apache.qpid.proton.message.Message;
 import org.reactivestreams.Subscription;
@@ -599,8 +599,8 @@ private void disposeReceiver(AmqpReceiveLink link) {
         }
 
         try {
-            if (link instanceof AsyncAutoCloseable) {
-                ((AsyncAutoCloseable) link).closeAsync().subscribe();
+            if (link instanceof AsyncCloseable) {
+                ((AsyncCloseable) link).closeAsync().subscribe();
             } else {
                 link.dispose();
             }