Expose client closed flag publicly

sdk/eventhub/Microsoft.Azure.EventHubs/src/Amqp/AmqpEventDataSender.cs

@@ -86,7 +86,7 @@ protected override async Task OnSendAsync(IEnumerable<EventData> eventDatas, str
                     {
                         // Evaluate retry condition?
                         TimeSpan? retryInterval = this.RetryPolicy.GetNextRetryInterval(ex, timeoutHelper.RemainingTime(), ++retryCount);
-                        if (retryInterval != null && !this.EventHubClient.CloseCalled)
+                        if (retryInterval != null && !this.EventHubClient.IsClosed)
                         {
                             await Task.Delay(retryInterval.Value).ConfigureAwait(false);
                             shouldRetry = true;

sdk/eventhub/Microsoft.Azure.EventHubs/src/Amqp/AmqpPartitionReceiver.cs

@@ -104,7 +104,7 @@ protected override async Task<IList<EventData>> OnReceiveAsync(int maxMessageCou
                 {
                     // Evaluate retry condition?
                     TimeSpan? retryInterval = this.RetryPolicy.GetNextRetryInterval(ex, timeoutHelper.RemainingTime(), ++retryCount);
-                    if (retryInterval != null && !this.EventHubClient.CloseCalled)
+                    if (retryInterval != null && !this.EventHubClient.IsClosed)
                     {
                         await Task.Delay(retryInterval.Value).ConfigureAwait(false);
                         shouldRetry = true;

sdk/eventhub/Microsoft.Azure.EventHubs/src/EventHubClient.cs

@@ -19,7 +19,6 @@ namespace Microsoft.Azure.EventHubs
     public abstract class EventHubClient : ClientEntity
     {
         readonly Lazy<EventDataSender> innerSender;
-        bool closeCalled = false;
 
         internal EventHubClient(EventHubsConnectionStringBuilder csb)
             : base($"{nameof(EventHubClient)}{ClientEntity.GetNextId()}({csb.EntityPath})")
@@ -448,8 +447,6 @@ public EventDataBatch CreateBatch(BatchOptions options)
         /// </summary>
         public IWebProxy WebProxy { get; set; }
 
-        internal bool CloseCalled => this.closeCalled;
-
         internal EventDataSender CreateEventSender(string partitionId = null)
         {
             return this.OnCreateEventSender(partitionId);

sdk/eventhub/Microsoft.Azure.EventHubs/src/PartitionReceiver.cs

@@ -224,6 +224,8 @@ public void SetReceiveHandler(IPartitionReceiveHandler receiveHandler, bool invo
         /// <returns>An asynchronous operation</returns>
         public sealed override Task CloseAsync()
         {
+            this.closeCalled = true;
+
             EventHubsEventSource.Log.ClientCloseStart(this.ClientId);
             try
             {

sdk/eventhub/Microsoft.Azure.EventHubs/src/PartitionSender.cs

@@ -179,6 +179,8 @@ public async Task SendAsync(EventDataBatch eventDataBatch)
         /// <returns>An asynchronous operation</returns>
         public override async Task CloseAsync()
         {
+            this.closeCalled  = true;
+
             EventHubsEventSource.Log.ClientCloseStart(this.ClientId);
             try
             {

sdk/eventhub/Microsoft.Azure.EventHubs/src/Primitives/ClientEntity.cs

@@ -20,6 +20,9 @@ public abstract class ClientEntity
         static int nextId;
         RetryPolicy retryPolicy;
 
+        /// <summary />
+        protected bool closeCalled = false;
+
         /// <summary></summary>
         /// <param name="clientId"></param>
         protected ClientEntity(string clientId)
@@ -61,6 +64,12 @@ public RetryPolicy RetryPolicy
         /// <returns>The asynchronous operation</returns>
         public abstract Task CloseAsync();
 
+        /// <summary>
+        /// Returns a boolean representing whether client object is closed or not.
+        /// </summary>
+        /// <value>Returns <see cref="System.Boolean" />.</value>
+        public bool IsClosed => this.closeCalled;
+
         /// <summary>
         /// Registers a <see cref="EventHubsPlugin"/> to be used with this client.
         /// </summary>

sdk/eventhub/Microsoft.Azure.EventHubs/tests/Client/MiscTests.cs

@@ -125,5 +125,75 @@ await using (var scope = await EventHubScope.CreateAsync(1))
                 }
             }
         }
+
+        [Fact]
+        [LiveTest]
+        [DisplayTestMethodName]
+        public async Task ClosingSenderEntity()
+        {
+            await using (var scope = await EventHubScope.CreateAsync(1))
+            {
+                var connectionString = TestUtility.BuildEventHubsConnectionString(scope.EventHubName);
+                var ehClient = EventHubClient.CreateFromConnectionString(connectionString);
+                var ehSender = ehClient.CreatePartitionSender("0");
+
+                await ehSender.CloseAsync();
+                Assert.True(ehSender.IsClosed, "ehSender.IsClosed is not true.");
+                Assert.True(!ehClient.IsClosed, "ehClient.IsClosed is not false.");
+            }
+        }
+
+        [Fact]
+        [LiveTest]
+        [DisplayTestMethodName]
+        public async Task ClosingReceiverEntity()
+        {
+            await using (var scope = await EventHubScope.CreateAsync(1))
+            {
+                var connectionString = TestUtility.BuildEventHubsConnectionString(scope.EventHubName);
+                var ehClient = EventHubClient.CreateFromConnectionString(connectionString);
+                var ehReceiver = ehClient.CreateReceiver(PartitionReceiver.DefaultConsumerGroupName, "0", EventPosition.FromStart());
+
+                await ehReceiver.CloseAsync();
+                Assert.True(ehReceiver.IsClosed, "ehReceiver.IsClosed is not true.");
+                Assert.True(!ehClient.IsClosed, "ehClient.IsClosed is not false.");
+            }
+        }
+
+        [Fact]
+        [LiveTest]
+        [DisplayTestMethodName]
+        public async Task ClosingEventHubClientClosesSenderEntities()
+        {
+            await using (var scope = await EventHubScope.CreateAsync(1))
+            {
+                var connectionString = TestUtility.BuildEventHubsConnectionString(scope.EventHubName);
+                var ehClient = EventHubClient.CreateFromConnectionString(connectionString);
+                var ehSender0 = ehClient.CreatePartitionSender("0");
+                var ehSender1 = ehClient.CreatePartitionSender("1");
+
+                await ehClient.CloseAsync();
+                Assert.True(ehSender0.IsClosed, "ehSender0.IsClosed is not true.");
+                Assert.True(ehSender1.IsClosed, "ehSender1.IsClosed is not true.");
+            }
+        }
+
+        [Fact]
+        [LiveTest]
+        [DisplayTestMethodName]
+        public async Task ClosingEventHubClientClosesReceiverEntities()
+        {
+            await using (var scope = await EventHubScope.CreateAsync(1))
+            {
+                var connectionString = TestUtility.BuildEventHubsConnectionString(scope.EventHubName);
+                var ehClient = EventHubClient.CreateFromConnectionString(connectionString);
+                var ehReceiver = ehClient.CreateReceiver(PartitionReceiver.DefaultConsumerGroupName, "0", EventPosition.FromStart());
+                var ehReceiverEpoch = ehClient.CreateEpochReceiver(PartitionReceiver.DefaultConsumerGroupName, "0", EventPosition.FromEnd(), 0);
+
+                await ehClient.CloseAsync();
+                Assert.True(ehReceiver.IsClosed, "ehReceiver.IsClosed is not true.");
+                Assert.True(ehReceiverEpoch.IsClosed, "ehReceiverEpoch.IsClosed is not true.");
+            }
+        }
     }
 }