Add SubscriberOptions

Author: ejsmith

samples/CleanArchitectureSample/src/Api/Handlers/ClientEventStreamHandler.cs

@@ -28,7 +28,7 @@ public async IAsyncEnumerable<ClientEvent> Handle(
         GetEventStream message,
         [EnumeratorCancellation] CancellationToken cancellationToken)
     {
-        await foreach (var evt in mediator.SubscribeAsync<IDispatchToClient>(cancellationToken: cancellationToken))
+        await foreach (var evt in mediator.SubscribeAsync<IDispatchToClient>(cancellationToken))
         {
             yield return new ClientEvent(evt.GetType().Name, evt);
         }

src/Foundatio.Mediator.Abstractions/HandlerRegistry.cs

@@ -430,22 +430,20 @@ public bool HasSubscribers
     /// The notification type to subscribe to. Can be a concrete type, base class, or interface.
     /// Messages are matched using <see cref="Type.IsAssignableFrom"/>.
     /// </typeparam>
-    /// <param name="maxCapacity">
-    /// Maximum number of items buffered per subscriber. When full, the oldest item is dropped.
-    /// Default is 100.
-    /// </param>
     /// <param name="cancellationToken">Token that ends the subscription when cancelled.</param>
+    /// <param name="options">Optional settings controlling buffer capacity and other subscription behavior.</param>
     /// <returns>An async stream of matching notifications.</returns>
     public async IAsyncEnumerable<T> SubscribeAsync<T>(
-        int maxCapacity = 100,
-        [EnumeratorCancellation] CancellationToken cancellationToken = default)
+        [EnumeratorCancellation] CancellationToken cancellationToken = default,
+        SubscriberOptions? options = null)
     {
         if (_disposed)
             throw new ObjectDisposedException(nameof(HandlerRegistry));
 
+        var maxCapacity = options?.MaxCapacity ?? 100;
         var channel = Channel.CreateBounded<T>(new BoundedChannelOptions(maxCapacity)
         {
-            FullMode = BoundedChannelFullMode.DropOldest,
+            FullMode = options?.FullMode ?? BoundedChannelFullMode.DropOldest,
             SingleWriter = false,
             SingleReader = true
         });

src/Foundatio.Mediator.Abstractions/IMediator.cs

@@ -128,11 +128,8 @@ public interface IMediator
     /// The notification type to subscribe to. Can be a concrete type, base class, or interface.
     /// Messages are matched at runtime using <see cref="Type.IsAssignableFrom"/>.
     /// </typeparam>
-    /// <param name="maxCapacity">
-    /// Maximum number of items buffered per subscriber. When full, the oldest item is dropped.
-    /// Default is 100.
-    /// </param>
     /// <param name="cancellationToken">Token that ends the subscription when cancelled.</param>
+    /// <param name="options">Optional settings controlling buffer capacity and other subscription behavior.</param>
     /// <returns>An async stream of matching notifications.</returns>
     /// <remarks>
     /// Typical usage is in streaming SSE endpoints that push domain events to connected clients:
@@ -141,9 +138,9 @@ public interface IMediator
     ///     GetEventStream message,
     ///     CancellationToken ct)
     /// {
-    ///     return mediator.SubscribeAsync&lt;INotification&gt;(cancellationToken: ct);
+    ///     return mediator.SubscribeAsync&lt;INotification&gt;(ct);
     /// }
     /// </code>
     /// </remarks>
-    IAsyncEnumerable<T> SubscribeAsync<T>(int maxCapacity = 100, CancellationToken cancellationToken = default);
+    IAsyncEnumerable<T> SubscribeAsync<T>(CancellationToken cancellationToken = default, SubscriberOptions? options = null);
 }

src/Foundatio.Mediator.Abstractions/Mediator.cs

@@ -83,8 +83,8 @@ public ValueTask PublishAsync(object message, CancellationToken cancellationToke
     }
 
     /// <inheritdoc />
-    public IAsyncEnumerable<T> SubscribeAsync<T>(int maxCapacity = 100, CancellationToken cancellationToken = default)
+    public IAsyncEnumerable<T> SubscribeAsync<T>(CancellationToken cancellationToken = default, SubscriberOptions? options = null)
     {
-        return _registry.SubscribeAsync<T>(maxCapacity, cancellationToken);
+        return _registry.SubscribeAsync<T>(cancellationToken, options);
     }
 }

