msg host config builder - re-use pipeline 4 more subscriber groups (#237

)

Co-authored-by: Radu Popovici <[email protected]>

src/Messaging/NBB.Messaging.Host/Builder/MessagingHostConfigurationBuilder.cs

@@ -38,8 +38,11 @@ public IMessagingHostOptionsBuilder AddSubscriberServices(Action<ITypeSourceSele
             _messageTypeProvider = subscriberServiceSelector;
             _topicProvider = subscriberServiceSelector;
 
-            _currentSubscriberGroup = new List<MessagingHostConfiguration.Subscriber>();
-             _subscriberGroups.Add(_currentSubscriberGroup);
+            if (_currentSubscriberGroup == null)
+            {
+                _currentSubscriberGroup = new List<MessagingHostConfiguration.Subscriber>();
+                _subscriberGroups.Add(_currentSubscriberGroup);
+            }
 
             return this;
         }
@@ -143,7 +146,7 @@ IMessagingHostPipelineBuilder WithOptions(
     /// <summary>
     /// Used to subscriberBuilder the messaging host pipeline
     /// </summary>
-    public interface IMessagingHostPipelineBuilder
+    public interface IMessagingHostPipelineBuilder : IMessagingHostConfigurationBuilder
     {
         /// <summary>
         /// Adds the message processing pipeline to the messaging host.
@@ -152,4 +155,4 @@ public interface IMessagingHostPipelineBuilder
         /// <returns>The messaging host subscriberBuilder to further subscriberBuilder the messaging host. It is used in the fluent API</returns>
         void UsePipeline(Action<IPipelineBuilder<MessagingContext>> configurePipeline);
     }
-}
+}

src/Messaging/NBB.Messaging.Host/README.md

@@ -73,7 +73,28 @@ services.AddMessagingHost(
         );
     }
 );
-```      
+```
+
+#### Multiple subscriber groups using the same pipeline
+
+In case you need to use the same pipeline for more subscriber groups, the builder lets you chain multiple `AddSubscriberServices(...).WithOptions(...)` function calls:
+
+```csharp
+
+services.AddMessagingHost(
+    Configuration,
+    hostBuilder =>
+    {
+        hostBuilder.Configure(configBuilder => configBuilder
+            .AddSubscriberServices(...)
+            .WithOptions(...)
+            .AddSubscriberServices(...)
+            .WithOptions(...)
+            .UsePipeline(...)
+        );
+    }
+);
+``` 
 
 #### Advanced scenarios
 There is an overload of the *Configure* method that allows async/await operations.

test/UnitTests/Messaging/NBB.Messaging.Host.Tests/MessagingHostBuilderTests.cs

@@ -5,6 +5,7 @@
 using MediatR;
 using Microsoft.Extensions.DependencyInjection;
 using Moq;
+using NBB.Core.Pipeline;
 using NBB.Messaging.Abstractions;
 using System;
 using System.Collections.Generic;
@@ -147,6 +148,29 @@ public void Should_register_handled_topics_singleton()
             config.Subscribers[0].Pipeline.Should().NotBeNull();
         }
 
+        [Fact]
+        public void Should_register_same_pipeline_for_more_subscriber_groups()
+        {
+            //Arrange
+            var services = Mock.Of<IServiceCollection>();
+            var provider = Mock.Of<IServiceProvider>();
+
+            //Act
+            var builder = new MessagingHostConfigurationBuilder(provider, services);
+            builder
+                .AddSubscriberServices(cfg => cfg.FromTopics("SomeTopicName")).WithDefaultOptions()
+                .AddSubscriberServices(cfg => cfg.FromTopics("OtherTopicName")).WithDefaultOptions()
+                .UsePipeline(_ => { });
+            var config = builder.Build();
+
+            //Assert
+            config.Subscribers.Should().HaveCount(2);
+            config.Subscribers[0].Options.Should().Be(MessagingSubscriberOptions.Default with { TopicName = "SomeTopicName" });
+            config.Subscribers[1].Options.Should().Be(MessagingSubscriberOptions.Default with { TopicName = "OtherTopicName" });
+            config.Subscribers[0].Pipeline.Should().Be(config.Subscribers[1].Pipeline);
+
+        }
+
         public record CommandMessage : IRequest;
 
         public record EventMessage : INotification;
@@ -186,4 +210,4 @@ public Task<string> Handle(QueryMessage request, CancellationToken cancellationT
             }
         }
     }
-}
+}