[Host] AutoStartConsumersEnabled not respected for child buses

Signed-off-by: Tomasz Maruszak <[email protected]>

Author: zarusz

src/Host.Plugin.Properties.xml

@@ -4,7 +4,7 @@
   <Import Project="Common.NuGet.Properties.xml" />
 
   <PropertyGroup>
-    <Version>3.3.2-rc100</Version>
+    <Version>3.3.2-rc102</Version>
   </PropertyGroup>
 
 </Project>

src/SlimMessageBus.Host.Configuration/Settings/MessageBusSettings.cs

@@ -37,15 +37,14 @@ public IServiceProvider ServiceProvider
     /// <summary>
     /// When true will start the message consumption on consumers after the bus is created.
     /// </summary>
-    public bool AutoStartConsumers { get; set; }
+    public bool? AutoStartConsumers { get; set; }
 
     public MessageBusSettings(MessageBusSettings parent = null)
     {
         _children = [];
         Producers = [];
         Consumers = [];
         SerializerType = typeof(IMessageSerializerProvider);
-        AutoStartConsumers = true;
 
         if (parent != null)
         {

src/SlimMessageBus.Host.Configuration/SlimMessageBus.Host.Configuration.csproj

@@ -6,7 +6,7 @@
     <Description>Core configuration interfaces of SlimMessageBus</Description>
     <PackageTags>SlimMessageBus</PackageTags>
     <RootNamespace>SlimMessageBus.Host</RootNamespace>
-    <Version>3.3.1</Version>
+    <Version>3.3.2-rc102</Version>
   </PropertyGroup>
 
   <ItemGroup>

src/SlimMessageBus.Host/DependencyResolver/MessageBusHostedService.cs

@@ -5,15 +5,9 @@
 /// <summary>
 /// <see cref="IHostedService"/> responsible for starting message bus consumers.
 /// </summary>
-public class MessageBusHostedService(IConsumerControl bus, MessageBusSettings messageBusSettings) : IHostedService
+internal class MessageBusHostedService(IMasterMessageBus bus) : IHostedService
 {
-    public async Task StartAsync(CancellationToken cancellationToken)
-    {
-        if (messageBusSettings.AutoStartConsumers)
-        {
-            await bus.Start();
-        }
-    }
+    public Task StartAsync(CancellationToken cancellationToken) => bus.AutoStart(cancellationToken);
 
     public Task StopAsync(CancellationToken cancellationToken) => bus.Stop();
 }

src/SlimMessageBus.Host/DependencyResolver/ServiceCollectionExtensions.cs

@@ -73,7 +73,14 @@ public static IServiceCollection AddSlimMessageBus(this IServiceCollection servi
             var currentBusProvider = svp.GetRequiredService<ICurrentMessageBusProvider>();
             MessageBus.SetProvider(currentBusProvider.GetCurrent);
 
-            return (IMasterMessageBus)mbb.Build();
+            var messageBus = (IMasterMessageBus)mbb.Build();
+
+            // If we are not running in the hosted environment, we need to start the bus ourselves after creation
+            // See MessageBusHostedService for the hosted environment
+            // Fire and forget
+            _ = Task.Run(() => messageBus.AutoStart(default));
+
+            return messageBus;
         });
 
         services.TryAddTransient<IConsumerControl>(svp => svp.GetRequiredService<IMasterMessageBus>());

src/SlimMessageBus.Host/Hybrid/HybridMessageBus.cs

@@ -68,6 +68,9 @@ protected virtual MessageBusBase BuildBus(MessageBusBuilder builder)
         return (MessageBusBase)bus;
     }
 
+    public Task AutoStart(CancellationToken cancellationToken)
+        => Task.WhenAll(_busByName.Values.Select(x => x.AutoStart(cancellationToken)));
+
     public Task Start() =>
         Task.WhenAll(_busByName.Values.Select(x => x.Start()));
 

src/SlimMessageBus.Host/IMasterMessageBus.cs

@@ -3,4 +3,6 @@
 public interface IMasterMessageBus : IMessageBusProducer, IConsumerControl, ITopologyControl, IMessageBusProvider
 {
     IMessageSerializerProvider SerializerProvider { get; }
+
+    Task AutoStart(CancellationToken cancellationToken);
 }

src/SlimMessageBus.Host/MessageBusBase.cs

@@ -101,7 +101,7 @@ protected MessageBusBase(MessageBusSettings settings)
 
         RuntimeTypeCache = settings.ServiceProvider.GetRequiredService<RuntimeTypeCache>();
 
-        MessageBusTarget = new MessageBusProxy(this, Settings.ServiceProvider);
+        MessageBusTarget = new MessageBusProxy(this, settings.ServiceProvider);
 
         TimeProvider = settings.ServiceProvider.GetRequiredService<TimeProvider>();
 
@@ -124,25 +124,10 @@ protected void OnBuildProvider()
 
         // Notify the bus has been created - before any message can be produced
         InitTaskList.Add(() => OnBusLifecycle(MessageBusLifecycleEventType.Created), CancellationToken);
-
-        // Auto start consumers if enabled
-        if (Settings.AutoStartConsumers)
-        {
-            // Fire and forget start
-            _ = Task.Run(async () =>
-            {
-                try
-                {
-                    await Start().ConfigureAwait(false);
-                }
-                catch (Exception e)
-                {
-                    LogCouldNotStartConsumers(e);
-                }
-            });
-        }
     }
 
