Added IsConsuming option for exchanges. Created FluentApi methods for adding consumption/production exchanges.

antonyvorontsov · antonyvorontsov · commit 4e72d841ada2 · 2019-11-04T21:42:55.000+03:00
diff --git a/README.md b/README.md
@@ -6,8 +6,12 @@ Wrapper-library of RabbitMQ.Client with Dependency Injection infrastructure unde
 
 ## Producer
 
-First of all you have to add all service dependencies in the `ConfigureServices` method. `AddRabbitMqClient` adds `IQueueService` that can send messages and `AddExchange` configures and adds an exchange. You can add multiple exchanges but the queue service will be single (and it will be added as singleton obviously).
+There are some step that you have to get through inside the `ConfigureServices` method for basic RabbitMQ configuration. The first mandatory step is to add `IQueueService` that contains all the logic of producing and consuming messages by calling `AddRabbitMqClient` method.
+The second step is add and configure exchanges using `AddExchange`, `AddProductionExchange` and `AddConsumptionExchange` methods. Exchanges have an option (flag) are the made to consume messages or only produce them. This is an important case when you want to use multiple exchanges in your application and want to consume messages from queues binded to chosen exchanges.
+So if you want to add an exchange and only produce messages, then use `AddProductionExchange` method. If you want to use full functionality, then use `AddConsumptionExchange` method. Or you can do any of them using `AddExchange` method and passing `isConsuming` parameter. Examples are provided.
+You can add multiple exchanges but the queue service will be added as singleton.
 
+After those two steps you are good to go. Down below is an example of the basic RabbitMQ configuration.
 ```csharp
 
 public static IConfiguration Configuration { get; set; }
@@ -18,7 +22,11 @@ public void ConfigureServices(IServiceCollection services)
     var exchangeSection = Configuration.GetSection("RabbitMqExchange");
 
     services.AddRabbitMqClient(rabbitMqSection)
-        .AddExchange("exchange.name", exchangeSection);
+        .AddProductionExchange("exchange.name", exchangeSection);
+
+	// Or other way
+    // services.AddRabbitMqClient(rabbitMqSection)
+    //    .AddExchange("exchange.name", isConsuming: false, exchangeSection);
 }
 ```
 
@@ -85,7 +93,7 @@ class Program
 	const string ExchangeName = "exchange.name";
 	public static IConfiguration Configuration { get; set; }
 
-	static void Main(string[] args)
+	static void Main()
 	{
 		var builder = new ConfigurationBuilder()
 			.SetBasePath(Directory.GetCurrentDirectory())
@@ -106,7 +114,7 @@ class Program
 		var exchangeSection = Configuration.GetSection("RabbitMqExchange");
 
 		services.AddRabbitMqClient(rabbitMqSection)
-			.AddExchange("exchange.name", exchangeSection)
+			.AddConsumptionExchange("exchange.name", exchangeSection)
 			.AddMessageHandlerSingleton<CustomMessageHandler>("routing.key")
 			.AddAsyncMessageHandlerSingleton<CustomAsyncMessageHandler>("other.routing.key");
 	}
@@ -149,7 +157,7 @@ public class CustomAsyncMessageHandler : IAsyncMessageHandler
 	public async Task Handle(string message, string routingKey)
 	{
 		// Do whatever you want asynchronously!
-		_logger.LogInformation("Merry christmas!");
+		_logger.LogInformation("Merry Christmas!");
 	}
 }
 ```
@@ -198,7 +206,7 @@ public class CustomAsyncMessageHandler : IAsyncNonCyclicMessageHandler
 And you have to register those classes the same way you did with simple handlers.
 ```csharp
 services.AddRabbitMqClient(rabbitMqSection)
