FromHandlerType routing: create listeners for separated handler chains, not just the saga (#3041)

src/Transports/RabbitMQ/Wolverine.RabbitMQ.Tests/ConventionalRouting/Bug_3041_saga_and_handler_with_handler_type_naming.cs

@@ -0,0 +1,83 @@
+using JasperFx.Core;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Shouldly;
+using Wolverine.ComplianceTests;
+using Wolverine.RabbitMQ.Internal;
+using Wolverine.Runtime;
+using Wolverine.Transports;
+using Wolverine.Util;
+using Xunit;
+
+namespace Wolverine.RabbitMQ.Tests.ConventionalRouting;
+
+// GH-3041: with NamingSource.FromHandlerType + MultipleHandlerBehavior.Separated, a message handled by
+// BOTH a saga (e.g. a Start method) and a regular handler only created a listener queue for the saga -
+// the regular handler's separated chain lives in chain.ByEndpoint, which the FromHandlerType listener
+// discovery ignored. Result: the regular handler never received the message.
+public class Bug_3041_saga_and_handler_with_handler_type_naming : IAsyncLifetime, IDisposable
+{
+    private IHost _host = null!;
+    private IWolverineRuntime _runtime = null!;
+
+    public async Task InitializeAsync()
+    {
+        _host = await WolverineHost.ForAsync(opts =>
+        {
+            opts.UseRabbitMq()
+                .AutoProvision()
+                .UseConventionalRouting(NamingSource.FromHandlerType);
+
+            opts.Policies.DisableConventionalLocalRouting();
+            opts.MultipleHandlerBehavior = MultipleHandlerBehavior.Separated;
+
+            opts.Discovery.DisableConventionalDiscovery()
+                .IncludeType(typeof(Bug3041Saga))
+                .IncludeType(typeof(Bug3041RegularHandler));
+        });
+
+        _runtime = _host.Services.GetRequiredService<IWolverineRuntime>();
+    }
+
+    Task IAsyncLifetime.DisposeAsync() => Task.CompletedTask;
+
+    [Fact]
+    public void both_the_saga_and_the_regular_handler_get_listener_queues()
+    {
+        var transport = _runtime.Options.RabbitMqTransport();
+        var sagaQueue = typeof(Bug3041Saga).ToMessageTypeName();
+        var handlerQueue = typeof(Bug3041RegularHandler).ToMessageTypeName();
+        var all = string.Join(", ", transport.Queues.Select(q => q.EndpointName));
+
+        transport.Queues.Contains(sagaQueue).ShouldBeTrue($"saga queue '{sagaQueue}' missing. Queues: {all}");
+        transport.Queues.Contains(handlerQueue).ShouldBeTrue($"regular-handler queue '{handlerQueue}' missing. Queues: {all}");
+    }
+
+    [Fact]
+    public void both_handler_queues_have_active_listeners()
+    {
+        var saga = $"rabbitmq://queue/{typeof(Bug3041Saga).ToMessageTypeName()}".ToUri();
+        var handler = $"rabbitmq://queue/{typeof(Bug3041RegularHandler).ToMessageTypeName()}".ToUri();
+
+        var listeners = _runtime.Endpoints.ActiveListeners().Select(x => x.Uri).ToArray();
+        listeners.ShouldContain(saga);
+        listeners.ShouldContain(handler);
+    }
+
+    public void Dispose() => _host.Dispose();
+}
+
+public record Bug3041Message;
+
+public class Bug3041Saga : Saga
+{
+    public Guid Id { get; set; }
+    public void Start(Bug3041Message message) => Id = Guid.NewGuid();
+}
+
+public class Bug3041RegularHandler
+{
+    public void Handle(Bug3041Message message)
+    {
+    }
+}

src/Wolverine/Transports/MessageRoutingConvention.cs

@@ -71,7 +71,7 @@ void IMessageRoutingConvention.DiscoverListeners(IWolverineRuntime runtime, IRea
                 continue;
             }
 
-            if (_namingSource == NamingSource.FromHandlerType && chain.Handlers.Any())
+            if (_namingSource == NamingSource.FromHandlerType && (chain.Handlers.Any() || chain.ByEndpoint.Any()))
             {
                 foreach (var handler in chain.Handlers)
                 {
@@ -82,6 +82,22 @@ void IMessageRoutingConvention.DiscoverListeners(IWolverineRuntime runtime, IRea
                         endpoint.StickyHandlers.Add(handlerType);
                     }
                 }
+
+                // Separated handler chains live in ByEndpoint, NOT chain.Handlers. When a message is
+                // handled by more than one handler under MultipleHandlerBehavior.Separated (e.g. a saga
+                // plus a regular handler), only the saga stays in chain.Handlers - the sibling handlers
+                // are moved to ByEndpoint. They each still need their own handler-named listener, or the
+                // message never reaches them and only chain.Handlers (the saga) is invoked. GH-3041.
+                foreach (var handlerChain in chain.ByEndpoint)
+                {
+                    var handlerType = handlerChain.Handlers.First().HandlerType;
+                    var endpoint = maybeCreateListenerForMessageOrHandlerType(transport, handlerType, runtime, messageType);
+                    if (endpoint != null)
+                    {
+                        handlerChain.RegisterEndpoint(endpoint);
+                        endpoint.StickyHandlers.Add(handlerType);
+                    }
+                }
             }
             else if (runtime.Options.MultipleHandlerBehavior == MultipleHandlerBehavior.ClassicCombineIntoOneLogicalHandler && chain.Handlers.Any())
             {