NetEvolve.Pulse 0.17.3

NetEvolve.Pulse

NetEvolve.Pulse is a high-performance CQRS mediator for ASP.NET Core that wires commands, queries, and events through a scoped, interceptor-enabled pipeline.

Features

• Typed CQRS mediator with single-handler enforcement for commands and queries plus fan-out events
• Minimal DI integration via services.AddPulse(...) with scoped lifetimes for handlers and interceptors
• Configurable interceptor pipeline (logging, metrics, tracing, validation) via IMediatorConfigurator
• Distributed query caching — register ICacheableQuery<TResponse> per query and enable transparent IDistributedCache caching with AddQueryCaching()
• Outbox pattern with background processor for reliable event delivery via AddOutbox()
• Parallel event dispatch for efficient domain event broadcasting
• TimeProvider-aware for deterministic testing and scheduling scenarios
• OpenTelemetry-friendly metrics and tracing through AddActivityAndMetrics()

Installation

NuGet Package Manager

Install-Package NetEvolve.Pulse

.NET CLI

dotnet add package NetEvolve.Pulse

PackageReference

<PackageReference Include="NetEvolve.Pulse" Version="x.x.x" />

Quick Start

using Microsoft.Extensions.DependencyInjection;
using NetEvolve.Pulse;
using NetEvolve.Pulse.Extensibility;

var services = new ServiceCollection();

// Register Pulse and handlers
services.AddPulse();
services.AddScoped<ICommandHandler<CreateOrderCommand, OrderCreated>, CreateOrderHandler>();

using var provider = services.BuildServiceProvider();
var mediator = provider.GetRequiredService<IMediator>();

var result = await mediator.SendAsync<CreateOrderCommand, OrderCreated>(
    new CreateOrderCommand("SKU-123"));

Console.WriteLine($"Created order {result.OrderId}");

public record CreateOrderCommand(string Sku) : ICommand<OrderCreated>;
public record OrderCreated(Guid OrderId);

public sealed class CreateOrderHandler
    : ICommandHandler<CreateOrderCommand, OrderCreated>
{
    public Task<OrderCreated> HandleAsync(
        CreateOrderCommand command,
        CancellationToken cancellationToken) =>
        Task.FromResult(new OrderCreated(Guid.NewGuid()));
}

Usage

Basic Example

services.AddPulse();
services.AddScoped<IQueryHandler<GetOrderQuery, Order>, GetOrderHandler>();
services.AddScoped<IEventHandler<OrderCreatedEvent>, OrderCreatedHandler>();

var order = await mediator.QueryAsync<GetOrderQuery, Order>(new GetOrderQuery(orderId));
await mediator.PublishAsync(new OrderCreatedEvent(order.Id));

public record GetOrderQuery(Guid Id) : IQuery<Order>;
public record Order(Guid Id, string Sku);
public record OrderCreatedEvent(Guid Id) : IEvent;

public sealed class GetOrderHandler : IQueryHandler<GetOrderQuery, Order>
{
    public Task<Order> HandleAsync(GetOrderQuery query, CancellationToken cancellationToken) =>
        Task.FromResult(new Order(query.Id, "SKU-123"));
}

public sealed class OrderCreatedHandler : IEventHandler<OrderCreatedEvent>
{
    public Task HandleAsync(OrderCreatedEvent @event, CancellationToken cancellationToken)
    {
        // React to the event (logging, projections, etc.)
        return Task.CompletedTask;
    }
}

Advanced Example

// Enable tracing and metrics and add custom interceptors
services.AddPulse(config =>
{
    config.AddActivityAndMetrics();
});

services.AddScoped<ICommandHandler<ShipOrderCommand, Void>, ShipOrderHandler>();

public record ShipOrderCommand(Guid Id) : ICommand;

public sealed class ShipOrderHandler : ICommandHandler<ShipOrderCommand, Void>
{
    public Task<Void> HandleAsync(ShipOrderCommand command, CancellationToken cancellationToken)
    {
        // Shipping workflow here
        return Task.FromResult(Void.Completed);
    }
}

Configuration

// Configure Pulse during startup
services.AddPulse(config =>
{
    // Built-in observability
    config.AddActivityAndMetrics();

    // Add your own configurator extensions for validation, caching, etc.
    // config.AddCustomValidation();
});

Distributed Query Caching

Enable transparent IDistributedCache caching for queries. Any query that implements ICacheableQuery<TResponse> (from NetEvolve.Pulse.Extensibility) is served from the cache on subsequent invocations; all other queries pass through unchanged.

// 1. Register an IDistributedCache implementation
services.AddDistributedMemoryCache(); // or Redis, SQL Server, etc.

// 2. Enable the caching interceptor (with optional options)
services.AddPulse(config => config.AddQueryCaching(options =>
{
    // Choose between absolute (default) and sliding expiration
    options.ExpirationMode = CacheExpirationMode.Sliding;

    // Supply custom JsonSerializerOptions for cache serialization
    options.JsonSerializerOptions = new JsonSerializerOptions
    {
        PropertyNamingPolicy = JsonNamingPolicy.CamelCase,
    };
}));

// 3. Implement ICacheableQuery<TResponse> on the queries you want cached
public record GetProductQuery(Guid Id) : ICacheableQuery<ProductDto>
{
    public string? CorrelationId { get; set; }

    // Unique per query result — include all discriminating parameters
    public string CacheKey => $"product:{Id}";

    // null = no explicit expiry (relies on cache defaults); or provide a TimeSpan
    public TimeSpan? Expiry => TimeSpan.FromMinutes(5);
}

Behavior summary:

Outbox Pattern Configuration

The outbox pattern ensures reliable event delivery by persisting events before dispatching:

services.AddPulse(config => config
    .AddOutbox(
        options => options.Schema = "pulse",
        processorOptions =>
        {
            processorOptions.BatchSize = 100;              // Messages per batch (default: 100)
            processorOptions.PollingInterval = TimeSpan.FromSeconds(5);  // Poll delay (default: 5s)
            processorOptions.MaxRetryCount = 3;            // Max retries before dead letter (default: 3)
            processorOptions.ProcessingTimeout = TimeSpan.FromSeconds(30); // Per-message timeout (default: 30s)
            processorOptions.EnableBatchSending = false;   // Use batch transport (default: false)
        })
    // Choose a persistence provider:
    // .AddEntityFrameworkOutbox<MyDbContext>()
    // .AddSqlServerOutbox(connectionString)
);

See NetEvolve.Pulse.EntityFramework or NetEvolve.Pulse.SqlServer for persistence provider setup.

Requirements

• .NET 8.0, .NET 9.0, or .NET 10.0
• ASP.NET Core environment with Microsoft.Extensions.DependencyInjection
• OpenTelemetry packages when using AddActivityAndMetrics()

Related Packages

• NetEvolve.Pulse.Dapr - Dapr pub/sub integration for event dispatch
• NetEvolve.Pulse.Extensibility - Core contracts and abstractions used by the mediator
• NetEvolve.Pulse.EntityFramework - Entity Framework Core persistence for the outbox pattern
• NetEvolve.Pulse.SqlServer - SQL Server ADO.NET persistence for the outbox pattern
• NetEvolve.Pulse.Polly - Polly v8 resilience policies integration

Documentation

For complete documentation, please visit the official documentation.

Contributing

Contributions are welcome! Please read the Contributing Guidelines before submitting a pull request.

Support

• Issues: Report bugs or request features on GitHub Issues
• Documentation: Read the full documentation at https://github.com/dailydevops/pulse

License

This project is licensed under the MIT License - see the LICENSE file for details.