eQuantic.Core.CQS 2.4.0

eQuantic.Core.CQS

A modern, high-performance CQS/CQRS framework for .NET with pipeline behaviors, notifications, streaming, sagas, and distributed system support. MIT-licensed alternative to MediatR.

Features

• ✅ Commands & Queries - Clear separation of read and write operations
• ✅ Pipeline Behaviors - Cross-cutting concerns (logging, validation, etc.)
• ✅ Notifications - Publish-subscribe pattern with multiple handlers
• ✅ Streaming - IAsyncEnumerable support for large datasets
• ✅ Sagas - Multi-step transactions with compensation
• ✅ Outbox Pattern - Reliable message publishing
• ✅ Job Scheduling - Deferred command execution
• ✅ Paged Queries - Built-in pagination support
• ✅ Modern .NET - Targets .NET 6, 8, and 9

Packages

Package	Description	NuGet
eQuantic.Core.CQS	Core framework
eQuantic.Core.CQS.Abstractions	Interfaces and contracts
eQuantic.Core.CQS.Generators	Source generators

Persistence Providers

Cloud Messaging

Telemetry Providers

Resilience Providers

Installation

# Core package
dotnet add package eQuantic.Core.CQS

# Optional: Provider packages
dotnet add package eQuantic.Core.CQS.Redis
dotnet add package eQuantic.Core.CQS.Azure

Quick Start

1. Register Services

// Basic setup
services.AddCQS(options => options
    .FromAssemblyContaining<Program>());

// With providers
services.AddCQS(options => options
    .FromAssemblyContaining<Program>()
    .UsePreProcessor = true;
    .UseRedis(redis => redis.ConnectionString = "localhost:6379")
    .UseAzureServiceBus(sb => {
        sb.ConnectionString = "Endpoint=sb://...";
        sb.QueueOrTopicName = "outbox";
    }));

2. Define a Query

public record GetUserByIdQuery(Guid Id) : IQuery<UserDto>;

3. Create a Handler

public class GetUserByIdHandler : IQueryHandler<GetUserByIdQuery, UserDto>
{
    public async Task<UserDto> Execute(GetUserByIdQuery query, CancellationToken ct)
    {
        return new UserDto(query.Id, "John Doe");
    }
}

4. Execute via Mediator

public class UsersController : ControllerBase
{
    private readonly IMediator _mediator;

    [HttpGet("{id}")]
    public async Task<UserDto> Get(Guid id)
        => await _mediator.ExecuteAsync(new GetUserByIdQuery(id));
}

Advanced Features

Commands

// Command without result
public record DeleteUserCommand(Guid Id) : ICommand;

// Command with result
public record CreateUserCommand(string Name) : ICommand<Guid>;

Notifications (Pub/Sub)

public record UserCreatedNotification(Guid UserId) : INotification;

public class SendWelcomeEmailHandler : INotificationHandler<UserCreatedNotification>
{
    public async Task Handle(UserCreatedNotification notification, CancellationToken ct)
    {
        // Send email
    }
}

// Publish to all handlers
await _notificationPublisher.Publish(new UserCreatedNotification(userId));

Streaming

public record GetAllUsersStreamQuery : IStreamQuery<UserDto>;

await foreach (var user in _mediator.ExecuteStreamAsync(new GetAllUsersStreamQuery()))
{
    Console.WriteLine(user.Name);
}

Sagas (Multi-step Transactions)

public class OrderSaga : Saga<OrderSagaData>
{
    protected override void ConfigureSteps()
    {
        Step("ProcessPayment",
            execute: async (data, ct) => { /* charge */ },
            compensate: async (data, ct) => { /* refund */ });

        Step("ReserveInventory",
            execute: async (data, ct) => { /* reserve */ },
            compensate: async (data, ct) => { /* release */ });

        Step("Ship",
            execute: async (data, ct) => { /* ship */ },
            compensate: async (data, ct) => { /* cancel shipment */ });
    }
}

// Execute - automatic compensation on failure
var result = await saga.Execute(new OrderSagaData { OrderId = orderId });
if (!result.IsSuccess)
{
    Console.WriteLine($"Saga failed: {result.Error?.Message}");
}

Azure Service Bus Integration

services.AddCQSAzureServiceBus(options =>
{
    options.ConnectionString = "Endpoint=sb://...";
    options.QueueOrTopicName = "outbox-messages";
    options.UseTopic = false;
});

// Publish outbox messages
await _outboxPublisher.PublishAsync(message);
await _outboxPublisher.PublishBatchAsync(messages);

AWS SQS Integration

services.AddCQSAwsSqs(options =>
{
    options.QueueUrl = "https://sqs.us-east-1.amazonaws.com/123456789/my-queue";
    options.Region = "us-east-1";
});

Pipeline Behaviors

public class LoggingBehavior<TRequest, TResponse> : IPipelineBehavior<TRequest, TResponse>
{
    public async Task<TResponse> Execute(TRequest request, CancellationToken ct, HandlerDelegate<TResponse> next)
    {
        _logger.LogInformation("Handling {Request}", typeof(TRequest).Name);
        var response = await next();
        _logger.LogInformation("Handled {Request}", typeof(TRequest).Name);
        return response;
    }
}

Comparison with MediatR

License

MIT License - See LICENSE for details.