Messaging: v2 stack amqp-core types and integration to Service Bus

.vscode/cspell.json

@@ -865,7 +865,10 @@
     {
       "filename": "sdk/servicebus/azure-messaging-servicebus/**",
       "words": [
+        "anu",
         "Conniey",
+        "liudmila",
+        "Milli",
         "qpid",
         "unretriable"
         ]

eng/code-quality-reports/src/main/resources/checkstyle/checkstyle-suppressions.xml

@@ -146,6 +146,8 @@ the main ServiceBusClientBuilder. -->
   <suppress checks="com.azure.tools.checkstyle.checks.ServiceClientCheck" files="ServiceBusSessionReceiver(Async)?Client.java"/>
 
   <suppress checks="com.azure.tools.checkstyle.checks.UseCaughtExceptionCauseCheck" files="ServiceBusMessageBatch.java"/>
+  <!-- Supress the BufferOverflowException thrown from encodedSize method, this exception is caught, mapped and logged at the callsite  -->
+  <suppress checks="com.azure.tools.checkstyle.checks.ThrowFromClientLoggerCheck" files="com.azure.messaging.servicebus.ServiceBusMessageBatch.java"/>
 
   <!-- Suppress the long package name in autoconfigure.implementation.redis package  -->
   <suppress checks="PackageName" files="com.azure.spring.cloud.autoconfigure.implementation.redis.*" />

eng/code-quality-reports/src/main/resources/spotbugs/spotbugs-exclude.xml

@@ -2436,6 +2436,18 @@
     <Bug pattern="SE_BAD_FIELD"/>
   </Match>
 
+    <!--  The returned Disposable can be ignored since Scheduler is closed when Reactor closes. -->
+  <Match>
+    <Class name="com.azure.core.amqp.implementation.ReactorExecutor"/>
+    <Method name="start"/>
+    <Bug pattern="RV_RETURN_VALUE_IGNORED_NO_SIDE_EFFECT"/>
+  </Match>
+    <Match>
+    <Class name="com.azure.core.amqp.implementation.ReactorExecutor"/>
+    <Method name="run"/>
+    <Bug pattern="RV_RETURN_VALUE_IGNORED_NO_SIDE_EFFECT"/>
+  </Match>
+
   <!-- For BinaryData, copying array contents degrades performance. The data returned as byte array from BinaryData is expected to be copied by the call if mutability is desired. -->
   <Match>
     <Class name="com.azure.core.implementation.util.ByteArrayContent"/>
@@ -2713,5 +2725,115 @@
     <Bug pattern="BC_UNCONFIRMED_CAST_OF_RETURN_VALUE"/>
   </Match>
 
+  <!-- Field is not initialized in Ctr but necessary safety measures are in place when accessing it -->
+  <Match>
+    <Class name="com.azure.messaging.servicebus.NonSessionProcessor"/>
+    <Field name="rollingMessagePump"/>
+    <Bug pattern="UWF_FIELD_NOT_INITIALIZED_IN_CONSTRUCTOR"/>
+  </Match>
+  <Match>
+    <Class name="com.azure.messaging.servicebus.SessionProcessor"/>
+    <Field name="rollingSessionsMessagePump"/>
+    <Bug pattern="UWF_FIELD_NOT_INITIALIZED_IN_CONSTRUCTOR"/>
+  </Match>
+  <Match>
+    <Class name="com.azure.messaging.servicebus.ServiceBusProcessor"/>
+    <Field name="rollingMessagePump"/>
+    <Bug pattern="UWF_FIELD_NOT_INITIALIZED_IN_CONSTRUCTOR"/>
+  </Match>
+  <!-- False positives -->
+  <Match>
+    <Class name="com.azure.messaging.servicebus.NonSessionProcessor$RollingNonSessionMessagePump"/>
+    <Or>
+      <Field name="builder"/>
+      <Field name="logger"/>
+      <Field name="disposable"/>
+    </Or>
+    <Bug pattern="SE_BAD_FIELD"/>
+  </Match>
+  <Match>
+    <Class name="com.azure.messaging.servicebus.SessionProcessor$RollingSessionsMessagePump"/>
+    <Or>
+      <Field name="builder"/>
+      <Field name="logger"/>
+      <Field name="disposable"/>
+    </Or>
+    <Bug pattern="SE_BAD_FIELD"/>
+  </Match>
+  <Match>
+    <Class name="com.azure.messaging.servicebus.ServiceBusProcessor$RollingMessagePump"/>
+    <Or>
+      <Field name="nonSessionBuilder"/>
+      <Field name="sessionBuilder"/>
+      <Field name="logger"/>
+      <Field name="disposable"/>
+    </Or>
+    <Bug pattern="SE_BAD_FIELD"/>
+  </Match>
+  <!-- The RollingNonSessionMessagePump and RollingSessionsMessagePump doesn't need to be serializable -->
+  <Match>
+    <Class
+      name="com.azure.messaging.servicebus.NonSessionProcessor$RollingNonSessionMessagePump"/>
+    <Bug pattern="SE_NO_SERIALVERSIONID"/>
+  </Match>
+  <Match>
+    <Class
+      name="com.azure.messaging.servicebus.SessionProcessor$RollingSessionsMessagePump"/>
+    <Bug pattern="SE_NO_SERIALVERSIONID"/>
+  </Match>
+  <Match>
+    <Class
+      name="com.azure.messaging.servicebus.ServiceBusProcessor$RollingMessagePump"/>
+    <Bug pattern="SE_NO_SERIALVERSIONID"/>
+  </Match>
+  <!-- False positives -->
+  <Match>
+    <Class
+        name="com.azure.messaging.servicebus.SessionsMessagePump$RollingSessionReceiver"/>
+    <Or>
+      <Field name="instrumentation"/>
+      <Field name="logger"/>
+      <Field name="messageFlux"/>
+      <Field name="receiversTracker"/>
+      <Field name="tracer"/>
+      <Field name="managementNode"/>
+      <Field name="serializer"/>
+      <Field name="workerScheduler"/>
+    </Or>
+    <Bug pattern="SE_BAD_FIELD"/>
+  </Match>
+  <Match>
+    <Class
+        name="com.azure.messaging.servicebus.SessionsMessagePump$RollingSessionReceiver$NextSessionStream"/>
+    <Or>
+      <Field name="newSession"/>
+    </Or>
+    <Bug pattern="SE_BAD_FIELD"/>
+  </Match>
+  <Match>
+    <Class
+        name="com.azure.messaging.servicebus.ServiceBusSingleSessionManager$SessionReceiverStream"/>
+    <Or>
+      <Field name="sessionReceiver"/>
+    </Or>
+    <Bug pattern="SE_BAD_FIELD"/>
+  </Match>
+  <!-- The RollingSessionReceiver, NextSessionStream, $SessionReceiverStream doesn't need to be serializable -->
+  <Match>
+    <Class
+        name="com.azure.messaging.servicebus.SessionsMessagePump$RollingSessionReceiver"/>
+    <Bug pattern="SE_NO_SERIALVERSIONID"/>
+  </Match>
+  <Match>
+    <Class
+        name="com.azure.messaging.servicebus.SessionsMessagePump$RollingSessionReceiver$NextSessionStream"/>
+    <Bug pattern="SE_NO_SERIALVERSIONID"/>
+  </Match>
+  <Match>
+    <Class
+        name="com.azure.messaging.servicebus.ServiceBusSingleSessionManager$SessionReceiverStream"/>
+    <Bug pattern="SE_NO_SERIALVERSIONID"/>
+  </Match>
+
   <!-- Comment to force CI to run Spotbugs against all SDKs -->
 </FindBugsFilter>

eng/versioning/version_client.txt

@@ -448,6 +448,9 @@ com.azure.tools:azure-sdk-build-tool;1.0.0;1.1.0-beta.1
 # In the pom, the version update tag after the version should name the unreleased package and the dependency version:
 # <!-- {x-version-update;unreleased_com.azure:azure-core;dependency} -->
 
+unreleased_com.azure:azure-core;1.46.0-beta.1
+unreleased_com.azure:azure-core-amqp;2.9.0-beta.8
+
 # Released Beta dependencies: Copy the entry from above, prepend "beta_", remove the current
 # version and set the version to the released beta. Released beta dependencies are only valid
 # for dependency versions. These entries are specifically for when we've released a beta for

sdk/core/azure-core-amqp/pom.xml

@@ -70,7 +70,7 @@
     <dependency>
       <groupId>com.azure</groupId>
       <artifactId>azure-core</artifactId>
-      <version>1.46.0-beta.1</version> <!-- {x-version-update;com.azure:azure-core;current} -->
+      <version>1.46.0-beta.1</version> <!-- {x-version-update;unreleased_com.azure:azure-core;dependency} -->
     </dependency>
     <dependency>
       <groupId>com.microsoft.azure</groupId>

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

@@ -0,0 +1,75 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.core.amqp.implementation;
+
+import com.azure.core.amqp.AmqpConnection;
+import com.azure.core.amqp.AmqpRetryOptions;
+import com.azure.core.amqp.implementation.handler.ReceiveLinkHandler;
+import com.azure.core.amqp.implementation.handler.ReceiveLinkHandler2;
+import com.azure.core.amqp.implementation.handler.SendLinkHandler;
+import org.apache.qpid.proton.engine.Receiver;
+import org.apache.qpid.proton.engine.Sender;
+import reactor.core.scheduler.Scheduler;
+
+/**
+ * Provides links for send and receive.
+ *
+ * Under normal execution, the provider provides the actual links, but when running under unit test scenarios,
+ * the provider enables tests to inject mock links.
+ * @see ReactorProvider
+ * @see ReactorHandlerProvider
+ * @see AmqpMetricsProvider
+ */
+public class AmqpLinkProvider {
+    /**
+     * Creates an Amqp Link to send messages.
+     *
+     * @param amqpConnection The connection to host the Amqp Link.
+     * @param entityPath The message broker address for the sender.
+     * @param sender The underlying QPID sender.
+     * @param handler The QPID handler associated with the QPID sender.
+     * @param reactorProvider The provider for QPID Reactor and Reactor Executor.
+     * @param tokenManager Token manager for authorising with the CBS node.
+     * @param messageSerializer the Serializer to deserialize and serialize AMQP messages.
+     * @param retryOptions The Retry options.
+     * @param scheduler The scheduler to timeout the send operations those are not acknowledged by the broker.
+     * @param metricsProvider The metric provider (e.g. to record operations such as send).
+     * @return An Amqp Link.
+     */
+    public AmqpSendLink createSendLink(AmqpConnection amqpConnection, String entityPath, Sender sender, SendLinkHandler handler,
+        ReactorProvider reactorProvider, TokenManager tokenManager, MessageSerializer messageSerializer,
+        AmqpRetryOptions retryOptions, Scheduler scheduler, AmqpMetricsProvider metricsProvider) {
+        return new ReactorSender(amqpConnection, entityPath, sender, handler, reactorProvider, tokenManager,
+            messageSerializer, retryOptions, scheduler, metricsProvider);
+    }
+
+    /**
+     * Creates an Amqp Link to receive messages.
+     *
+     * @param amqpConnection The connection to host the Amqp Link.
+     * @param entityPath The message broker address for the receiver.
+     * @param receiver The underlying QPID receiver.
+     * @param handler The QPID handler associated with the QPID receiver.
+     * @param tokenManager Token manager for authorising with the CBS node.
+     * @param dispatcher The dispatcher to schedule work to QPID Reactor Executor.
+     * @param retryOptions The Retry options.
+     * @param metricsProvider The metric provider (e.g. to record operations such as sending flow).
+     * @return An Amqp Link.
+     */
+    public AmqpReceiveLink createReceiveLink(AmqpConnection amqpConnection, String entityPath, Receiver receiver,
+        ReceiveLinkHandler handler, TokenManager tokenManager, ReactorDispatcher dispatcher, AmqpRetryOptions retryOptions,
+        AmqpMetricsProvider metricsProvider) {
+        return new ReactorReceiver(amqpConnection, entityPath, receiver, new ReceiveLinkHandlerWrapper(handler), tokenManager, dispatcher, retryOptions,
+            metricsProvider);
+    }
+
+    // Note: ReceiveLinkHandler2 will become the ReceiveLinkHandler once the side by side support for v1 and v2 stack
+    // is removed. At that point "ReceiveLinkHandlerWrapper" and this createReceiveLink method will also be removed.
+    public AmqpReceiveLink createReceiveLink(AmqpConnection amqpConnection, String entityPath, Receiver receiver,
+        ReceiveLinkHandler2 handler, TokenManager tokenManager, ReactorDispatcher dispatcher,
+        AmqpRetryOptions retryOptions, AmqpMetricsProvider metricsProvider) {
+        return new ReactorReceiver(amqpConnection, entityPath, receiver, new ReceiveLinkHandlerWrapper(handler), tokenManager, dispatcher, retryOptions,
+            metricsProvider);
+    }
+}

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

@@ -56,7 +56,7 @@ public static LoggingEventBuilder addSignalTypeAndResult(LoggingEventBuilder log
      * and {@code getDescription()} under {@code errorDescription} keys.
      *
      * If errorCondition is {@code null} does not add properties.
-
+     *
      * @return updated {@link LoggingEventBuilder} for chaining.
      */
     public static LoggingEventBuilder addErrorCondition(LoggingEventBuilder logBuilder, ErrorCondition errorCondition) {

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

@@ -4,11 +4,13 @@
 package com.azure.core.amqp.implementation;
 
 import com.azure.core.amqp.AmqpLink;
+import org.apache.qpid.proton.amqp.transport.DeliveryState;
 import org.apache.qpid.proton.message.Message;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import java.io.UncheckedIOException;
+import java.util.concurrent.RejectedExecutionException;
 import java.util.function.Supplier;
 
 /**
@@ -18,6 +20,14 @@
  * Specification 1.0: Links</a>
  */
 public interface AmqpReceiveLink extends AmqpLink {
+    /**
+     * Gets the unique identifier of the Amqp connection hosting the receive link.
+     *
+     * @return The connection identifier.
+     */
+    // Note: Ideally, we may expose connectionId in AmqpLink, but given it's a public contract, lets not do that until a use case needing it.
+    String getConnectionId();
+
     /**
      * Initialises the link from the client to the message broker and begins to receive messages from the broker.
      *
@@ -26,6 +36,16 @@ public interface AmqpReceiveLink extends AmqpLink {
      */
     Flux<Message> receive();
 
+    /**
+     * Updates the disposition state of a message uniquely identified by the given delivery tag.
+     *
+     * @param deliveryTag delivery tag of message.
+     * @param deliveryState Delivery state of message.
+     *
+     * @return A Mono that completes when the state is successfully updated and acknowledged by message broker.
+     */
+    Mono<Void> updateDisposition(String deliveryTag, DeliveryState deliveryState);
+
     /**
      * Schedule to adds the specified number of credits to the link.
      *
@@ -39,6 +59,17 @@ public interface AmqpReceiveLink extends AmqpLink {
      */
     Mono<Void> addCredits(int credits);
 
+    /**
+     * Schedules an event to send a credit to the broker. The API takes a {@link Supplier} that returns the credit
+     * to send. The supplier allows providing the most up-to-date credit value when the scheduler picks the scheduled
+     * work for execution rather than the credit at the time of scheduling.
+     *
+     * @param creditSupplier the supplier that returns the credit to send.
+     * @throws RejectedExecutionException if the scheduler rejects the scheduling attempt (e.g., the scheduler is closed).
+     * @throws UncheckedIOException if an IO error occurs when scheduling.
+     */
+    void addCredit(Supplier<Long> creditSupplier);
+
     /**
      * Gets the current number of credits this link has.
      *

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

@@ -17,17 +17,25 @@ public final class ClientConstants {
     public static final String LINK_NAME_KEY = "linkName";
     public static final String ENTITY_PATH_KEY = "entityPath";
     public static final String ENTITY_NAME_KEY = "entityName";
+    public static final String UPDATED_LINK_CREDIT_KEY = "updatedLinkCredit";
+    public static final String REMOTE_CREDIT_KEY = "remoteCredit";
+    public static final String IS_PARTIAL_DELIVERY_KEY = "delivery.isPartial";
+    public static final String IS_SETTLED_DELIVERY_KEY = "delivery.isSettled";
     public static final String SESSION_NAME_KEY = "sessionName";
     public static final String FULLY_QUALIFIED_NAMESPACE_KEY = "namespace";
     public static final String OPERATION_NAME_KEY = "amqpOperation";
+    public static final String DELIVERY_KEY = "delivery";
     public static final String DELIVERY_STATE_KEY = "deliveryState";
+    public static final String DELIVERY_TAG_KEY = "lockToken";
     public static final String ERROR_CONDITION_KEY = "errorCondition";
     public static final String ERROR_DESCRIPTION_KEY = "errorDescription";
     public static final String EMIT_RESULT_KEY = "emitResult";
     public static final String SIGNAL_TYPE_KEY = "signalType";
     public static final String HOSTNAME_KEY = "hostName";
-    public static final String INTERVAL_KEY = "interval_ms";
+    public static final String INTERVAL_KEY = "intervalMs";
     public static final String SUBSCRIBER_ID_KEY = "subscriberId";
+    public static final String PUMP_ID_KEY = "pumpId";
+    public static final String CALL_SITE_KEY = "callSite";
 
     /**
      * The default maximum allowable size, in bytes, for a batch to be sent.