restore internal close

Signed-off-by: Gabriele Santomaggio <[email protected]>

Author: Gsantomaggio

RabbitMQ.Stream.Client/Client.cs

@@ -396,15 +396,17 @@ await Request<DeletePublisherRequest, DeletePublisherResponse>(corr =>
                 if (response.ResponseCode == ResponseCode.Ok)
                     return (subscriptionId, response);
 
-                ClientExceptions.MaybeThrowException(response.ResponseCode, $"Error while creating consumer for stream {config.Stream}");
+                ClientExceptions.MaybeThrowException(response.ResponseCode,
+                    $"Error while creating consumer for stream {config.Stream}");
             }
             catch (Exception e)
             {
                 // if the response code is not ok we need to remove the subscription
                 // and close the connection if necessary. 
                 consumers.Remove(subscriptionId);
                 await MaybeClose("Create Consumer Exception", config.Stream, config.Pool).ConfigureAwait(false);
-                throw new CreateConsumerException($"Error while creating consumer for stream {config.Stream}, error: {e.Message}");
+                throw new CreateConsumerException(
+                    $"Error while creating consumer for stream {config.Stream}, error: {e.Message}");
             }
 
             return (subscriptionId, new SubscribeResponse(subscriptionId, ResponseCode.InternalError));
@@ -724,7 +726,7 @@ private async ValueTask<bool> SendHeartBeat()
         private void InternalClose()
         {
             _heartBeatHandler.Close();
-            // IsClosed = true;
+            IsClosed = true;
         }
 
         private bool HasEntities()
@@ -747,6 +749,7 @@ public async Task<CloseResponse> Close(string reason)
                 return new CloseResponse(0, ResponseCode.Ok);
             }
 
+            InternalClose();
             try
             {
                 var result =
@@ -769,8 +772,6 @@ public async Task<CloseResponse> Close(string reason)
                 connection.Dispose();
             }
 
-            InternalClose();
-
             return new CloseResponse(0, ResponseCode.Ok);
         }
 

RabbitMQ.Stream.Client/Reliable/ReliableBase.cs

@@ -60,11 +60,12 @@ private async Task Init(bool boot, IReconnectStrategy reconnectStrategy)
         await SemaphoreSlim.WaitAsync().ConfigureAwait(false);
         try
         {
-            await CreateNewEntity(boot).ConfigureAwait(false);
             lock (_lock)
             {
                 _isOpen = true;
             }
+
+            await CreateNewEntity(boot).ConfigureAwait(false);
         }
 
         catch (Exception e)