src/Foundatio.Mediator.Abstractions/SubscriberOptions.cs

@@ -0,0 +1,22 @@
+using System.Threading.Channels;
+
+namespace Foundatio.Mediator;
+
+/// <summary>
+/// Options for configuring a dynamic subscription created via
+/// <see cref="IMediator.SubscribeAsync{T}"/>.
+/// </summary>
+public class SubscriberOptions
+{
+    /// <summary>
+    /// Maximum number of items buffered per subscriber. When full, the behavior is
+    /// determined by <see cref="FullMode"/>. Default is 100.
+    /// </summary>
+    public int MaxCapacity { get; set; } = 100;
+
+    /// <summary>
+    /// The behavior when the buffer is full and a new item arrives.
+    /// Default is <see cref="BoundedChannelFullMode.DropOldest"/>.
+    /// </summary>
+    public BoundedChannelFullMode FullMode { get; set; } = BoundedChannelFullMode.DropOldest;
+}

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

@@ -41,7 +41,7 @@ public async Task SubscribeAsync_ReceivesConcrete_WhenPublished()
 
         var subscriberTask = Task.Run(async () =>
         {
-            await foreach (var item in mediator.SubscribeAsync<TestEvent>(cancellationToken: cts.Token))
+            await foreach (var item in mediator.SubscribeAsync<TestEvent>(cts.Token))
             {
                 received.Add(item);
             }
@@ -77,7 +77,7 @@ public async Task SubscribeAsync_ReceivesInterface_WhenDerivedTypePublished()
 
         var subscriberTask = Task.Run(async () =>
         {
-            await foreach (var item in mediator.SubscribeAsync<ITestEvent>(cancellationToken: cts.Token))
+            await foreach (var item in mediator.SubscribeAsync<ITestEvent>(cts.Token))
             {
                 received.Add(item);
             }
@@ -113,7 +113,7 @@ public async Task SubscribeAsync_DoesNotReceive_UnrelatedTypes()
 
         var subscriberTask = Task.Run(async () =>
         {
-            await foreach (var item in mediator.SubscribeAsync<TestEvent>(cancellationToken: cts.Token))
+            await foreach (var item in mediator.SubscribeAsync<TestEvent>(cts.Token))
             {
                 received.Add(item);
             }
@@ -149,13 +149,13 @@ public async Task SubscribeAsync_MultipleSubscribers_AllReceive()
 
         var sub1 = Task.Run(async () =>
         {
-            await foreach (var item in mediator.SubscribeAsync<TestEvent>(cancellationToken: cts.Token))
+            await foreach (var item in mediator.SubscribeAsync<TestEvent>(cts.Token))
                 received1.Add(item);
         });
 
         var sub2 = Task.Run(async () =>
         {
-            await foreach (var item in mediator.SubscribeAsync<TestEvent>(cancellationToken: cts.Token))
+            await foreach (var item in mediator.SubscribeAsync<TestEvent>(cts.Token))
                 received2.Add(item);
         });
 
@@ -192,7 +192,7 @@ public async Task SubscribeAsync_CancellationEndsStream()
 
         var subscriberTask = Task.Run(async () =>
         {
-            await foreach (var item in mediator.SubscribeAsync<TestEvent>(cancellationToken: cts.Token))
+            await foreach (var item in mediator.SubscribeAsync<TestEvent>(cts.Token))
             {
                 received.Add(item);
                 if (received.Count == 2)
@@ -233,7 +233,7 @@ public async Task SubscribeAsync_DropsOldest_WhenBufferFull()
         // Use a tiny buffer (capacity 2) so we can verify drop behavior.
         var subscriberTask = Task.Run(async () =>
         {
-            await foreach (var item in mediator.SubscribeAsync<TestEvent>(maxCapacity: 2, cancellationToken: cts.Token))
+            await foreach (var item in mediator.SubscribeAsync<TestEvent>(cts.Token, new SubscriberOptions { MaxCapacity = 2 }))
             {
                 received.Add(item);
             }
@@ -279,7 +279,7 @@ public async Task HasSubscribers_ReflectsActiveSubscriptions()
 
         var subscriberTask = Task.Run(async () =>
         {
-            await foreach (var _ in mediator.SubscribeAsync<TestEvent>(cancellationToken: cts.Token))
+            await foreach (var _ in mediator.SubscribeAsync<TestEvent>(cts.Token))
             { }
         });