Avoid reconnection fix (#140)

* Handle CreateProducer/Consumer error

* When A client is disconnected due to a Permissions error it has to be closed
to avoid reconnection loop

* Raise an exception during the initialization
* Avoid the reconnection during the init part
Signed-off-by: Gabriele Santomaggio <[email protected]>
Co-authored-by: Luke Bakken <[email protected]>

Author: Gsantomaggio

RabbitMQ.Stream.Client/Reliable/ReliableConsumer.cs

@@ -91,18 +91,15 @@ protected override async Task GetNewReliable(bool boot)
             _reliableConsumerConfig.ReconnectStrategy.WhenConnected(ToString());
         }
 
-        catch (CreateProducerException ce)
+        catch (Exception e)
         {
-            LogEventSource.Log.LogError("ReliableConsumer closed. ", ce);
+            LogEventSource.Log.LogError("Error during consumer initialization: ", e);
+            throw;
         }
-        catch (Exception e)
+        finally
         {
-            LogEventSource.Log.LogError("Error during initialization: ", e);
             SemaphoreSlim.Release();
-            await TryToReconnect(_reliableConsumerConfig.ReconnectStrategy);
         }
-
-        SemaphoreSlim.Release();
     }
 
     // just close the consumer. See base/metadataupdate
@@ -111,7 +108,10 @@ protected override async Task CloseReliable()
         await SemaphoreSlim.WaitAsync(10);
         try
         {
-            await _consumer.Close();
+            if (_consumer != null)
+            {
+                await _consumer.Close();
+            }
         }
         finally
         {

RabbitMQ.Stream.Client/Reliable/ReliableProducer.cs

@@ -116,27 +116,26 @@ protected override async Task GetNewReliable(bool boot)
                     await _producer.GetLastPublishingId());
             }
         }
-
-        catch (CreateProducerException ce)
+        catch (Exception e)
         {
-            LogEventSource.Log.LogError("ReliableProducer closed", ce);
+            LogEventSource.Log.LogError("Error during producer initialization: ", e);
+            throw;
         }
-        catch (Exception e)
+        finally
         {
-            LogEventSource.Log.LogError("Error during initialization: ", e);
             SemaphoreSlim.Release();
-            await TryToReconnect(_reliableProducerConfig.ReconnectStrategy);
         }
-
-        SemaphoreSlim.Release();
     }
 
     protected override async Task CloseReliable()
     {
         await SemaphoreSlim.WaitAsync(10);
         try
         {
-            await _producer.Close();
+            if (_producer != null)
+            {
+                await _producer.Close();
+            }
         }
         finally
         {
@@ -151,7 +150,10 @@ public override async Task Close()
         {
             _needReconnect = false;
             _confirmationPipe.Stop();
-            await _producer.Close();
+            if (_producer != null)
+            {
+                await _producer.Close();
+            }
         }
         finally
         {

Tests/PermissionTests.cs

@@ -2,7 +2,9 @@
 // 2.0, and the Mozilla Public License, version 2.0.
 // Copyright (c) 2007-2020 VMware, Inc.
 
+using System;
 using RabbitMQ.Stream.Client;
+using RabbitMQ.Stream.Client.Reliable;
 using Xunit;
 
 namespace Tests
@@ -18,38 +20,68 @@ public async void AccessToStreamWithoutGrantsShouldRaiseErrorTest()
             // load definition creates users and streams to test the access
             // the user "test" can't access on "no_access_stream"
             const string stream = "no_access_stream";
-            var config = new StreamSystemConfig()
-            {
-                Password = "test",
-                UserName = "test",
-                VirtualHost = "/"
-            };
+            var config = new StreamSystemConfig() { Password = "test", UserName = "test", VirtualHost = "/" };
             var system = await StreamSystem.Create(config);
 
             await Assert.ThrowsAsync<CreateProducerException>(
                 async () =>
                 {
                     await system.CreateProducer(
-                        new ProducerConfig
-                        {
-                            Reference = "producer",
-                            Stream = stream,
-                        });
+                        new ProducerConfig { Reference = "producer", Stream = stream, });
+                }
+            );
+
+            await Assert.ThrowsAsync<CreateProducerException>(
+                async () =>
+                {
+                    await ReliableProducer.CreateReliableProducer(
+                        new ReliableProducerConfig() { Stream = stream, StreamSystem = system });
                 }
             );
 
+            ReliableProducer reliableProducer = null;
+            try
+            {
+                reliableProducer = await ReliableProducer.CreateReliableProducer(
+                    new ReliableProducerConfig() { Stream = stream, StreamSystem = system });
+            }
+            catch (Exception)
+            {
+                // we already tested the Exception (CreateProducerException)
+            }
+            // here the reliableProducer must be closed due of the CreateProducerException
+            Assert.False(reliableProducer != null && reliableProducer.IsOpen());
+
             await Assert.ThrowsAsync<CreateConsumerException>(
                 async () =>
                 {
                     await system.CreateConsumer(
-                        new ConsumerConfig()
-                        {
-                            Reference = "consumer",
-                            Stream = stream,
-                        });
+                        new ConsumerConfig() { Reference = "consumer", Stream = stream, });
                 }
             );
 
+            await Assert.ThrowsAsync<CreateConsumerException>(
+                async () =>
+                {
+                    await ReliableConsumer.CreateReliableConsumer(
+                        new ReliableConsumerConfig() { Stream = stream, StreamSystem = system });
+                }
+            );
+
+            ReliableConsumer reliableConsumer = null;
+            try
+            {
+                reliableConsumer = await ReliableConsumer.CreateReliableConsumer(
+                    new ReliableConsumerConfig() { Stream = stream, StreamSystem = system });
+            }
+            catch (Exception)
+            {
+                // we already tested the Exception (CreateConsumerException)
+            }
+
+            // here the reliableConsumer must be closed due of the CreateConsumerException
+            Assert.False(reliableConsumer != null && reliableConsumer.IsOpen());
+
             await system.Close();
         }
     }