merge: parallel event bus dispatch with catch-all, fault isolation, and DI auto-subscription

Evolve IEventBus/EventBusService from the serialized channel model to a parallel
per-listener dispatcher: single-listener fast path, catch-all (IEvent) listeners,
per-listener fault isolation, slow-listener telemetry, delegate Subscribe, and
IDisposable unsubscribe tokens. Unify the sync/async listener split into a single
contravariant IEventListener<T>. Add DI-native auto-subscription via
RegisterEventListener<TEvent,TListener>() and a low-priority EventListenerActivator.

Author: tgiachi

samples/SquidStd.Samples.Email/Program.cs

@@ -62,7 +62,7 @@
 
 // Inbound: react to each received email on the event bus.
 var eventBus = bootstrap.Resolve<IEventBus>();
-eventBus.RegisterAsyncListener(new MailReceivedLogger());
+eventBus.RegisterListener(new MailReceivedLogger());
 
 var outgoing = new OutgoingMailMessage
 {
@@ -85,7 +85,7 @@
 await bootstrap.StopAsync();
 
 /// <summary>Logs every received email as it arrives on the event bus.</summary>
-public sealed class MailReceivedLogger : IAsyncEventListener<MailReceivedEvent>
+public sealed class MailReceivedLogger : IEventListener<MailReceivedEvent>
 {
     /// <summary>Handles a received-mail event.</summary>
     public Task HandleAsync(MailReceivedEvent eventData, CancellationToken cancellationToken)

samples/SquidStd.Samples.EventsJobsScheduling/Program.cs

@@ -20,7 +20,7 @@
 #region step-1
 
 var eventBus = bootstrap.Resolve<IEventBus>();
-eventBus.RegisterAsyncListener(new PingListener());
+eventBus.RegisterListener(new PingListener());
 await eventBus.PublishAsync(new PingEvent("hello"), CancellationToken.None);
 
 #endregion
@@ -54,7 +54,7 @@
 public sealed record PingEvent(string Message) : IEvent;
 
 /// <summary>Handles <see cref="PingEvent" />.</summary>
-public sealed class PingListener : IAsyncEventListener<PingEvent>
+public sealed class PingListener : IEventListener<PingEvent>
 {
     public Task HandleAsync(PingEvent eventData, CancellationToken cancellationToken)
     {

src/SquidStd.Abstractions/Data/Internal/Events/EventListenerRegistration.cs

@@ -0,0 +1,13 @@
+using DryIoc;
+using SquidStd.Core.Interfaces.Events;
+
+namespace SquidStd.Abstractions.Data.Internal.Events;
+
+/// <summary>
+/// A declarative event-listener registration consumed by the bootstrap activator.
+/// The <see cref="Subscribe" /> closure captures the concrete event and listener types at
+/// registration time, so subscription needs no reflection.
+/// </summary>
+/// <param name="ListenerType">The concrete listener implementation type.</param>
+/// <param name="Subscribe">Resolves the listener and subscribes it to the bus.</param>
+public sealed record EventListenerRegistration(Type ListenerType, Action<IEventBus, IResolverContext> Subscribe);

src/SquidStd.Abstractions/Extensions/Events/RegisterEventListenerExtension.cs

@@ -0,0 +1,34 @@
+using DryIoc;
+using SquidStd.Abstractions.Data.Internal.Events;
+using SquidStd.Abstractions.Extensions.Container;
+using SquidStd.Core.Interfaces.Events;
+
+namespace SquidStd.Abstractions.Extensions.Events;
+
+/// <summary>
+/// Registers event listeners for DI-native auto-subscription at bootstrap.
+/// </summary>
+public static class RegisterEventListenerExtension
+{
+    /// <summary>
+    /// Registers a listener implementation as a singleton and records it for auto-subscription.
+    /// </summary>
+    /// <typeparam name="TEvent">The event type the listener handles.</typeparam>
+    /// <typeparam name="TListener">The listener implementation type.</typeparam>
+    /// <param name="container">The DryIoc container.</param>
+    /// <returns>The same container for chaining.</returns>
+    public static IContainer RegisterEventListener<TEvent, TListener>(this IContainer container)
+        where TEvent : IEvent
+        where TListener : class, IEventListener<TEvent>
+    {
+        container.Register<TListener>(Reuse.Singleton);
+        container.AddToRegisterTypedList(
+            new EventListenerRegistration(
+                typeof(TListener),
+                (bus, resolver) => bus.RegisterListener<TEvent>(resolver.Resolve<TListener>())
+            )
+        );
+
+        return container;
+    }
+}

src/SquidStd.Core/Data/Events/EventBusOptions.cs

@@ -0,0 +1,12 @@
+namespace SquidStd.Core.Data.Events;
+
+/// <summary>
+/// Options controlling event bus dispatch behavior.
+/// </summary>
+public sealed record EventBusOptions
+{
+    /// <summary>
+    /// Listeners whose handling exceeds this duration are logged as slow. Defaults to 100ms.
+    /// </summary>
+    public TimeSpan SlowListenerThreshold { get; init; } = TimeSpan.FromMilliseconds(100);
+}

src/SquidStd.Core/Interfaces/Events/IEventBus.cs

@@ -1,37 +1,39 @@
 namespace SquidStd.Core.Interfaces.Events;
 
 /// <summary>
-/// Dispatches synchronous and asynchronous events to registered listeners.
+/// In-process event bus that dispatches events to registered listeners in parallel.
 /// </summary>
 public interface IEventBus
 {
     /// <summary>
-    /// Dispatches an event to synchronous listeners and waits until every listener has completed.
+    /// Publishes an event and blocks until every listener has completed.
     /// </summary>
     /// <param name="eventData">The event payload.</param>
     /// <typeparam name="TEvent">The event type.</typeparam>
     void Publish<TEvent>(TEvent eventData) where TEvent : IEvent;
 
     /// <summary>
-    /// Dispatches an event to asynchronous listeners and waits until every listener has completed.
+    /// Publishes an event and completes when every listener has finished.
     /// </summary>
     /// <param name="eventData">The event payload.</param>
     /// <param name="cancellationToken">The cancellation token.</param>
     /// <typeparam name="TEvent">The event type.</typeparam>
-    /// <returns>A task that completes after all asynchronous listeners finish.</returns>
-    Task PublishAsync<TEvent>(TEvent eventData, CancellationToken cancellationToken) where TEvent : IEvent;
+    /// <returns>A task that completes after all listeners finish.</returns>
+    Task PublishAsync<TEvent>(TEvent eventData, CancellationToken cancellationToken = default) where TEvent : IEvent;
 
     /// <summary>
-    /// Registers an asynchronous listener for the specified event type.
+    /// Registers a listener for the specified event type. Dispose the returned token to unsubscribe.
     /// </summary>
-    /// <param name="listener">Listener that handles published events asynchronously.</param>
+    /// <param name="listener">The listener to register.</param>
     /// <typeparam name="TEvent">The event type.</typeparam>
-    void RegisterAsyncListener<TEvent>(IAsyncEventListener<TEvent> listener) where TEvent : IEvent;
+    /// <returns>A token that unsubscribes the listener when disposed.</returns>
+    IDisposable RegisterListener<TEvent>(IEventListener<TEvent> listener) where TEvent : IEvent;
 
     /// <summary>
-    /// Registers a synchronous listener for the specified event type.
+    /// Subscribes a delegate handler for the specified event type. Dispose the returned token to unsubscribe.
     /// </summary>
-    /// <param name="listener">Listener that handles published events.</param>
+    /// <param name="handler">The handler invoked for each published event.</param>
     /// <typeparam name="TEvent">The event type.</typeparam>
-    void RegisterListener<TEvent>(ISyncEventListener<TEvent> listener) where TEvent : IEvent;
+    /// <returns>A token that unsubscribes the handler when disposed.</returns>
+    IDisposable Subscribe<TEvent>(Func<TEvent, CancellationToken, Task> handler) where TEvent : IEvent;
 }

…Interfaces/Events/IAsyncEventListener.cs …Core/Interfaces/Events/IEventListener.cssrc/SquidStd.Core/Interfaces/Events/IAsyncEventListener.cs renamed to src/SquidStd.Core/Interfaces/Events/IEventListener.cs src/SquidStd.Core/Interfaces/Events/IAsyncEventListener.cs renamed to src/SquidStd.Core/Interfaces/Events/IEventListener.cs

@@ -1,16 +1,16 @@
 namespace SquidStd.Core.Interfaces.Events;
 
 /// <summary>
-/// Handles an event asynchronously.
+/// Handles an event published on the SquidStd event bus.
 /// </summary>
-/// <typeparam name="TEvent">The event type.</typeparam>
-public interface IAsyncEventListener<in TEvent> where TEvent : IEvent
+/// <typeparam name="TEvent">The event type handled by the listener.</typeparam>
+public interface IEventListener<in TEvent> where TEvent : IEvent
 {
     /// <summary>
-    /// Handles the event.
+    /// Handles a published event.
     /// </summary>
     /// <param name="eventData">The event payload.</param>
     /// <param name="cancellationToken">The cancellation token.</param>
     /// <returns>A task that completes when the listener finishes handling the event.</returns>
-    Task HandleAsync(TEvent eventData, CancellationToken cancellationToken);
+    Task HandleAsync(TEvent eventData, CancellationToken cancellationToken = default);
 }

src/SquidStd.Core/Interfaces/Events/ISyncEventListener.cs

@@ -4,6 +4,7 @@
 using SquidStd.Abstractions.Extensions.Container;
 using SquidStd.Abstractions.Extensions.Services;
 using SquidStd.Core.Data.Bootstrap;
+using SquidStd.Core.Data.Events;
 using SquidStd.Core.Data.Jobs;
 using SquidStd.Core.Data.Metrics;
 using SquidStd.Core.Data.Storage;
@@ -114,7 +115,19 @@ public IContainer RegisterDefaultCoreConfigSections()
         /// </summary>
         /// <returns>The same container for chaining.</returns>
         public IContainer RegisterEventBusService()
-            => container.RegisterStdService<IEventBus, EventBusService>(-1);
+        {
+            container.RegisterInstance(new EventBusOptions());
+            container.RegisterDelegate<IEventBus>(
+                resolver => new EventBusService(resolver.Resolve<EventBusOptions>()),
+                Reuse.Singleton
+            );
+            container.AddToRegisterTypedList(
+                new ServiceRegistrationData(typeof(IEventBus), typeof(EventBusService), -1)
+            );
+            container.RegisterStdService<EventListenerActivator, EventListenerActivator>(-900);
+
+            return container;
+        }
 
         /// <summary>
         /// Registers the default job system service in the container.