Misc fixes

sdk/servicebus/Azure.Messaging.ServiceBus/Directory.Build.props

@@ -3,4 +3,16 @@
     Add any shared properties you want for the projects under this package directory that need to be set before the auto imported Directory.Build.props
   -->
   <Import Project="$([MSBuild]::GetDirectoryNameOfFileAbove($(MSBuildThisFileDirectory).., Directory.Build.props))\Directory.Build.props" />
+
+  <PropertyGroup>
+    <!--
+    These analyzers are blocked by https://github.com/Azure/azure-sdk-tools/issues/127
+     -->
+    <NoWarn>
+      $(NoWarn);
+      AZC0006;
+      AZC0007;
+    </NoWarn>
+  </PropertyGroup>
+
 </Project>

sdk/servicebus/Azure.Messaging.ServiceBus/src/Amqp/AmqpConsumer.cs

@@ -8,16 +8,16 @@
 using System.Threading;
 using System.Threading.Tasks;
 using Azure.Core;
-using Azure.Messaging.ServiceBus.Receiver;
 using Azure.Messaging.ServiceBus.Core;
 using Azure.Messaging.ServiceBus.Diagnostics;
 using Microsoft.Azure.Amqp;
+using Microsoft.Azure.Amqp.Framing;
 
 namespace Azure.Messaging.ServiceBus.Amqp
 {
     /// <summary>
     ///   A transport client abstraction responsible for brokering operations for AMQP-based connections.
-    ///   It is intended that the public <see cref="ServiceBusReceiverClient" /> make use of an instance
+    ///   It is intended that the public <see cref="ServiceBusReceiver" /> make use of an instance
     ///   via containment and delegate operations to it.
     /// </summary>
     ///
@@ -45,7 +45,7 @@ internal class AmqpConsumer : TransportConsumer
         ///   The name of the Service Bus entity to which the client is bound.
         /// </summary>
         ///
-        private string EntityName { get; }
+        public override string EntityName { get; }
 
         /// <summary>
         ///   The identifier of the Service Bus entity partition that this consumer is associated with.  Events will be read
@@ -58,25 +58,23 @@ internal class AmqpConsumer : TransportConsumer
         ///   The policy to use for determining retry behavior for when an operation fails.
         /// </summary>
         ///
-        private ServiceBusRetryPolicy RetryPolicy { get; }
-
-        ///// <summary>
-        /////   The converter to use for translating between AMQP messages and client library
-        /////   types.
-        ///// </summary>
-        //private AmqpMessageConverter MessageConverter { get; }
-
-        ///// <summary>
-        /////   The AMQP connection scope responsible for managing transport constructs for this instance.
-        ///// </summary>
-        /////
-        //internal AmqpConnectionScope ConnectionScope { get; }
-
-        ///// <summary>
-        /////   The AMQP link intended for use with receiving operations.
-        ///// </summary>
-        /////
-        //internal FaultTolerantAmqpObject<ReceivingAmqpLink> ReceiveLink { get; }
+        private readonly ServiceBusRetryPolicy _retryPolicy;
+
+        /// <summary>
+        /// Indicates whether or not this is a receiver scoped to a session.
+        /// </summary>
+        private readonly bool _isSessionReceiver;
+
+        /// <summary>
+        ///   The AMQP connection scope responsible for managing transport constructs for this instance.
+        /// </summary>
+        ///
+        private readonly AmqpConnectionScope _connectionScope;
+
+        /// <inheritdoc/>
+        public override TransportConnectionScope ConnectionScope =>
+            _connectionScope;
+
 
         /// <summary>
         ///   Initializes a new instance of the <see cref="AmqpConsumer"/> class.
@@ -88,6 +86,7 @@ internal class AmqpConsumer : TransportConsumer
         /// <param name="connectionScope">The AMQP connection context for operations .</param>
         /// <param name="retryPolicy">The retry policy to consider when an operation fails.</param>
         /// <param name="sessionId"></param>
+        /// <param name="isSessionReceiver"></param>
         ///
         /// <remarks>
         ///   As an internal type, this class performs only basic sanity checks against its arguments.  It
@@ -104,25 +103,26 @@ public AmqpConsumer(
             uint? prefetchCount,
             AmqpConnectionScope connectionScope,
             ServiceBusRetryPolicy retryPolicy,
-            string sessionId)
+            string sessionId,
+            bool isSessionReceiver)
         {
             Argument.AssertNotNullOrEmpty(entityName, nameof(entityName));
             Argument.AssertNotNull(connectionScope, nameof(connectionScope));
             Argument.AssertNotNull(retryPolicy, nameof(retryPolicy));
             EntityName = entityName;
-            ConnectionScope = connectionScope;
-            RetryPolicy = retryPolicy;
+            _connectionScope = connectionScope;
+            _retryPolicy = retryPolicy;
+            _isSessionReceiver = isSessionReceiver;
 
             ReceiveLink = new FaultTolerantAmqpObject<ReceivingAmqpLink>(
                 timeout =>
-                    ConnectionScope.OpenConsumerLinkAsync(
-                        //consumerGroup,
-                        //partitionId,
-                        timeout,
-                        prefetchCount ?? DefaultPrefetchCount,
-                        ownerLevel,
-                        sessionId,
-                        CancellationToken.None),
+                    _connectionScope.OpenConsumerLinkAsync(
+                        timeout: timeout,
+                        prefetchCount: prefetchCount ?? DefaultPrefetchCount,
+                        ownerLevel: ownerLevel,
+                        sessionId: sessionId,
+                        isSessionReceiver: isSessionReceiver,
+                        cancellationToken: CancellationToken.None),
                 link =>
                 {
                     CloseLink(link);
@@ -155,7 +155,7 @@ public override async Task<IEnumerable<ServiceBusMessage>> ReceiveAsync(
 
             var receivedMessageCount = 0;
             var failedAttemptCount = 0;
-            var tryTimeout = RetryPolicy.CalculateTryTimeout(0);
+            var tryTimeout = _retryPolicy.CalculateTryTimeout(0);
             var waitTime = (maximumWaitTime ?? tryTimeout);
             var link = default(ReceivingAmqpLink);
             var retryDelay = default(TimeSpan?);
@@ -222,14 +222,14 @@ public override async Task<IEnumerable<ServiceBusMessage>> ReceiveAsync(
                         // Otherwise, bubble the exception.
 
                         ++failedAttemptCount;
-                        retryDelay = RetryPolicy.CalculateRetryDelay(activeEx, failedAttemptCount);
+                        retryDelay = _retryPolicy.CalculateRetryDelay(activeEx, failedAttemptCount);
 
                         if ((retryDelay.HasValue) && (!ConnectionScope.IsDisposed) && (!cancellationToken.IsCancellationRequested))
                         {
                             ServiceBusEventSource.Log.MessageReceiveError(EntityName, activeEx.Message);
                             await Task.Delay(UseMinimum(retryDelay.Value, waitTime.CalculateRemaining(stopWatch.Elapsed)), cancellationToken).ConfigureAwait(false);
 
-                            tryTimeout = RetryPolicy.CalculateTryTimeout(failedAttemptCount);
+                            tryTimeout = _retryPolicy.CalculateTryTimeout(failedAttemptCount);
                         }
                         else if (ex is AmqpException)
                         {
@@ -316,5 +316,30 @@ private static TimeSpan UseMinimum(
             TimeSpan firstOption,
             TimeSpan secondOption) =>
             (firstOption < secondOption) ? firstOption : secondOption;
+
+        /// <summary>
+        /// Get the session Id corresponding to this consumer
+        /// </summary>
+        /// <param name="cancellationToken"></param>
+        /// <returns></returns>
+        public override async Task<string> GetSessionId(CancellationToken cancellationToken = default)
+        {
+            if (!_isSessionReceiver)
+            {
+                return null;
+            }
+            ReceivingAmqpLink openedLink = null;
+            await _retryPolicy.RunOperation(
+                async (timeout) =>
+                openedLink = await ReceiveLink.GetOrCreateAsync(timeout).ConfigureAwait(false),
+                EntityName,
+                ConnectionScope,
+                cancellationToken).ConfigureAwait(false);
+
+            var source = (Source)openedLink.Settings.Source;
+            source.FilterSet.TryGetValue<string>(AmqpClientConstants.SessionFilterName, out var sessionId);
+            return sessionId;
+        }
+
     }
 }

sdk/servicebus/Azure.Messaging.ServiceBus/src/Amqp/AmqpSender.cs

@@ -10,15 +10,14 @@
 using Azure.Core;
 using Azure.Messaging.ServiceBus.Core;
 using Azure.Messaging.ServiceBus.Diagnostics;
-using Azure.Messaging.ServiceBus.Sender;
 using Microsoft.Azure.Amqp;
 using Microsoft.Azure.Amqp.Framing;
 
 namespace Azure.Messaging.ServiceBus.Amqp
 {
     /// <summary>
     ///   A transport producer abstraction responsible for brokering operations for AMQP-based connections.
-    ///   It is intended that the public <see cref="ServiceBusSenderClient" /> make use of an instance
+    ///   It is intended that the public <see cref="ServiceBusSender" /> make use of an instance
     ///   via containment and delegate operations to it.
     /// </summary>
     ///

sdk/servicebus/Azure.Messaging.ServiceBus/src/Core/TransportClient.cs

@@ -111,14 +111,16 @@ public abstract Task CancelScheduledMessageAsync(
         /// <param name="ownerLevel">The relative priority to associate with the link; for a non-exclusive link, this value should be <c>null</c>.</param>
         /// <param name="prefetchCount">Controls the number of events received and queued locally without regard to whether an operation was requested.  If <c>null</c> a default will be used.</param>
         /// <param name="sessionId"></param>
+        /// <param name="isSessionReceiver"></param>
         ///
         /// <returns>A <see cref="TransportConsumer" /> configured in the requested manner.</returns>
         ///
         public abstract TransportConsumer CreateConsumer(
             ServiceBusRetryPolicy retryPolicy,
             long? ownerLevel,
             uint? prefetchCount,
-            string sessionId);
+            string sessionId,
+            bool isSessionReceiver);
 
         /// <summary>
         ///   Closes the connection to the transport client instance.

sdk/servicebus/Azure.Messaging.ServiceBus/src/Core/TransportConnectionScope.cs

@@ -0,0 +1,29 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+using System;
+
+namespace Azure.Messaging.ServiceBus.Core
+{
+    internal abstract class TransportConnectionScope : IDisposable
+    {
+        /// <summary>
+        ///   Indicates whether this <see cref="TransportConnectionScope"/> has been disposed.
+        /// </summary>
+        ///
+        /// <value><c>true</c> if disposed; otherwise, <c>false</c>.</value>
+        ///
+        public abstract bool IsDisposed { get; protected set; }
+
+        /// <summary>
+        ///   The recommended timeout to associate with the session.
+        /// </summary>
+        ///
+        public abstract TimeSpan SessionTimeout { get; }
+
+        /// <summary>
+        /// Disposes of the connection scope.
+        /// </summary>
+        public abstract void Dispose();
+    }
+}