Merge pull request #127 from Project-MONAI/nds-hunting-rabbit

neildsouth · web-flow · commit 525b1d105e63 · 2022-11-16T16:33:38.000Z
Nds hunting rabbit
diff --git a/src/Plugins/RabbitMQ/ChannelType.cs b/src/Plugins/RabbitMQ/ChannelType.cs
@@ -0,0 +1,24 @@
+﻿/*
+ * Copyright 2021-2022 MONAI Consortium
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+namespace Monai.Deploy.Messaging.RabbitMQ
+{
+    public enum ChannelType
+    {
+        Subscriber,
+        Publisher
+    }
+}
diff --git a/src/Plugins/RabbitMQ/Factory/IRabbitMQConnectionFactory.cs b/src/Plugins/RabbitMQ/Factory/IRabbitMQConnectionFactory.cs
@@ -32,6 +32,6 @@ public interface IRabbitMQConnectionFactory
         /// <param name="useSSL">Encrypt communication</param>
         /// <param name="portNumber">Port Number</param>
         /// <returns>Instance of <see cref="IModel"/>.</returns>
-        IModel? CreateChannel( string hostName, string username, string password, string virtualHost, string useSSL, string portNumber);
+        IModel? CreateChannel( ChannelType type, string hostName, string username, string password, string virtualHost, string useSSL, string portNumber);
     }
 }
diff --git a/src/Plugins/RabbitMQ/Factory/RabbitMqConnectionFactory.cs b/src/Plugins/RabbitMQ/Factory/RabbitMqConnectionFactory.cs
@@ -41,17 +41,17 @@ public RabbitMQConnectionFactory(ILogger<RabbitMQConnectionFactory> logger)
         }
 
 
-        public IModel CreateChannel(string hostName, string username, string password, string virtualHost, string useSSL, string portNumber)
+        public IModel CreateChannel(ChannelType type, string hostName, string username, string password, string virtualHost, string useSSL, string portNumber)
         {
             Guard.Against.NullOrWhiteSpace(hostName);
             Guard.Against.NullOrWhiteSpace(username);
             Guard.Against.NullOrWhiteSpace(password);
             Guard.Against.NullOrWhiteSpace(virtualHost);
 
-            var key = $"{hostName}{username}{HashPassword(password)}{virtualHost}";
+            var key = $"{type}{hostName}{username}{HashPassword(password)}{virtualHost}";
 
             var connection = _connections.AddOrUpdate(key,
-                x => MakeConnection(hostName, username, password, virtualHost, key, useSSL, portNumber),
+                x => MakeConnection(type, hostName, username, password, virtualHost, key, useSSL, portNumber),
                 (updateKey, updateConnection) =>
                 {
                     // If connection to RMQ is lost and:
@@ -62,13 +62,13 @@ public IModel CreateChannel(string hostName, string username, string password, s
                     {
                         if (updateConnection.model.IsClosed)
                         {
-                            updateConnection.model = MakeChannel(updateConnection.connection, key);
+                            updateConnection.model = MakeChannel(type, updateConnection.connection, key);
                         }
                         return updateConnection;
                     }
                     else
                     {
-                        return MakeConnection(hostName, username, password, virtualHost, key, useSSL, portNumber);
+                        return MakeConnection(type, hostName, username, password, virtualHost, key, useSSL, portNumber);
                     }
                 });
 
@@ -87,18 +87,22 @@ private void OnException(CallbackExceptionEventArgs args, IConnection value, str
 
         }
 
-        private (IConnection, IModel) MakeConnection(string hostName, string username, string password, string virtualHost, string key, string useSSL, string portNumber)
+        private (IConnection, IModel) MakeConnection(ChannelType type, string hostName, string username, string password, string virtualHost, string key, string useSSL, string portNumber)
         {
             var connection = CreateConnectionOnly(hostName, username, password, virtualHost, key, useSSL, portNumber);
-            var model = MakeChannel(connection, key);
+            var model = MakeChannel(type, connection, key);
             return (connection, model);
         }
 
-        private IModel MakeChannel(IConnection connection, string key)
+        private IModel MakeChannel(ChannelType type, IConnection connection, string key)
         {
             var model = connection.CreateModel();
             model.CallbackException += (sender, args) => OnException(args, connection, key);
             model.ModelShutdown += (sender, args) => ConnectionShutdown(args, connection, key);
+            if (type == ChannelType.Publisher)
+            {
+                model.ConfirmSelect();
+            }
             return model;
         }
 
diff --git a/src/Plugins/RabbitMQ/Publisher/RabbitMqMessagePublisherService.cs b/src/Plugins/RabbitMQ/Publisher/RabbitMqMessagePublisherService.cs
@@ -108,12 +108,11 @@ public Task Publish(string topic, Message message)
 
             _logger.PublshingRabbitMQ(_endpoint, _virtualHost, _exchange, topic);
 
-            var channel = _rabbitMqConnectionFactory.CreateChannel(_endpoint, _username, _password, _virtualHost, _useSSL, _portNumber);
+            var channel = _rabbitMqConnectionFactory.CreateChannel(ChannelType.Publisher, _endpoint, _username, _password, _virtualHost, _useSSL, _portNumber);
 
             if (channel is null) { throw new NullReferenceException("RabbitMq channel returned null"); }
 
             channel.ExchangeDeclare(_exchange, ExchangeType.Topic, durable: true, autoDelete: false);
-            channel.ConfirmSelect();
 
             var properties = channel.CreateBasicProperties();
             properties.Persistent = true;
diff --git a/src/Plugins/RabbitMQ/RabbitMQHealthCheck.cs b/src/Plugins/RabbitMQ/RabbitMQHealthCheck.cs
@@ -47,6 +47,7 @@ public RabbitMQHealthCheck(
             try
             {
                 var channel = _connectionFactory.CreateChannel(
+                    ChannelType.Subscriber,
                     _options[ConfigurationKeys.EndPoint],
                     _options[ConfigurationKeys.Username],
                     _options[ConfigurationKeys.Password],
diff --git a/src/Plugins/RabbitMQ/Subscriber/RabbitMqMessageSubscriberService.cs b/src/Plugins/RabbitMQ/Subscriber/RabbitMqMessageSubscriberService.cs
@@ -108,7 +108,7 @@ private void CreateChannel()
                     .Execute(() =>
                     {
                         _logger.ConnectingToRabbitMQ(Name, _endpoint, _virtualHost);
-                        _channel = _rabbitMqConnectionFactory.CreateChannel(_endpoint, _username, _password, _virtualHost, _useSSL, _portNumber);
+                        _channel = _rabbitMqConnectionFactory.CreateChannel(ChannelType.Subscriber, _endpoint, _username, _password, _virtualHost, _useSSL, _portNumber);
                         _channel.ExchangeDeclare(_exchange, ExchangeType.Topic, durable: true, autoDelete: false);
                         _channel.ExchangeDeclare(_deadLetterExchange, ExchangeType.Topic, durable: true, autoDelete: false);
                         _channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);
diff --git a/src/Plugins/RabbitMQ/Tests/RabbitMQHealthCheckTest.cs b/src/Plugins/RabbitMQ/Tests/RabbitMQHealthCheckTest.cs
@@ -42,7 +42,7 @@ public RabbitMQHealthCheckTest()
         [Fact]
         public async Task CheckHealthAsync_WhenFailed_ReturnUnhealthy()
         {
-            _connectionFactory.Setup(p => p.CreateChannel(It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>()))
+            _connectionFactory.Setup(p => p.CreateChannel(It.IsAny<ChannelType>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>()))
                 .Throws(new Exception("error"));
 
             var healthCheck = new RabbitMQHealthCheck(_connectionFactory.Object, _options, _logger.Object, (d) => { });
@@ -58,7 +58,7 @@ public async Task CheckHealthAsync_WhenSucceeds_ReturnHealthy()
         {
             var channel = new Mock<IModel>();
             channel.Setup(p => p.Close());
-            _connectionFactory.Setup(p => p.CreateChannel(It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>()))
+            _connectionFactory.Setup(p => p.CreateChannel(It.IsAny<ChannelType>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>()))
                 .Returns(channel.Object);
             var healthCheck = new RabbitMQHealthCheck(_connectionFactory.Object, _options, _logger.Object, (d) => { });
             var results = await healthCheck.CheckHealthAsync(new HealthCheckContext()).ConfigureAwait(false);
diff --git a/src/Plugins/RabbitMQ/Tests/RabbitMqMessagePublisherServiceTest.cs b/src/Plugins/RabbitMQ/Tests/RabbitMqMessagePublisherServiceTest.cs
@@ -31,15 +31,15 @@ public class RabbitMQMessagePublisherServiceTest
         private readonly Mock<ILogger<RabbitMQMessagePublisherService>> _logger;
         private readonly Mock<IRabbitMQConnectionFactory> _connectionFactory;
         private readonly Mock<IModel> _model;
-        private static readonly object mutex = new();
+        private static readonly object Mutex = new();
         public RabbitMQMessagePublisherServiceTest()
         {
             _options = Options.Create(new MessageBrokerServiceConfiguration());
             _logger = new Mock<ILogger<RabbitMQMessagePublisherService>>();
             _connectionFactory = new Mock<IRabbitMQConnectionFactory>();
             _model = new Mock<IModel>();
 
-            _connectionFactory.Setup(p => p.CreateChannel(It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>()))
+            _connectionFactory.Setup(p => p.CreateChannel(It.IsAny<ChannelType>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>()))
                 .Returns(_model.Object);
         }
 
@@ -117,7 +117,7 @@ public async Task IntegrationTestRabbitPublish()
             var pubService = new RabbitMQMessagePublisherService(Options.Create(options), new Mock<ILogger<RabbitMQMessagePublisherService>>().Object, connectionFactory);
             var subService = new RabbitMQMessageSubscriberService(Options.Create(options), new Mock<ILogger<RabbitMQMessageSubscriberService>>().Object, connectionFactory);
 
-            var count = 10000;
+            var count = 100;
             var subRecievedCount = 0;
             var skipped = 0;
 
@@ -139,7 +139,7 @@ await Task.Run(async () =>
                     try
                     {
                         subService.Acknowledge(args.Message);
-                        lock (mutex)
+                        lock (Mutex)
                             subRecievedCount++;
                     }
                     catch (Exception)
@@ -163,15 +163,15 @@ await Task.Run(async () =>
                 Assert.Equal(HealthStatus.Healthy, result1.Status);
             }
 
-            await Task.Delay(5000);
+            await Task.Delay(15000);
 
             result1 = await hc1.CheckHealthAsync(new HealthCheckContext());
             Assert.Equal(HealthStatus.Healthy, result1.Status);
 
             result1 = await hc2.CheckHealthAsync(new HealthCheckContext());
             Assert.Equal(HealthStatus.Healthy, result1.Status);
 
-            Assert.Equal((count * 2) -skipped, subRecievedCount);
+            Assert.Equal((count * 2) - skipped, subRecievedCount);
 
         }
         private async Task<int> PublishMessage(RabbitMQMessagePublisherService pubService, string topic, Message message)
diff --git a/src/Plugins/RabbitMQ/Tests/RabbitMqMessageSubscriberServiceTest.cs b/src/Plugins/RabbitMQ/Tests/RabbitMqMessageSubscriberServiceTest.cs
@@ -40,7 +40,7 @@ public RabbitMQMessageSubscriberServiceTest()
             _connectionFactory = new Mock<IRabbitMQConnectionFactory>();
             _model = new Mock<IModel>();
 
-            _connectionFactory.Setup(p => p.CreateChannel(It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>()))
+            _connectionFactory.Setup(p => p.CreateChannel(It.IsAny<ChannelType>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>()))
                 .Returns(_model.Object);
         }
 

Original file line number	Diff line number	Diff line change
`@@ -32,6 +32,6 @@ public interface IRabbitMQConnectionFactory`
`32`	`32`	`/// <param name="useSSL">Encrypt communication</param>`
`33`	`33`	`/// <param name="portNumber">Port Number</param>`
`34`	`34`	`/// <returns>Instance of <see cref="IModel"/>.</returns>`
`35`		`- IModel? CreateChannel( string hostName, string username, string password, string virtualHost, string useSSL, string portNumber);`
	`35`	`+ IModel? CreateChannel( ChannelType type, string hostName, string username, string password, string virtualHost, string useSSL, string portNumber);`
`36`	`36`	`}`
`37`	`37`	`}`
Original file line number	Diff line number	Diff line change
`@@ -41,17 +41,17 @@ public RabbitMQConnectionFactory(ILogger<RabbitMQConnectionFactory> logger)`
`41`	`41`	`}`
`42`	`42`
`43`	`43`
`44`		`- public IModel CreateChannel(string hostName, string username, string password, string virtualHost, string useSSL, string portNumber)`
	`44`	`+ public IModel CreateChannel(ChannelType type, string hostName, string username, string password, string virtualHost, string useSSL, string portNumber)`
`45`	`45`	`{`
`46`	`46`	`Guard.Against.NullOrWhiteSpace(hostName);`
`47`	`47`	`Guard.Against.NullOrWhiteSpace(username);`
`48`	`48`	`Guard.Against.NullOrWhiteSpace(password);`
`49`	`49`	`Guard.Against.NullOrWhiteSpace(virtualHost);`
`50`	`50`
`51`		`- var key = $"{hostName}{username}{HashPassword(password)}{virtualHost}";`
	`51`	`+ var key = $"{type}{hostName}{username}{HashPassword(password)}{virtualHost}";`
`52`	`52`
`53`	`53`	`var connection = _connections.AddOrUpdate(key,`
`54`		`- x => MakeConnection(hostName, username, password, virtualHost, key, useSSL, portNumber),`
	`54`	`+ x => MakeConnection(type, hostName, username, password, virtualHost, key, useSSL, portNumber),`
`55`	`55`	`(updateKey, updateConnection) =>`
`56`	`56`	`{`
`57`	`57`	`// If connection to RMQ is lost and:`
`@@ -62,13 +62,13 @@ public IModel CreateChannel(string hostName, string username, string password, s`
`62`	`62`	`{`
`63`	`63`	`if (updateConnection.model.IsClosed)`
`64`	`64`	`{`
`65`		`- updateConnection.model = MakeChannel(updateConnection.connection, key);`
	`65`	`+ updateConnection.model = MakeChannel(type, updateConnection.connection, key);`
`66`	`66`	`}`
`67`	`67`	`return updateConnection;`
`68`	`68`	`}`
`69`	`69`	`else`
`70`	`70`	`{`
`71`		`- return MakeConnection(hostName, username, password, virtualHost, key, useSSL, portNumber);`
	`71`	`+ return MakeConnection(type, hostName, username, password, virtualHost, key, useSSL, portNumber);`
`72`	`72`	`}`
`73`	`73`	`});`
`74`	`74`
`@@ -87,18 +87,22 @@ private void OnException(CallbackExceptionEventArgs args, IConnection value, str`
`87`	`87`
`88`	`88`	`}`
`89`	`89`
`90`		`- private (IConnection, IModel) MakeConnection(string hostName, string username, string password, string virtualHost, string key, string useSSL, string portNumber)`
	`90`	`+ private (IConnection, IModel) MakeConnection(ChannelType type, string hostName, string username, string password, string virtualHost, string key, string useSSL, string portNumber)`
`91`	`91`	`{`
`92`	`92`	`var connection = CreateConnectionOnly(hostName, username, password, virtualHost, key, useSSL, portNumber);`
`93`		`- var model = MakeChannel(connection, key);`
	`93`	`+ var model = MakeChannel(type, connection, key);`
`94`	`94`	`return (connection, model);`
`95`	`95`	`}`
`96`	`96`
`97`		`- private IModel MakeChannel(IConnection connection, string key)`
	`97`	`+ private IModel MakeChannel(ChannelType type, IConnection connection, string key)`
`98`	`98`	`{`
`99`	`99`	`var model = connection.CreateModel();`
`100`	`100`	`model.CallbackException += (sender, args) => OnException(args, connection, key);`
`101`	`101`	`model.ModelShutdown += (sender, args) => ConnectionShutdown(args, connection, key);`
	`102`	`+ if (type == ChannelType.Publisher)`
	`103`	`+ {`
	`104`	`+ model.ConfirmSelect();`
	`105`	`+ }`
`102`	`106`	`return model;`
`103`	`107`	`}`
`104`	`108`
Original file line number	Diff line number	Diff line change
`@@ -47,6 +47,7 @@ public RabbitMQHealthCheck(`
`47`	`47`	`try`
`48`	`48`	`{`
`49`	`49`	`var channel = _connectionFactory.CreateChannel(`
	`50`	`+ ChannelType.Subscriber,`
`50`	`51`	`_options[ConfigurationKeys.EndPoint],`
`51`	`52`	`_options[ConfigurationKeys.Username],`
`52`	`53`	`_options[ConfigurationKeys.Password],`
Original file line number	Diff line number	Diff line change
`@@ -108,7 +108,7 @@ private void CreateChannel()`
`108`	`108`	`.Execute(() =>`
`109`	`109`	`{`
`110`	`110`	`_logger.ConnectingToRabbitMQ(Name, _endpoint, _virtualHost);`
`111`		`- _channel = _rabbitMqConnectionFactory.CreateChannel(_endpoint, _username, _password, _virtualHost, _useSSL, _portNumber);`
	`111`	`+ _channel = _rabbitMqConnectionFactory.CreateChannel(ChannelType.Subscriber, _endpoint, _username, _password, _virtualHost, _useSSL, _portNumber);`
`112`	`112`	`_channel.ExchangeDeclare(_exchange, ExchangeType.Topic, durable: true, autoDelete: false);`
`113`	`113`	`_channel.ExchangeDeclare(_deadLetterExchange, ExchangeType.Topic, durable: true, autoDelete: false);`
`114`	`114`	`_channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);`
Original file line number	Diff line number	Diff line change
`@@ -40,7 +40,7 @@ public RabbitMQMessageSubscriberServiceTest()`
`40`	`40`	`_connectionFactory = new Mock<IRabbitMQConnectionFactory>();`
`41`	`41`	`_model = new Mock<IModel>();`
`42`	`42`
`43`		`- _connectionFactory.Setup(p => p.CreateChannel(It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>()))`
	`43`	`+ _connectionFactory.Setup(p => p.CreateChannel(It.IsAny<ChannelType>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>()))`
`44`	`44`	`.Returns(_model.Object);`
`45`	`45`	`}`
`46`	`46`