+
+
     protected virtual void Build()
     {
         ProducerSettingsByMessageType = new ProducerByMessageTypeCache<ProducerSettings>(_logger, BuildProducerByBaseMessageType(), RuntimeTypeCache);
@@ -178,6 +163,22 @@ private async Task OnBusLifecycle(MessageBusLifecycleEventType eventType)
         }
     }
 
+    public async virtual Task AutoStart(CancellationToken cancellationToken)
+    {
+        // Auto start consumers if enabled for this bus (check first on this bus if setting set, otherwise check parent bus, fallack to true)
+        if (Settings.AutoStartConsumers ?? Settings.Parent?.AutoStartConsumers ?? true)
+        {
+            try
+            {
+                await Start().ConfigureAwait(false);
+            }
+            catch (Exception e)
+            {
+                LogCouldNotStartConsumers(e);
+            }
+        }
+    }
+
     public async Task Start()
     {
         lock (_startLock)

src/Tests/SlimMessageBus.Host.Test/MessageBusBaseTests.cs

@@ -1,11 +1,12 @@
 namespace SlimMessageBus.Host.Test;
 
 using SlimMessageBus.Host.Collections;
+using SlimMessageBus.Host.Hybrid;
 using SlimMessageBus.Host.Test.Common;
 
 public class MessageBusBaseTests : IDisposable
 {
-    private MessageBusBuilder BusBuilder { get; }
+    private MessageBusBuilder BusBuilder { get; set; }
     private readonly Lazy<MessageBusTested> _busLazy;
     private MessageBusTested Bus => _busLazy.Value;
     private readonly DateTimeOffset _timeZero;
@@ -59,7 +60,15 @@ public MessageBusBaseTests()
                 };
             });
 
-        _busLazy = new Lazy<MessageBusTested>(() => (MessageBusTested)BusBuilder.Build());
+        _busLazy = new Lazy<MessageBusTested>(CreateMessageBus<MessageBusTested>);
+    }
+
+    private T CreateMessageBus<T>() where T : IMasterMessageBus
+    {
+        var bus = (T)BusBuilder.Build();
+        _ = Task.Run(() => bus.AutoStart(default));
+        Thread.Sleep(200);
+        return bus;
     }
 
     public void Dispose()
@@ -94,6 +103,74 @@ public void When_Create_Given_ConfigurationThatDeclaresSameMessageTypeMoreThanOn
             .WithMessage("*was declared more than once*");
     }
 
+    [Theory]
+    [InlineData(null, null, true, true)]
+    [InlineData(null, false, false, true)]
+    [InlineData(false, null, false, false)]
+    [InlineData(false, true, true, false)]
+    public async Task When_Create_Given_TwoChildBusAndOneHasAutoStartConsumersAsOff_Then_OnlyChildBusIsStarted(bool? rootBusEnabled, bool? child1Enabled, bool child1ShouldStart, bool child2ShouldStart)
+    {
+        // arrange
+        BusBuilder = MessageBusBuilder
+            .Create()
+            .WithServiceProvider(_serviceProviderMock.Object)
+            .WithProviderHybrid();
+
+        Mock<MessageBusBase> childBusMock1 = null;
+        Mock<MessageBusBase> childBusMock2 = null;
+
+        if (rootBusEnabled != null)
+        {
+            BusBuilder.AutoStartConsumersEnabled(rootBusEnabled.Value);
+        }
+
+        BusBuilder.AddChildBus("child1", mbb =>
+        {
+
+            if (child1Enabled != null)
+            {
+                mbb.AutoStartConsumersEnabled(child1Enabled.Value);
+            }
+            mbb.WithProvider((s) =>
+            {
+                var childBusSettings = new MessageBusSettings(s)
+                {
+                    Name = "child1"
+                };
+                childBusMock1 = new Mock<MessageBusBase>(childBusSettings) { CallBase = true };
+                return childBusMock1.Object;
+            });
+        });
+
+        BusBuilder.AddChildBus("child2", mbb =>
+        {
+            mbb.WithProvider((s) =>
+            {
+                var childBusSettings = new MessageBusSettings(s)
+                {
+                    Name = "child2"
+                };
+                childBusMock2 = new Mock<MessageBusBase>(childBusSettings) { CallBase = true };
+                return childBusMock2.Object;
+            });
+        });
+
+
+        // act
+        var bus = CreateMessageBus<IMasterMessageBus>();
+
+        await Task.Delay(TimeSpan.FromSeconds(2));
+
+        // assert
+        childBusMock1.Should().NotBeNull();
+        childBusMock2.Should().NotBeNull();
+
+        childBusMock1.Verify(x => x.AutoStart(It.IsAny<CancellationToken>()), Times.Once);
+        childBusMock2.Verify(x => x.AutoStart(It.IsAny<CancellationToken>()), Times.Once);
+        childBusMock1.Verify(x => x.OnStart(), child1ShouldStart ? Times.Once : Times.Never);
+        childBusMock2.Verify(x => x.OnStart(), child2ShouldStart ? Times.Once : Times.Never);
+    }
+
     [Fact]
     public async Task When_Create_Then_BusLifecycleCreatedIsSentToRegisteredInterceptors()
     {