Merge pull request #137 from cnblogs/add-maximum-buffer-size

feat: add maximum buffer size option

Author: ikesnowy

src/Cnblogs.Architecture.Ddd.EventBus.Abstractions/EventBufferOverflowException.cs

@@ -0,0 +1,15 @@
+namespace Cnblogs.Architecture.Ddd.EventBus.Abstractions;
+
+/// <summary>
+///     The exception that is thrown when <see cref="IEventBuffer"/> reaches its maximum capacity configured in <see cref="EventBusOptions.MaximumBufferSize"/>.
+/// </summary>
+public sealed class EventBufferOverflowException : Exception
+{
+    /// <summary>
+    ///     Creates an <see cref="EventBufferOverflowException"/>.
+    /// </summary>
+    public EventBufferOverflowException()
+        : base("Event buffer reached its maximum capacity")
+    {
+    }
+}

src/Cnblogs.Architecture.Ddd.EventBus.Abstractions/EventBusOptions.cs

@@ -17,6 +17,11 @@ public class EventBusOptions
     /// </summary>
     public int? MaximumBatchSize { get; set; }
 
+    /// <summary>
+    ///     Maximum number of events that can be stored in buffer. An <see cref="EventBufferOverflowException"/> would be thrown when the number of events in buffer exceeds this limit. Pass <c>null</c> to disable limit. Defaults to <c>null</c>.
+    /// </summary>
+    public int? MaximumBufferSize { get; set; }
+
     /// <summary>
     ///     The maximum number of failure before downgrade. Defaults to 5.
     /// </summary>

src/Cnblogs.Architecture.Ddd.EventBus.Abstractions/IEventBuffer.cs

@@ -16,6 +16,7 @@ public interface IEventBuffer
     /// <param name="name">The name of integration event.</param>
     /// <param name="event">The event.</param>
     /// <typeparam name="TEvent">The type of integration event.</typeparam>
+    /// <exception cref="EventBufferOverflowException">Throws when the number of events in buffer exceeds <see cref="EventBusOptions.MaximumBufferSize"/>.</exception>
     void Add<TEvent>(string name, TEvent @event)
         where TEvent : IntegrationEvent;
 

src/Cnblogs.Architecture.Ddd.EventBus.Abstractions/InMemoryEventBuffer.cs

@@ -1,4 +1,5 @@
 using System.Collections.Concurrent;
+using Microsoft.Extensions.Options;
 
 namespace Cnblogs.Architecture.Ddd.EventBus.Abstractions;
 
@@ -8,6 +9,16 @@ namespace Cnblogs.Architecture.Ddd.EventBus.Abstractions;
 public class InMemoryEventBuffer : IEventBuffer
 {
     private readonly ConcurrentQueue<BufferedIntegrationEvent> _queue = new();
+    private readonly EventBusOptions _options;
+
+    /// <summary>
+    ///     Creates an <see cref="InMemoryEventBuffer"/>.
+    /// </summary>
+    /// <param name="options">The Eventbus options.</param>
+    public InMemoryEventBuffer(IOptions<EventBusOptions> options)
+    {
+        _options = options.Value;
+    }
 
     /// <inheritdoc />
     public int Count => _queue.Count;
@@ -16,6 +27,11 @@ public class InMemoryEventBuffer : IEventBuffer
     public void Add<TEvent>(string name, TEvent @event)
         where TEvent : IntegrationEvent
     {
+        if (_queue.Count >= _options.MaximumBufferSize)
+        {
+            throw new EventBufferOverflowException();
+        }
+
         _queue.Enqueue(new BufferedIntegrationEvent(name, @event));
     }
 

test/Cnblogs.Architecture.IntegrationTests/IntegrationEventPublishTests.cs

@@ -1,3 +1,4 @@
+using System.Net;
 using System.Net.Http.Json;
 using Cnblogs.Architecture.Ddd.EventBus.Abstractions;
 using Cnblogs.Architecture.IntegrationTestProject;
@@ -81,6 +82,76 @@ public async Task EventBus_Downgrading_DowngradeAsync()
             Times.Exactly(2));
     }
 
+    [Fact]
+    public async Task EventBus_MaximumBufferSizeReached_ThrowAsync()
+    {
+        // Arrange
+        const string data = "hello";
+        var builder = new WebApplicationFactory<Program>();
+        var eventBusMock = new Mock<IEventBusProvider>();
+        builder = builder.WithWebHostBuilder(
+            b => b.ConfigureServices(
+                services =>
+                {
+                    services.RemoveAll<IEventBusProvider>();
+                    services.AddScoped<IEventBusProvider>(_ => eventBusMock.Object);
+                    services.Configure<EventBusOptions>(
+                        o =>
+                        {
+                            o.MaximumBufferSize = 1;
+                            o.FailureCountBeforeDowngrade = 1;
+                            o.DowngradeInterval = 3000;
+                        });
+                }));
+        eventBusMock.Setup(x => x.PublishAsync(It.IsAny<string>(), It.IsAny<IntegrationEvent>()))
+            .ThrowsAsync(new InvalidOperationException());
+        var client = builder.CreateClient();
+        await client.PostAsJsonAsync(
+            "/api/v1/strings",
+            new CreatePayload(false, data));
+
+        // Act
+        var response = await client.PostAsJsonAsync("/api/v1/strings", new CreatePayload(false, data));
+
+        // Assert
+        response.Should().HaveStatusCode(HttpStatusCode.InternalServerError);
+    }
+
+    [Fact]
+    public async Task EventBus_MaximumBatchSize_OneBatchAsync()
+    {
+        // Arrange
+        const string data = "hello";
+        var builder = new WebApplicationFactory<Program>();
+        var eventBusMock = new Mock<IEventBusProvider>();
+        builder = builder.WithWebHostBuilder(
+            b => b.ConfigureServices(
+                services =>
+                {
+                    services.RemoveAll<IEventBusProvider>();
+                    services.AddScoped<IEventBusProvider>(_ => eventBusMock.Object);
+                    services.Configure<EventBusOptions>(
+                        o =>
+                        {
+                            o.MaximumBatchSize = 1;
+                            o.FailureCountBeforeDowngrade = 1;
+                            o.DowngradeInterval = 3000;
+                        });
+                }));
+        var client = builder.CreateClient();
+        for (var i = 0; i < 3; i++)
+        {
+            // put 3 events
+            await client.PostAsJsonAsync("/api/v1/strings", new CreatePayload(false, data));
+        }
+
+        // Act
+        await Task.Delay(1000);
+
+        // Assert
+        eventBusMock.Verify(x => x.PublishAsync(It.IsAny<string>(), It.IsAny<IntegrationEvent>()), Times.Once);
+    }
+
     [Fact]
     public async Task EventBus_DowngradeThenRecover_RecoverAsync()
     {