[Event Hubs] Track Two (Link Close Fix) (#9352)

The focus of these changes is to explicitly close the session associated with an AMQP link when the link itself is closed via the FaultTolerantAmqpObject abstraction.
Azure · Jan 7, 2020 · 02c143c · 02c143c
1 parent f2f0ed4
commit 02c143c
Show file tree

Hide file tree

Showing 3 changed files with 32 additions and 13 deletions.
diff --git a/sdk/eventhub/Azure.Messaging.EventHubs/src/Amqp/AmqpClient.cs b/sdk/eventhub/Azure.Messaging.EventHubs/src/Amqp/AmqpClient.cs
@@ -156,7 +156,14 @@ protected AmqpClient(string host,
                 Credential = credential;
                 MessageConverter = messageConverter ?? new AmqpMessageConverter();
                 ConnectionScope = connectionScope ?? new AmqpConnectionScope(ServiceEndpoint, eventHubName, credential, clientOptions.TransportType, clientOptions.Proxy);
-                ManagementLink = new FaultTolerantAmqpObject<RequestResponseAmqpLink>(timeout => ConnectionScope.OpenManagementLinkAsync(timeout, CancellationToken.None), link => link.SafeClose());
+
+                ManagementLink = new FaultTolerantAmqpObject<RequestResponseAmqpLink>(
+                    timeout => ConnectionScope.OpenManagementLinkAsync(timeout, CancellationToken.None),
+                    link =>
+                    {
+                        link.Session?.SafeClose();
+                        link.SafeClose();
+                    });
             }
             finally
             {

diff --git a/sdk/eventhub/Azure.Messaging.EventHubs/src/Amqp/AmqpConsumer.cs b/sdk/eventhub/Azure.Messaging.EventHubs/src/Amqp/AmqpConsumer.cs
@@ -159,17 +159,22 @@ public AmqpConsumer(string eventHubName,
             RetryPolicy = retryPolicy;
             MessageConverter = messageConverter;
 
-            ReceiveLink = new FaultTolerantAmqpObject<ReceivingAmqpLink>(timeout =>
-                ConnectionScope.OpenConsumerLinkAsync(
-                    consumerGroup,
-                    partitionId,
-                    CurrentEventPosition,
-                    timeout,
-                    prefetchCount ?? DefaultPrefetchCount,
-                    ownerLevel,
-                    trackLastEnqueuedEventProperties,
-                    CancellationToken.None),
-                link => link.SafeClose());
+            ReceiveLink = new FaultTolerantAmqpObject<ReceivingAmqpLink>(
+                timeout =>
+                    ConnectionScope.OpenConsumerLinkAsync(
+                        consumerGroup,
+                        partitionId,
+                        CurrentEventPosition,
+                        timeout,
+                        prefetchCount ?? DefaultPrefetchCount,
+                        ownerLevel,
+                        trackLastEnqueuedEventProperties,
+                        CancellationToken.None),
+                link =>
+                {
+                    link.Session?.SafeClose();
+                    link.SafeClose();
+                });
         }
 
         /// <summary>

diff --git a/sdk/eventhub/Azure.Messaging.EventHubs/src/Amqp/AmqpProducer.cs b/sdk/eventhub/Azure.Messaging.EventHubs/src/Amqp/AmqpProducer.cs
@@ -125,7 +125,14 @@ public AmqpProducer(string eventHubName,
             RetryPolicy = retryPolicy;
             ConnectionScope = connectionScope;
             MessageConverter = messageConverter;
-            SendLink = new FaultTolerantAmqpObject<SendingAmqpLink>(timeout => CreateLinkAndEnsureProducerStateAsync(partitionId, timeout, CancellationToken.None), link => link.SafeClose());
+
+            SendLink = new FaultTolerantAmqpObject<SendingAmqpLink>(
+                timeout => CreateLinkAndEnsureProducerStateAsync(partitionId, timeout, CancellationToken.None),
+                link =>
+                {
+                    link.Session?.SafeClose();
+                    link.SafeClose();
+                });
         }
 
         /// <summary>