AwadyLab.Mediator.Redis 1.7.2

AwadyLab.Mediator.Redis

Redis Streams queue and distributed pub/sub broker transport for AwadyLab.Mediator.

Overview

AwadyLab.Mediator.Redis brings robust message streaming and distributed pub/sub to AwadyLab.Mediator powered by Redis Streams and Consumer Groups:

flowchart TD
    Mediator{{"AwadyLab.Mediator"}}
    
    Mediator -->|"Delivery = Queue"| QMode["Queue Mode<br/>(RedisMessageQueue)"]
    Mediator -->|"Delivery = Broker"| BMode["Broker Mode<br/>(RedisBroker)"]
    
    QMode --> SingleStream[("Single Redis Stream<br/>awadylab:mediator:queue")]
    BMode --> TopicStream[("Topic Redis Stream<br/>awadylab:events:{stream}")]
    
    SingleStream --> Pump["Internal Queue Pump<br/>(Consumer Group)"]
    TopicStream --> Group1[("Consumer Group: Billing")]
    TopicStream --> Group2[("Consumer Group: Inventory")]
    
    Group1 --> H1{{"IExternalNotificationHandler&lt;&gt;"}}
    Group2 --> H2{{"IExternalNotificationHandler&lt;&gt;"}}

Why Redis Streams?

Unlike Redis PUBLISH/SUBSCRIBE (which drops messages if a subscriber is offline and does not support acknowledgments), Redis Streams provides:

• Durable At-Least-Once Delivery: Messages are persisted in the stream until explicitly acknowledged (XACK).
• Consumer Groups: Independent microservices or service instances can form consumer groups (XREADGROUP), load-balancing or competing for messages safely.
• Pending Entries & Auto-Claim: If a worker process crashes mid-execution, unacknowledged entries exceeding ClaimIdleTimeout are automatically claimed and retried by surviving workers (XAUTOCLAIM).

Features & Reliability

• Caller-Owned Connection Lifecycle: The multiplexer is provided via ConnectionMultiplexerFactory, seamlessly integrating with existing Redis configurations and never forcibly closing caller connections.
• Automatic Stream Trimming: Streams are capped using approximate trimming (MAXLEN ~ MaxStreamLength), ensuring predictable Redis memory usage with minimal CPU overhead.
• Auto-Claim Crashed Workers: Unacknowledged messages exceeding ClaimIdleTimeout are automatically reclaimed and delivered to active workers, preventing message loss when pods or instances crash.
• Auto-Reconnection: Re-establishes stream subscriptions automatically if Redis disconnects or fails over.
• Dead-Letter Routing: Envelopes that exhaust retries are routed to the configured DeadLetterStream or local IDeadLetterSink.

Installation

dotnet add package AwadyLab.Mediator.Redis

Quickstart

1. Register in DI (Program.cs)

using AwadyLab.Mediator;
using AwadyLab.Mediator.Redis;
using AwadyLab.Mediator.Redis.Options;
using StackExchange.Redis;

var builder = WebApplication.CreateBuilder(args);

// Register existing ConnectionMultiplexer in DI (or supply factory directly)
builder.Services.AddSingleton<IConnectionMultiplexer>(sp =>
    ConnectionMultiplexer.Connect("localhost:6379"));

builder.Services.AddMediator([typeof(Program).Assembly], options =>
{
    options.UseRedis(new RedisOptions
    {
        // Reuses existing IConnectionMultiplexer or creates a new connection
        ConnectionMultiplexerFactory = () => Task.FromResult(
            builder.Services.BuildServiceProvider().GetRequiredService<IConnectionMultiplexer>()),

        // 1. Enable Broker Pub/Sub Mode
        Broker = new RedisBrokerOptions
        {
            StreamPrefix = "myapp:events",
            Prefetch = 15,
            MaxStreamLength = 50_000
        },

        // 2. (Optional) Enable Durable Queue Mode for local queue delivery
        Queue = new RedisQueueOptions
        {
            StreamKey = "myapp:mediator:queue",
            Prefetch = 10
        }
    });
});

var app = builder.Build();

Broker Mode (Cross-Service Pub/Sub)

Broadcasts events across microservices using Redis Streams. Subscribing services bind their consumer groups to the stream.

1. Define Notification & Stream Topology

Implement IExternalNotificationDefinition<T> to declare the target stream:

using AwadyLab.Mediator.Abstraction;
using AwadyLab.Mediator.Abstraction.Messaging;
using AwadyLab.Mediator.Models.Options;
using AwadyLab.Mediator.Redis;

// Domain event contract
public record OrderPlacedNotification(Guid OrderId, decimal Total) : INotification;

// Topology definition (stream mapping)
public sealed class OrderPlacedDefinition : IExternalNotificationDefinition<OrderPlacedNotification>
{
    public void Define(ExternalNotificationOptions options)
    {
        // Resulting Redis stream key will be '{StreamPrefix}:orders.placed'
        options.Redis.Stream = "orders.placed";
        options.Redis.MaxStreamLength = 100_000; // Optional per-stream retention limit
    }
}

2. Implement External Subscriber Handler

Consumers implement IExternalNotificationHandler<T> and specify a static QueueName (which serves as the Consumer Group name):

using AwadyLab.Mediator.Abstraction.Messaging;
using AwadyLab.Mediator.Models.Options;

[ExternalSubscription(Prefetch = 20, MaxRetries = 3)]
public class BillingServiceHandler(IBillingService billing) 
    : IExternalNotificationHandler<OrderPlacedNotification>
{
    // Required: Acts as the Redis Consumer Group on the stream
    public static string QueueName => "billing-service";

    public async Task Handle(OrderPlacedNotification notification, CancellationToken ct)
    {
        await billing.GenerateInvoiceAsync(notification.OrderId, notification.Total, ct);
    }
}

3. Publishing to the Broker

await mediator.Publish(new OrderPlacedNotification(orderId, 250m), options =>
{
    options.Delivery = NotificationDelivery.Broker;

    // Optional: extra headers carried with this stream entry (Redis Streams don't route on headers —
    // this is metadata a consumer can read, not a filter, since every consumer group sees every entry).
    options.Redis.Headers["region"] = "eu";
});

Queue Mode (Durable In-Process Channel)

When RedisOptions.Queue is configured, NotificationDelivery.Queue routes envelopes through a durable Redis Stream instead of an in-memory channel:

await mediator.Publish(new OrderPlacedNotification(orderId, 250m), options =>
{
    options.Delivery = NotificationDelivery.Queue;
});

• Survives service restarts.
• Drained by the mediator's background NotificationQueuePump workers with consumer group coordination.

Options Reference

RedisOptions

RedisBrokerOptions

RedisQueueOptions

License

Package by Mohamed Elawady

Copyright (c) Mohamed Elawady. All rights reserved.

This software is closed source and proprietary. Subject to the terms below,
you are granted a free, worldwide, non-exclusive license to:

  - Use this software, in source or compiled (NuGet package) form, in your
    own applications, including commercial applications, at no charge.

You may NOT:

  - Redistribute, sublicense, sell, or publish this software (or any
    modified version of it) as a standalone product, library, or package,
    including republishing it under a different name on NuGet.org or any
    other package repository.
  - Reverse engineer, decompile, or disassemble the compiled package except
    to the extent applicable law expressly permits this despite this
    limitation.
  - Remove or alter any copyright, trademark, or other proprietary notices.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHOR BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.