CommandPool

Author: Rafal Maciag

src/MicroPlumberd.DirectConnect/MicroPlumberd.Services.Grpc.DirectConnect.csproj

@@ -25,7 +25,7 @@
   </ItemGroup>
 
   <ItemGroup>
-    <PackageReference Include="ModelingEvolution.DirectConnect" Version="1.0.7.25" />
+    <PackageReference Include="ModelingEvolution.DirectConnect" Version="1.1.0.1" />
     <PackageReference Include="System.Collections.Immutable" Version="9.0.1" />
   </ItemGroup>
 

src/MicroPlumberd.Services/CommandBus.cs

@@ -3,7 +3,6 @@
 using System.Diagnostics;
 using System.IO;
 using System.Linq;
-using System.Net;
 using System.Reflection;
 using System.Runtime.Serialization;
 using System.Security.Cryptography.X509Certificates;
@@ -17,10 +16,10 @@
 
 namespace MicroPlumberd.Services;
 
-
 class CommandBus : ICommandBus, IEventHandler
 {
     private readonly IPlumber _plumber;
+    private readonly ICommandBusPool _pool;
     private readonly ILogger<CommandBus> _log;
     private readonly string _streamIn;
     private readonly string _streamOut;
@@ -31,9 +30,10 @@ class CommandBus : ICommandBus, IEventHandler
     private readonly object _sync = new object();
     private IAsyncDisposable? _subscription;
     public Guid SessionId { get; } = Guid.NewGuid();
-    public CommandBus(IPlumber plumber, ILogger<CommandBus> log)
+    public CommandBus(IPlumber plumber, ICommandBusPool pool, ILogger<CommandBus> log)
     {
         _plumber = plumber;
+        _pool = pool;
         _log = log;
         var servicesConventions = plumber.Config.Conventions.ServicesConventions();
         _streamIn = servicesConventions.SessionInStreamFromSessionIdConvention(SessionId);
@@ -75,8 +75,8 @@ private bool TryMapEventResponse(string type, out Type t)
 
     public async Task QueueAsync(object recipientId, object command, TimeSpan? timeout = null, bool fireAndForget = true, CancellationToken token = default)
     {
-        await using CommandBus bus = new CommandBus(this._plumber, this._log);
-        await bus.SendAsync(recipientId, command, timeout ?? TimeSpan.MaxValue, fireAndForget, token);
+        using var scope = await _pool.RentScope(token);
+        await scope.SendAsync(recipientId, command, timeout ?? TimeSpan.MaxValue, fireAndForget, token);
     }
     public async Task SendAsync(object recipientId, object command, TimeSpan? timeout = null, bool fireAndForget = false, CancellationToken token = default)
     {
@@ -170,56 +170,7 @@ async Task IEventHandler.Handle(Metadata m, object ev)
     public ValueTask DisposeAsync() => _subscription?.DisposeAsync() ?? ValueTask.CompletedTask;
 }
 
-
-[AttributeUsage(AttributeTargets.Method | AttributeTargets.Class, AllowMultiple=true)]
-public class ThrowsFaultExceptionAttribute<TMessage>() : ThrowsFaultExceptionAttribute(typeof(TMessage));
-
 public abstract class ThrowsFaultExceptionAttribute(Type thrownType) : Attribute
 {
     public Type ThrownType { get; init; } = thrownType;
-}
-
-
-public class CommandExecutionResults 
-{
-    public async ValueTask<bool> Handle(Metadata m, object ev)
-    {
-        switch (ev)
-        {
-            case CommandExecuted ce:
-            {
-                IsSuccess = true;
-                IsReady.SetResult(true);
-                return true;
-            }
-            case ICommandFailedEx ef:
-            {
-                IsSuccess = false;
-                ErrorMessage = ef.Message;
-                ErrorData = ef.Fault;
-                ErrorCode = ef.Code;
-                IsReady.SetResult(true);
-                return true;
-            }
-            case ICommandFailed cf:
-            {
-                IsSuccess = false;
-                ErrorMessage = cf.Message;
-                ErrorCode = cf.Code;
-                IsReady.SetResult(true);
-                return true;
-            }
-        }
-
-        return false;
-    }
-
-    public HttpStatusCode ErrorCode { get; private set; }
-
-
-    public string ErrorMessage { get; private set; }
-    public object? ErrorData { get; private set; }
-    public bool IsSuccess { get; private set; }
-    public TaskCompletionSource<bool> IsReady { get; private set; } = new TaskCompletionSource<bool>();
-    
 }

src/MicroPlumberd.Services/CommandBusPool.cs

@@ -0,0 +1,95 @@
+using System.Collections.Concurrent;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging;
+
+namespace MicroPlumberd.Services;
+
+class CommandBusPool : IAsyncDisposable, ICommandBusPool
+{
+    private class CommandBusOwner : ICommandBusOwner
+    {
+        private readonly CommandBusPool _parent;
+        private readonly ICommandBus _commandBus;
+
+        public Task SendAsync(object recipientId, object command, TimeSpan? timeout = null, bool fireAndForget = false,
+            CancellationToken token = default)
+        {
+            return _commandBus.SendAsync(recipientId, command, timeout, fireAndForget, token);
+        }
+
+
+        internal CommandBusOwner(CommandBusPool parent, ICommandBus cb)
+        {
+            _parent = parent;
+            this._commandBus = cb;
+        }
+
+        public void Dispose()
+        {
+            _parent.Return(this);
+        }
+
+        public ICommandBus CommandBus => _commandBus;
+    }
+    private readonly IServiceProvider _sp;
+    protected readonly int _maxCount;
+    private ConcurrentStack<CommandBusOwner> _pool;
+    private SemaphoreSlim _semaphore;
+    private bool _disposed;
+    public CommandBusPool(IServiceProvider sp, int maxCount)
+    {
+        _sp = sp;
+        _maxCount = maxCount;
+    }
+
+    internal void Return(ICommandBusOwner o)
+    {
+        if (o is not CommandBusOwner cbo)
+            throw new ArgumentException();
+
+        if (_disposed)
+        {
+            cbo.CommandBus.DisposeAsync();
+            return;
+        }
+
+        _pool.Push(cbo);
+        _semaphore.Release();
+    }
+    public ICommandBusPool Init()
+    {
+        _pool = new ConcurrentStack<CommandBusOwner>(Create(number: _maxCount).Select(x=>new CommandBusOwner(this,x)));
+        _semaphore = new SemaphoreSlim(_maxCount);
+        return this;
+    }
+
+    public async ValueTask<ICommandBusOwner> RentScope(CancellationToken ct = default)
+    {
+        await _semaphore.WaitAsync(ct);
+        if (!_pool.TryPop(out var x))
+            throw new InvalidOperationException();
+        return x;
+    }
+    public virtual IEnumerable<ICommandBus> Create(int number)
+    {
+        // Command is configured to be singleton in the container.
+        var pl = _sp.GetRequiredService<IPlumber>();
+        var logger = _sp.GetRequiredService<ILogger<CommandBus>>();
+        for (int i = 0; i < number; ++i) 
+            yield return new CommandBus(pl,this, logger);
+    }
+
+    public virtual async ValueTask DisposeAsync()
+    {
+        if (_disposed) return;
+        _disposed = true;
+        
+        if (_semaphore is IAsyncDisposable semaphoreAsyncDisposable)
+            await semaphoreAsyncDisposable.DisposeAsync();
+        else
+            _semaphore.Dispose();
+
+        foreach (var i in _pool.ToArray())
+            await i.CommandBus.DisposeAsync();
+    }
+}

src/MicroPlumberd.Services/CommandBusPoolScoped.cs

@@ -0,0 +1,15 @@
+using Microsoft.Extensions.DependencyInjection;
+
+namespace MicroPlumberd.Services;
+
+sealed class CommandBusPoolScoped(IServiceProvider sp, int maxCount) : CommandBusPool(sp, maxCount)
+{
+    private readonly IServiceScope _scope = sp.CreateScope();
+
+    public override IEnumerable<ICommandBus> Create(int number)
+    {
+        for (int i = 0; i < _maxCount; ++i)
+            yield return _scope.ServiceProvider.GetRequiredService<ICommandBus>();
+    }
+
+}

src/MicroPlumberd.Services/CommandExecutionResults.cs

@@ -0,0 +1,47 @@
+using System.Net;
+
+namespace MicroPlumberd.Services;
+
+public class CommandExecutionResults 
+{
+    public async ValueTask<bool> Handle(Metadata m, object ev)
+    {
+        switch (ev)
+        {
+            case CommandExecuted ce:
+            {
+                IsSuccess = true;
+                IsReady.SetResult(true);
+                return true;
+            }
+            case ICommandFailedEx ef:
+            {
+                IsSuccess = false;
+                ErrorMessage = ef.Message;
+                ErrorData = ef.Fault;
+                ErrorCode = ef.Code;
+                IsReady.SetResult(true);
+                return true;
+            }
+            case ICommandFailed cf:
+            {
+                IsSuccess = false;
+                ErrorMessage = cf.Message;
+                ErrorCode = cf.Code;
+                IsReady.SetResult(true);
+                return true;
+            }
+        }
+
+        return false;
+    }
+
+    public HttpStatusCode ErrorCode { get; private set; }
+
+
+    public string ErrorMessage { get; private set; }
+    public object? ErrorData { get; private set; }
+    public bool IsSuccess { get; private set; }
+    public TaskCompletionSource<bool> IsReady { get; private set; } = new TaskCompletionSource<bool>();
+    
+}

src/MicroPlumberd.Services/ContainerExtensions.cs

@@ -23,27 +23,54 @@ public async Task Handle(Metadata m, object ev)
 public static class ContainerExtensions
 {
     public static IServiceCollection AddPlumberd(this IServiceCollection collection,
-        EventStoreClientSettings? settings = null, Action<IServiceProvider, IPlumberConfig>? configure = null, bool scopedCommandBus = false) =>
-        collection.AddPlumberd(sp => settings, configure, scopedCommandBus);
+        EventStoreClientSettings? settings = null, Action<IServiceProvider, IPlumberConfig>? configure = null, bool scopedCommandBus = false, int commandBusPoolSize = 64) =>
+        collection.AddPlumberd(sp => settings, configure, scopedCommandBus, commandBusPoolSize);
 
+    /// <summary>
+    /// Adds the MicroPlumberd services to the specified <see cref="IServiceCollection"/>.
+    /// </summary>
+    /// <param name="collection">
+    /// The <see cref="IServiceCollection"/> to which the services will be added.
+    /// </param>
+    /// <param name="settingsFactory">
+    /// A factory function to create <see cref="EventStoreClientSettings"/> for configuring the Event Store client.
+    /// </param>
+    /// <param name="configure">
+    /// An optional action to configure the <see cref="IPlumberConfig"/> instance.
+    /// </param>
+    /// <param name="scopedCommandBus">
+    /// A boolean value indicating whether the <see cref="ICommandBus"/> should be registered as scoped.
+    /// </param>
+    /// <param name="commandBusPoolSize">
+    /// The size of the command bus pool - it's used for QueueAsync operation on ICommandBus. Defaults to 64.
+    /// </param>
+    /// <returns>
+    /// The updated <see cref="IServiceCollection"/> with the MicroPlumberd services added.
+    /// </returns>
     public static IServiceCollection AddPlumberd(this IServiceCollection collection,
-        Func<IServiceProvider, EventStoreClientSettings> settingsFactory, Action<IServiceProvider, IPlumberConfig>? configure = null, bool scopedCommandBus = false)
+        Func<IServiceProvider, EventStoreClientSettings> settingsFactory, Action<IServiceProvider, IPlumberConfig>? configure = null, bool scopedCommandBus = false, int commandBusPoolSize=64)
     {
         collection.AddSingleton(sp => Plumber.Create(settingsFactory(sp), x =>
         {
             configure?.Invoke(sp, x);
             x.ServiceProvider = sp;
         }));
         collection.AddSingleton<StartupHealthCheck>();
-
+        
         collection.AddBackgroundServiceIfMissing<CommandHandlerService>();
         collection.AddBackgroundServiceIfMissing<EventHandlerService>();
 
         collection.TryAddSingleton(typeof(ISnapshotPolicy<>), typeof(AttributeSnaphotPolicy<>));
-        if(scopedCommandBus)
+        if (scopedCommandBus)
+        {
             collection.TryAddScoped<ICommandBus, CommandBus>();
+            collection.TryAddSingleton<ICommandBusPool>(sp => new CommandBusPoolScoped(sp, commandBusPoolSize).Init());
+        }
         else
+        {
             collection.TryAddSingleton<ICommandBus, CommandBus>();
+            collection.TryAddSingleton<ICommandBusPool>(sp => new CommandBusPool(sp, commandBusPoolSize).Init());
+        }
         collection.TryAddSingleton(typeof(IEventHandler<>), typeof(EventHandlerExecutor<>));
 
 

src/MicroPlumberd.Services/ICommandBusOwner.cs

@@ -0,0 +1,7 @@
+namespace MicroPlumberd.Services;
+
+public interface ICommandBusOwner : IDisposable
+{
+    Task SendAsync(object recipientId, object command, TimeSpan? timeout = null, bool fireAndForget = false,
+        CancellationToken token = default);
+}

src/MicroPlumberd.Services/ICommandBusPool.cs

@@ -0,0 +1,21 @@
+namespace MicroPlumberd.Services;
+
+/// <summary>
+/// Represents a pool of command buses that can be rented and returned for managing command execution.
+/// </summary>
+public interface ICommandBusPool
+{
+    /// <summary>
+    /// Rents an <see cref="ICommandBusOwner"/> instance from the pool, allowing the caller to send commands
+    /// using the rented command bus. The rented instance must be returned to the pool after use by calling dispose
+    /// </summary>
+    /// <param name="ct">A <see cref="CancellationToken"/> to observe while waiting for a command bus to become available.</param>
+    /// <returns>
+    /// A <see cref="ValueTask{TResult}"/> representing the asynchronous operation. The result contains
+    /// an <see cref="ICommandBusOwner"/> instance that can be used to send commands.
+    /// </returns>
+    /// <exception cref="InvalidOperationException">
+    /// Thrown if the pool is empty and no command bus is available for rent.
+    /// </exception>
+    ValueTask<ICommandBusOwner> RentScope(CancellationToken ct = default);
+}

src/MicroPlumberd.Services/ThrowsFaultExceptionAttribute.cs

@@ -0,0 +1,4 @@
+namespace MicroPlumberd.Services;
+
+[AttributeUsage(AttributeTargets.Method | AttributeTargets.Class, AllowMultiple=true)]
+public class ThrowsFaultExceptionAttribute<TMessage>() : ThrowsFaultExceptionAttribute(typeof(TMessage));