-	.AddExchange("exchange.name", exchangeSection)
+	.AddConsumptionExchange("exchange.name", exchangeSection)
 	.AddNonCyclicMessageHandlerSingleton<CustomMessageHandler>("routing.key")
 	.AddAsyncNonCyclicMessageHandlerSingleton<CustomAsyncMessageHandler>("other.routing.key");
 ```
diff --git a/src/Examples.ConsumerConsole/Program.cs b/src/Examples.ConsumerConsole/Program.cs
@@ -31,7 +31,7 @@ static void ConfigureServices(IServiceCollection services)
             var exchangeSection = Configuration.GetSection("RabbitMqExchange");
 
             services.AddRabbitMqClient(rabbitMqSection)
-                .AddExchange("exchange.name", exchangeSection)
+                .AddConsumptionExchange("exchange.name", exchangeSection)
                 .AddAsyncMessageHandlerSingleton<CustomAsyncMessageHandler>("routing.key")
                 .AddAsyncNonCyclicMessageHandlerSingleton<CustomAsyncNonCyclicMessageHandler>("routing.key");
         }
diff --git a/src/Examples.ConsumerHost/Program.cs b/src/Examples.ConsumerHost/Program.cs
@@ -26,7 +26,7 @@ public static async Task Main()
                   var exchangeSection = hostContext.Configuration.GetSection("RabbitMqExchange");
 
                   services.AddRabbitMqClient(rabbitMqSection)
-                      .AddExchange("exchange.name", exchangeSection)
+                      .AddConsumptionExchange("exchange.name", exchangeSection)
                       .AddMessageHandlerTransient<CustomMessageHandler>("routing.key")
                       .AddNonCyclicMessageHandlerSingleton<CustomNonCyclicMessageHandler>("routing.key");
 
diff --git a/src/Examples.Producer/Program.cs b/src/Examples.Producer/Program.cs
@@ -46,7 +46,7 @@ static void ConfigureServices(IServiceCollection services)
                 }
             };
             services.AddRabbitMqClient(rabbitMqConfiguration)
-                .AddExchange("exchange.name", exchangeOptions);
+                .AddProductionExchange("exchange.name", exchangeOptions);
         }
     }
 }
diff --git a/src/RabbitMQ.Client.Core.DependencyInjection/Configuration/RabbitMqClientOptions.cs b/src/RabbitMQ.Client.Core.DependencyInjection/Configuration/RabbitMqClientOptions.cs
@@ -9,22 +9,22 @@ public class RabbitMqClientOptions
         /// RabbitMQ server.
         /// </summary>
         public string HostName { get; set; } = "127.0.0.1";
-        
+
         /// <summary>
         /// Port.
         /// </summary>
         public int Port { get; set; } = 5672;
-        
+
         /// <summary>
         /// UserName that connects to the server.
         /// </summary>
         public string UserName { get; set; } = "guest";
-        
+
         /// <summary>
         /// Password of the chosen user.
         /// </summary>
         public string Password { get; set; } = "guest";
-        
+
         /// <summary>
         /// Virtual host.
         /// </summary>
diff --git a/src/RabbitMQ.Client.Core.DependencyInjection/Configuration/RabbitMqExchangeOptions.cs b/src/RabbitMQ.Client.Core.DependencyInjection/Configuration/RabbitMqExchangeOptions.cs
@@ -3,7 +3,7 @@
 namespace RabbitMQ.Client.Core.DependencyInjection.Configuration
 {
     /// <summary>
-    /// Echange options.
+    /// Exchange options.
     /// </summary>
     public class RabbitMqExchangeOptions
     {
@@ -28,7 +28,7 @@ public class RabbitMqExchangeOptions
         public string DeadLetterExchange { get; set; } = "default.dlx.exchange";
 
         /// <summary>
-        /// Option to requeue failed messages (once).
+        /// Option to re-queue failed messages (once).
         /// </summary>
         public bool RequeueFailedMessages { get; set; } = true;
 
diff --git a/src/RabbitMQ.Client.Core.DependencyInjection/ExchangeServiceDescriptor.cs b/src/RabbitMQ.Client.Core.DependencyInjection/ExchangeServiceDescriptor.cs
@@ -4,7 +4,7 @@
 namespace RabbitMQ.Client.Core.DependencyInjection
 {
     /// <summary>
-    /// A service extension for registrating exchange singleton "services".
+    /// A service extension for registration exchange singleton "services".
     /// </summary>
     internal class ExchangeServiceDescriptor : ServiceDescriptor
     {
diff --git a/src/RabbitMQ.Client.Core.DependencyInjection/IQueueService.cs b/src/RabbitMQ.Client.Core.DependencyInjection/IQueueService.cs
@@ -16,9 +16,9 @@ public interface IQueueService
         /// RabbitMQ channel.
         /// </summary>
         IModel Channel { get; }
-        
+
         /// <summary>
-        /// Start comsuming (getting messages).
+        /// Start consuming (getting messages).
         /// </summary>
         void StartConsuming();
 
@@ -93,7 +93,7 @@ public interface IQueueService
         /// <param name="routingKey">Routing key.</param>
         /// <param name="secondsDelay">Delay time.</param>
         void Send(byte[] bytes, IBasicProperties properties, string exchangeName, string routingKey, int secondsDelay);
-        
+
         /// <summary>
         /// Send a message asynchronously.
         /// </summary>
diff --git a/src/RabbitMQ.Client.Core.DependencyInjection/QueueService.cs b/src/RabbitMQ.Client.Core.DependencyInjection/QueueService.cs
@@ -124,7 +124,7 @@ public void Dispose()
         }
 
         /// <summary>
-        /// Start comsuming (getting messages).
+        /// Start consuming (getting messages).
         /// </summary>
         public void StartConsuming()
         {
@@ -136,7 +136,8 @@ public void StartConsuming()
             _consumer.Received += _receivedMessage;
             _consumingStarted = true;
 
-            foreach (var exchange in _exchanges)
+            var consumptionExchanges = _exchanges.Where(x => x.IsConsuming);
+            foreach (var exchange in consumptionExchanges)
             {
                 foreach (var queue in exchange.Options.Queues)
                 {
@@ -387,7 +388,7 @@ void StartClient()
                 }
                 catch (Exception exception)
                 {
-                    _logger.LogError(new EventId(), exception, $"An error occured while processing recieved message with delivery tag {@event.DeliveryTag}.");
+                    _logger.LogError(new EventId(), exception, $"An error occurred while processing received message with delivery tag {@event.DeliveryTag}.");
 
                     Channel.BasicAck(@event.DeliveryTag, false);
 
@@ -513,7 +514,7 @@ void ValidateArguments(string exchangeName, string routingKey)
             var deadLetterExchanges = _exchanges.Select(x => x.Options.DeadLetterExchange).Distinct();
             if (!_exchanges.Any(x => x.Name == exchangeName) && !deadLetterExchanges.Any(x => x == exchangeName))
             {
-                throw new ArgumentException($"Exchange {nameof(exchangeName)} has not been deaclared yet.", nameof(exchangeName));
+                throw new ArgumentException($"Exchange {nameof(exchangeName)} has not been declared yet.", nameof(exchangeName));
             }
         }
 
diff --git a/src/RabbitMQ.Client.Core.DependencyInjection/RabbitMQ.Client.Core.DependencyInjection.csproj b/src/RabbitMQ.Client.Core.DependencyInjection/RabbitMQ.Client.Core.DependencyInjection.csproj
@@ -3,7 +3,7 @@
   <PropertyGroup>
     <TargetFramework>netcoreapp3.0</TargetFramework>
     <LangVersion>latest</LangVersion>
-    <Version>3.0.1</Version>
+    <Version>3.0.2</Version>
     <PackageTags>RabbitMQ</PackageTags>
     <PackageLicenseFile>LICENSE.txt</PackageLicenseFile>
     <RepositoryUrl>https://github.com/AntonyVorontsov/RabbitMQ.Client.Core.DependencyInjection</RepositoryUrl>
@@ -22,7 +22,7 @@
     <PackageReference Include="Microsoft.Extensions.Logging" Version="3.0.0" />
     <PackageReference Include="Microsoft.Extensions.Logging.Console" Version="3.0.0" />
     <PackageReference Include="Newtonsoft.Json" Version="12.0.2" />
-    <PackageReference Include="RabbitMQ.Client" Version="5.1.1" />
+    <PackageReference Include="RabbitMQ.Client" Version="5.1.2" />
   </ItemGroup>
 
 </Project>
diff --git a/src/RabbitMQ.Client.Core.DependencyInjection/RabbitMqClientDependencyInjectionExtensions.cs b/src/RabbitMQ.Client.Core.DependencyInjection/RabbitMqClientDependencyInjectionExtensions.cs
@@ -37,7 +37,7 @@ public static IServiceCollection AddRabbitMqClient(this IServiceCollection servi
         public static IServiceCollection AddRabbitMqClient(this IServiceCollection services, RabbitMqClientOptions configuration)
         {
             services.AddLogging(options => options.AddConsole());
-            services.Configure<RabbitMqClientOptions>(opt => 
+            services.Configure<RabbitMqClientOptions>(opt =>
             {
                 opt.HostName = configuration.HostName;
                 opt.Port = configuration.Port;
@@ -53,20 +53,65 @@ public static IServiceCollection AddRabbitMqClient(this IServiceCollection servi
             return services;
         }
 
+        /// <summary>
+        /// Add consumption exchange as singleton.
+        /// Consumption exchange can be used for producing messages as well as for consuming.
+        /// </summary>
+        /// <param name="services">Service collection.</param>
+        /// <param name="exchangeName">Exchange name.</param>
+        /// <param name="configuration">Exchange configuration section.</param>
+        /// <returns>Service collection.</returns>
+        public static IServiceCollection AddConsumptionExchange(this IServiceCollection services, string exchangeName, IConfiguration configuration) =>
+            services.AddExchange(exchangeName, isConsuming: true, configuration);
+
+        /// <summary>
+        /// Add production exchange as singleton.
+        /// Production exchange is made only for producing messages into queues and cannot consume at all.
+        /// </summary>
+        /// <param name="services">Service collection.</param>
+        /// <param name="exchangeName">Exchange name.</param>
+        /// <param name="configuration">Exchange configuration section.</param>
+        /// <returns>Service collection.</returns>
+        public static IServiceCollection AddProductionExchange(this IServiceCollection services, string exchangeName, IConfiguration configuration) =>
+            services.AddExchange(exchangeName, isConsuming: false, configuration);
+
+        /// <summary>
+        /// Add consumption exchange as singleton.
+        /// Consumption exchange can be used for producing messages as well as for consuming.
+        /// </summary>
+        /// <param name="services">Service collection.</param>
+        /// <param name="exchangeName">Exchange name.</param>
+        /// <param name="options">Exchange configuration <see cref="RabbitMqExchangeOptions"/>.</param>
+        /// <returns>Service collection.</returns>
+        public static IServiceCollection AddConsumptionExchange(this IServiceCollection services, string exchangeName, RabbitMqExchangeOptions options) =>
+            services.AddExchange(exchangeName, isConsuming: true, options);
+
+        /// <summary>
+        /// Add production exchange as singleton.
+        /// Production exchange is made only for producing messages into queues and cannot consume at all.
+        /// </summary>
+        /// <param name="services">Service collection.</param>
+        /// <param name="exchangeName">Exchange name.</param>
+        /// <param name="options">Exchange configuration <see cref="RabbitMqExchangeOptions"/>.</param>
+        /// <returns>Service collection.</returns>
+        public static IServiceCollection AddProductionExchange(this IServiceCollection services, string exchangeName, RabbitMqExchangeOptions options) =>
+            services.AddExchange(exchangeName, isConsuming: false, options);
+
         /// <summary>
         /// Add exchange as singleton.
         /// </summary>
         /// <param name="services">Service collection.</param>
         /// <param name="exchangeName">Exchange name.</param>
         /// <param name="configuration">Exchange configuration section.</param>
+        /// <param name="isConsuming">Flag whether an exchange made for consumption.</param>
         /// <returns>Service collection.</returns>
-        public static IServiceCollection AddExchange(this IServiceCollection services, string exchangeName, IConfiguration configuration)
+        public static IServiceCollection AddExchange(this IServiceCollection services, string exchangeName, bool isConsuming, IConfiguration configuration)
         {
             CheckExchangeExists(services, exchangeName);
 
             var options = new RabbitMqExchangeOptions();
             configuration.Bind(options);
-            return services.AddExchange(exchangeName, options);
+            return services.AddExchange(exchangeName, isConsuming, options);
         }
 
         /// <summary>
@@ -75,13 +120,19 @@ public static IServiceCollection AddExchange(this IServiceCollection services, s
         /// <param name="services">Service collection.</param>
         /// <param name="exchangeName">Exchange name.</param>
         /// <param name="options">Exchange configuration <see cref="RabbitMqExchangeOptions"/>.</param>
+        /// <param name="isConsuming">Flag whether an exchange made for consumption.</param>
         /// <returns>Service collection.</returns>
-        public static IServiceCollection AddExchange(this IServiceCollection services, string exchangeName, RabbitMqExchangeOptions options)
+        public static IServiceCollection AddExchange(this IServiceCollection services, string exchangeName, bool isConsuming, RabbitMqExchangeOptions options)
         {
             CheckExchangeExists(services, exchangeName);
 
             var exchangeOptions = options ?? new RabbitMqExchangeOptions();
-            var exchange = new RabbitMqExchange { Name = exchangeName, Options = exchangeOptions };
+            var exchange = new RabbitMqExchange
+            {
+                Name = exchangeName,
+                IsConsuming = isConsuming,
+                Options = exchangeOptions
+            };
             var service = new ExchangeServiceDescriptor(typeof(RabbitMqExchange), exchange)
             {
                 ExchangeName = exchangeName
@@ -96,7 +147,7 @@ static void CheckExchangeExists(IServiceCollection services, string exchangeName
                               && x.Lifetime == ServiceLifetime.Singleton
                               && string.Equals(((ExchangeServiceDescriptor)x).ExchangeName, exchangeName, StringComparison.OrdinalIgnoreCase));
             if (exchangeExists)
-                throw new ArgumentException($"Exchange {exchangeName} has already been added!");
+                throw new ArgumentException($"Exchange {exchangeName} has been added already!");
         }
 
         /// <summary>
@@ -261,7 +312,7 @@ public static IServiceCollection AddAsyncNonCyclicMessageHandlerSingleton<T>(thi
 
         static IServiceCollection AddInstanceTransient<U, T>(this IServiceCollection services, IEnumerable<string> routingKeys)
             where U : class
-            where T : class, U 
+            where T : class, U
         {
             services.AddTransient<U, T>();
             var router = new MessageHandlerRouter { Type = typeof(T), RoutingKeys = routingKeys.ToList() };
diff --git a/src/RabbitMQ.Client.Core.DependencyInjection/RabbitMqExchange.cs b/src/RabbitMQ.Client.Core.DependencyInjection/RabbitMqExchange.cs
@@ -12,6 +12,12 @@ public class RabbitMqExchange
         /// </summary>
         public string Name { get; set; }
 
+        /// <summary>
+        /// Flag determining whether the exchange is made for message consumption.
+        /// If false then an exchange made only for publishing.
+        /// </summary>
+        public bool IsConsuming { get; set; }
+
         /// <summary>
         /// Exchange options.
         /// </summary>

Original file line number	Diff line number	Diff line change
`@@ -31,7 +31,7 @@ static void ConfigureServices(IServiceCollection services)`
`31`	`31`	`var exchangeSection = Configuration.GetSection("RabbitMqExchange");`
`32`	`32`
`33`	`33`	`services.AddRabbitMqClient(rabbitMqSection)`
`34`		`- .AddExchange("exchange.name", exchangeSection)`
	`34`	`+ .AddConsumptionExchange("exchange.name", exchangeSection)`
`35`	`35`	`.AddAsyncMessageHandlerSingleton<CustomAsyncMessageHandler>("routing.key")`
`36`	`36`	`.AddAsyncNonCyclicMessageHandlerSingleton<CustomAsyncNonCyclicMessageHandler>("routing.key");`
`37`	`37`	`}`
Original file line number	Diff line number	Diff line change
`@@ -46,7 +46,7 @@ static void ConfigureServices(IServiceCollection services)`
`46`	`46`	`}`
`47`	`47`	`};`
`48`	`48`	`services.AddRabbitMqClient(rabbitMqConfiguration)`
`49`		`- .AddExchange("exchange.name", exchangeOptions);`
	`49`	`+ .AddProductionExchange("exchange.name", exchangeOptions);`
`50`	`50`	`}`
`51`	`51`	`}`
`52`	`52`	`}`
Original file line number	Diff line number	Diff line change
`@@ -4,7 +4,7 @@`
`4`	`4`	`namespace RabbitMQ.Client.Core.DependencyInjection`
`5`	`5`	`{`
`6`	`6`	`/// <summary>`
`7`		`- /// A service extension for registrating exchange singleton "services".`
	`7`	`+ /// A service extension for registration exchange singleton "services".`
`8`	`8`	`/// </summary>`
`9`	`9`	`internal class ExchangeServiceDescriptor : ServiceDescriptor`
`10`	`10`	`{`
Original file line number	Diff line number	Diff line change
`@@ -124,7 +124,7 @@ public void Dispose()`
`124`	`124`	`}`
`125`	`125`
`126`	`126`	`/// <summary>`
`127`		`- /// Start comsuming (getting messages).`
	`127`	`+ /// Start consuming (getting messages).`
`128`	`128`	`/// </summary>`
`129`	`129`	`public void StartConsuming()`
`130`	`130`	`{`
`@@ -136,7 +136,8 @@ public void StartConsuming()`
`136`	`136`	`_consumer.Received += _receivedMessage;`
`137`	`137`	`_consumingStarted = true;`
`138`	`138`
`139`		`- foreach (var exchange in _exchanges)`
	`139`	`+ var consumptionExchanges = _exchanges.Where(x => x.IsConsuming);`
	`140`	`+ foreach (var exchange in consumptionExchanges)`
`140`	`141`	`{`
`141`	`142`	`foreach (var queue in exchange.Options.Queues)`
`142`	`143`	`{`
`@@ -387,7 +388,7 @@ void StartClient()`
`387`	`388`	`}`
`388`	`389`	`catch (Exception exception)`
`389`	`390`	`{`
`390`		`- _logger.LogError(new EventId(), exception, $"An error occured while processing recieved message with delivery tag {@event.DeliveryTag}.");`
	`391`	`+ _logger.LogError(new EventId(), exception, $"An error occurred while processing received message with delivery tag {@event.DeliveryTag}.");`
`391`	`392`
`392`	`393`	`Channel.BasicAck(@event.DeliveryTag, false);`
`393`	`394`
`@@ -513,7 +514,7 @@ void ValidateArguments(string exchangeName, string routingKey)`
`513`	`514`	`var deadLetterExchanges = _exchanges.Select(x => x.Options.DeadLetterExchange).Distinct();`
`514`	`515`	`if (!_exchanges.Any(x => x.Name == exchangeName) && !deadLetterExchanges.Any(x => x == exchangeName))`
`515`	`516`	`{`
`516`		`- throw new ArgumentException($"Exchange {nameof(exchangeName)} has not been deaclared yet.", nameof(exchangeName));`
	`517`	`+ throw new ArgumentException($"Exchange {nameof(exchangeName)} has not been declared yet.", nameof(exchangeName));`
`517`	`518`	`}`
`518`	`519`	`}`
`519`	`520`