Session manager cleanup (Azure#32575)

* Cleanup ReceiverManager CloseReceiverIfNeeded

* Replace CancelSessionAsync to Cancel

* Address Review comments

Author: danielmarbach

sdk/servicebus/Azure.Messaging.ServiceBus/src/Processor/ProcessSessionEventArgs.cs

@@ -140,7 +140,7 @@ public virtual async Task SetSessionStateAsync(
         public virtual void ReleaseSession() =>
             // manager will be null if instance created using the public constructor which is exposed for testing purposes
             // This will be awaited when closing the receiver.
-            _ = _manager?.CancelSessionAsync();
+            _ = _manager?.CancelAsync();
 
         ///<inheritdoc cref="ServiceBusSessionReceiver.RenewSessionLockAsync(CancellationToken)"/>
         public virtual async Task RenewSessionLockAsync(CancellationToken cancellationToken = default)

sdk/servicebus/Azure.Messaging.ServiceBus/src/Processor/ProcessSessionMessageEventArgs.cs

@@ -232,7 +232,7 @@ await _sessionReceiver.DeferMessageAsync(
         public virtual void ReleaseSession() =>
             // manager will be null if instance created using the public constructor which is exposed for testing purposes
             // This will be awaited when closing the receiver.
-            _ = _manager?.CancelSessionAsync();
+            _ = _manager?.CancelAsync();
 
         ///<inheritdoc cref="ServiceBusSessionReceiver.RenewSessionLockAsync(CancellationToken)"/>
         public virtual async Task RenewSessionLockAsync(CancellationToken cancellationToken = default)

sdk/servicebus/Azure.Messaging.ServiceBus/src/Processor/ReceiverManager.cs

@@ -59,9 +59,7 @@ public ReceiverManager(
             _scopeFactory = scopeFactory;
         }
 
-        public virtual async Task CloseReceiverIfNeeded(
-            CancellationToken cancellationToken,
-            bool forceClose = false)
+        public virtual async Task CloseReceiverIfNeeded(CancellationToken cancellationToken)
         {
             var capturedReceiver = Receiver;
             if (capturedReceiver != null)
@@ -123,6 +121,8 @@ await RaiseExceptionReceived(
             }
         }
 
+        public virtual Task CancelAsync() => Task.CompletedTask;
+
         protected async Task ProcessOneMessageWithinScopeAsync(ServiceBusReceivedMessage message, string activityName, CancellationToken cancellationToken)
         {
             using DiagnosticScope scope = _scopeFactory.CreateScope(activityName, DiagnosticScope.ActivityKind.Consumer);

sdk/servicebus/Azure.Messaging.ServiceBus/src/Processor/ServiceBusProcessor.cs

@@ -695,7 +695,7 @@ private void ReconcileReceiverManagers(int maxConcurrentSessions)
                             _orphanedReceiverManagers.Add(_receiverManagers[0]);
 
                             // these tasks will be awaited when closing the orphaned receivers as part of CloseAsync
-                            _ = ((SessionReceiverManager) _receiverManagers[0]).CancelSessionAsync();
+                            _ = _receiverManagers[0].CancelAsync();
                             _receiverManagers.RemoveAt(0);
                         }
                     }
@@ -1013,9 +1013,7 @@ public virtual async Task CloseAsync(
 
             foreach (ReceiverManager receiverManager in _receiverManagers.Concat(_orphanedReceiverManagers))
             {
-                await receiverManager.CloseReceiverIfNeeded(
-                        cancellationToken,
-                        forceClose: true)
+                await receiverManager.CloseReceiverIfNeeded(cancellationToken)
                     .ConfigureAwait(false);
             }
         }

sdk/servicebus/Azure.Messaging.ServiceBus/src/Processor/SessionReceiverManager.cs

@@ -145,21 +145,24 @@ private async Task CreateReceiver(CancellationToken processorCancellationToken)
             }
         }
 
-        public override async Task CloseReceiverIfNeeded(
-            CancellationToken processorCancellationToken,
-            bool forceClose = false)
+        public override async Task CloseReceiverIfNeeded(CancellationToken cancellationToken)
+        {
+            await CloseReceiverCore(forceClose: true, cancellationToken).ConfigureAwait(false);
+        }
+
+        private async Task CloseReceiverCore(bool forceClose, CancellationToken cancellationToken)
         {
             bool releaseSemaphore = false;
             try
             {
-                // Intentionally not including processor cancellation token as
+                // Intentionally not including cancellation token as
                 // we need to ensure that we at least attempt to close the receiver if needed.
                 await WaitSemaphore(CancellationToken.None).ConfigureAwait(false);
                 releaseSemaphore = true;
 
                 if (forceClose)
                 {
-                    await CloseReceiver(processorCancellationToken).ConfigureAwait(false);
+                    await CloseReceiver(cancellationToken).ConfigureAwait(false);
                     return;
                 }
 
@@ -180,7 +183,7 @@ public override async Task CloseReceiverIfNeeded(
                         // as this means the session lock was lost or the user requested to close the session.
                         _sessionCancellationSource.IsCancellationRequested)
                     {
-                        await CloseReceiver(processorCancellationToken).ConfigureAwait(false);
+                        await CloseReceiver(cancellationToken).ConfigureAwait(false);
                     }
                 }
             }
@@ -225,7 +228,7 @@ await RaiseExceptionReceived(
                 // cancel the automatic session lock renewal
                 try
                 {
-                    await CancelSessionAsync().ConfigureAwait(false);
+                    await CancelAsync().ConfigureAwait(false);
                 }
                 catch (Exception ex) when (ex is TaskCanceledException)
                 {
@@ -317,7 +320,7 @@ await ProcessOneMessageWithinScopeAsync(
                     if (sbException.Reason == ServiceBusFailureReason.SessionLockLost)
                     {
                         // this will be awaited when closing the receiver
-                        _ = CancelSessionAsync();
+                        _ = CancelAsync();
                     }
                 }
                 await RaiseExceptionReceived(
@@ -334,7 +337,7 @@ await RaiseExceptionReceived(
             {
                 if (canProcess)
                 {
-                    await CloseReceiverIfNeeded(processorCancellationToken).ConfigureAwait(false);
+                    await CloseReceiverCore(forceClose: false, processorCancellationToken).ConfigureAwait(false);
                 }
             }
         }
@@ -406,7 +409,7 @@ protected override async Task RaiseExceptionReceived(ProcessErrorEventArgs event
             }
         }
 
-        internal async Task CancelSessionAsync()
+        public override async Task CancelAsync()
         {
             if (_sessionCancellationSource is { IsCancellationRequested: false })
             {