Skip to content

feat: RabbitMQ channel pooling in RabbitMqMessageTransport #241

Description

@samtrion

User Story

As a developer using RabbitMQ with Pulse under high throughput, I want RabbitMqMessageTransport to use a channel pool instead of a single shared channel, so that concurrent publish operations are not serialized and throughput scales with connection parallelism.


Background

RabbitMqMessageTransport is registered as Singleton and lazily initializes a single IRabbitMqChannelAdapter. All concurrent SendAsync and SendBatchAsync calls share this one channel, serializing all publishes. Under batch-send scenarios or high concurrency this becomes a bottleneck. RabbitMQ channels are lightweight; pooling them allows concurrent publishes over the same connection without multiplying connections.


Requirements

  • Define IRabbitMqChannelPool interface:
    ValueTask<IRabbitMqChannelAdapter> RentAsync(CancellationToken cancellationToken);
    void Return(IRabbitMqChannelAdapter channel);
    Task<bool> IsHealthyAsync(CancellationToken cancellationToken);
  • Add int MaxChannelPoolSize { get; set; } = 10; to RabbitMqTransportOptions.
  • Implement RabbitMqChannelPool : IRabbitMqChannelPool, IDisposable:
    • ConcurrentQueue<IRabbitMqChannelAdapter> for pooled channels.
    • SemaphoreSlim(MaxChannelPoolSize) to limit concurrent checkouts.
    • RentAsync: acquire semaphore, dequeue existing channel or create new; if dequeued channel is closed, dispose it and create a new one.
    • Return: if channel is open, enqueue back; if closed, dispose and release semaphore.
    • Dispose: dispose all pooled channels.
  • Update RabbitMqMessageTransport to replace _channel / _initializationLock with IRabbitMqChannelPool; use pool.RentAsync / pool.Return in SendAsync and SendBatchAsync; always Return in a finally block.
  • Register IRabbitMqChannelPoolRabbitMqChannelPool as Singleton in RabbitMqMediatorBuilderExtensions.

Acceptance Criteria

  • RabbitMqChannelPool correctly limits concurrent checkouts to MaxChannelPoolSize.
  • A closed channel returned to the pool is disposed; a new one is created on next RentAsync.
  • Return is always called after SendAsync, even when an exception occurs.
  • RabbitMqMessageTransport no longer holds a single channel field.
  • IRabbitMqChannelPool is registered as Singleton.
  • Unit tests cover: concurrent rent, return of open channel, return of closed channel, pool exhaustion waits, dispose.

Out of Scope

  • Publisher confirms.
  • Connection-level pooling.

Metadata

Metadata

Assignees

No one assigned

    Labels

    type:featureIndicates a new feature or enhancement to be added.

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions