Fix bug with dynamic subscribers

Author: ejsmith

src/Foundatio.Mediator/HandlerGenerator.cs

@@ -1355,6 +1355,7 @@ private static void GenerateCascadingHandlerCallsForeachAwait(
             }
 
             source.AppendLine($"var cascadeHandlers_{publishItem.Name} = ((global::Foundatio.Mediator.Mediator)mediator).Registry.GetPublishHandlersForType(typeof({typeFullName}));");
+            source.AppendLine($"if (((global::Foundatio.Mediator.Mediator)mediator).Registry.HasSubscribers) ((global::Foundatio.Mediator.Mediator)mediator).Registry.TryWriteSubscription({access});");
             source.AppendLine($"for (int i_{publishItem.Name} = 0; i_{publishItem.Name} < cascadeHandlers_{publishItem.Name}.Length; i_{publishItem.Name}++)");
             source.AppendLine("{");
             source.IncrementIndent();
@@ -1396,6 +1397,7 @@ private static void GenerateCascadingHandlerCallsTaskWhenAll(
             }
 
             source.AppendLine($"var cascadeHandlers_{publishItem.Name} = ((global::Foundatio.Mediator.Mediator)mediator).Registry.GetPublishHandlersForType(typeof({typeFullName}));");
+            source.AppendLine($"if (((global::Foundatio.Mediator.Mediator)mediator).Registry.HasSubscribers) ((global::Foundatio.Mediator.Mediator)mediator).Registry.TryWriteSubscription({access});");
             source.AppendLine($"for (int i_{publishItem.Name} = 0; i_{publishItem.Name} < cascadeHandlers_{publishItem.Name}.Length; i_{publishItem.Name}++)");
             source.AppendLine("{");
             source.AppendLine($"    allCascadeTasks.Add(cascadeHandlers_{publishItem.Name}[i_{publishItem.Name}](mediator, {access}, cancellationToken));");
@@ -1435,6 +1437,7 @@ private static void GenerateCascadingHandlerCallsFireAndForget(
             }
 
             source.AppendLine($"var cascadeHandlers_{publishItem.Name} = ((global::Foundatio.Mediator.Mediator)mediator).Registry.GetPublishHandlersForType(typeof({typeFullName}));");
+            source.AppendLine($"if (((global::Foundatio.Mediator.Mediator)mediator).Registry.HasSubscribers) ((global::Foundatio.Mediator.Mediator)mediator).Registry.TryWriteSubscription({access});");
             source.AppendLine($"for (int i_{publishItem.Name} = 0; i_{publishItem.Name} < cascadeHandlers_{publishItem.Name}.Length; i_{publishItem.Name}++)");
             source.AppendLine("{");
             source.IncrementIndent();

tests/Foundatio.Mediator.Tests/Integration/E2E_SubscribeAsyncTests.cs

@@ -17,6 +17,14 @@ public record TestEvent(string Name) : ITestEvent;
     public record OtherEvent(string Name) : ITestEvent;
     public record UnrelatedEvent(string Name);
 
+    public record CascadeCommand(string Name);
+
+    public class CascadeHandler
+    {
+        public (Result, TestEvent) Handle(CascadeCommand cmd)
+            => (Result.Success(), new TestEvent(cmd.Name));
+    }
+
     /// <summary>Polls until <paramref name="condition"/> returns true or the timeout expires.</summary>
     private static async Task WaitUntilAsync(Func<bool> condition, int timeoutMs = 2000)
     {
@@ -355,4 +363,39 @@ await Assert.ThrowsAsync<ObjectDisposedException>(async () =>
             { }
         });
     }
+
+    [Fact]
+    public async Task SubscribeAsync_ReceivesCascadingEvents()
+    {
+        var services = new ServiceCollection();
+        services.AddLogging(c => c.AddTestLogger(o => o.UseOutputHelper(() => _output)));
+        services.AddMediator(b => b.AddAssembly<TestEvent>());
+
+        await using var provider = services.BuildServiceProvider();
+        var registry = provider.GetRequiredService<HandlerRegistry>();
+        var mediator = provider.GetRequiredService<IMediator>();
+
+        using var cts = new CancellationTokenSource();
+        var received = new List<TestEvent>();
+
+        var subscriberTask = Task.Run(async () =>
+        {
+            await foreach (var item in mediator.SubscribeAsync<TestEvent>(cts.Token))
+            {
+                received.Add(item);
+            }
+        });
+
+        await WaitUntilAsync(() => registry.HasSubscribers);
+
+        // Invoke a command whose handler returns a cascading TestEvent via tuple
+        await mediator.InvokeAsync<Result>(new CascadeCommand("from-cascade"));
+
+        await WaitUntilAsync(() => received.Count >= 1);
+        cts.Cancel();
+        await subscriberTask;
+
+        Assert.Single(received);
+        Assert.Equal("from-cascade", received[0].Name);
+    }
 }