FlowCore 2.2.3

📦 FlowCore

FlowCore is a .NET 8+ framework for CQRS, Event-Driven and Microservices architectures. It provides an extensible Mediator with Pipeline Behaviors, multi-provider EventBus (RabbitMQ, Kafka, InMemory), Outbox, Inbox, Saga, Scheduling, Retry, DLQ, OpenTelemetry, Execution Scope, Handler Discovery (hybrid: Source Generator + reflection fallback), Module Manifest, Health Checks, Metrics Context, Resilience Policies, Hosted Workers, Plugin Model and Testing Infrastructure.

🎯 Features

CQRS + Pipeline

• Commands, Queries and Events
• Pipeline Behaviors: Logging, Validation (FluentValidation), Caching, Transactions (EF Core), Event Dispatcher
• Hybrid Handler Discovery: GeneratedHandlerRegistry (compile-time) with reflection fallback
• Optional Source Generator to eliminate Runtime Reflection

EventBus

• IEventBus — single abstraction for event publishing
• Providers: InMemory (default), RabbitMQ, Kafka
• DiagnosticsEventBus — decorator with tracing, metrics and IDiagnosticsContext
• Providers as IMessageProvider with Start/Stop lifecycle managed by BootstrapCoordinator

Module Manifest

• IModuleManifest — official identity of each module (Name, Version, Capabilities, Dependencies)
• IModuleRegistry — central catalog of all loaded modules
• PluginModule — base class for third-party plugins
• Automatic version compatibility validation on Bootstrap

Hosting & Application Lifecycle

• IBootstrapCoordinator — ordered startup and shutdown (Core → Providers → Workers)
• BootstrapHostedService — integration with .NET Generic Host
• Reverse shutdown order: Workers → Providers → Core
• Startup failures abort initialization; shutdown failures never block

Hosted Workers

• IHostedWorker — unified interface for continuous processing workers
• IHostedWorkerManager — manages all workers lifecycle
• Each work unit creates a new ExecutionScope (mandatory isolation)
• Supports: RabbitMQ Consumer, Kafka Consumer, Outbox, Inbox, Scheduler, Dead Letter

Health Checks

• IHealthCheck — component health verification interface
• HealthCheckResult with Status (Healthy, Degraded, Unhealthy), Duration, Metadata
• IHealthCheckRegistry — centralized check registration
• ASP.NET Core Health Checks integration via AddHealthCheck<T>()
• Each module registers only its own checks

Metrics Context

• IMetricsContext — per-ExecutionScope metrics collection
• MetricEntry with Name, Type (Counter, Gauge, Histogram, Timer), Value, Tags
• Pipeline, EventBus, Providers, Retry and Scheduler can record metrics
• Isolated context per execution — never shared between executions

Resilience

• IResiliencePolicy — unified resilience policy abstraction
• Policies: Timeout, Circuit Breaker, Bulkhead, Fallback, Rate Limiter
• PolicyComposer — chained composition of multiple policies
• Integration with Pipeline, EventBus and Providers
• Existing IRetryPolicy + ImmediateRetryPolicy

Messaging Providers

• RabbitMQ — AddRabbitMQ() with publish/consumer worker, auto-reconnect
• Kafka — AddKafka() with publish/consumer groups, managed commit
• Registered via IProviderRegistry + lifecycle managed by Bootstrap

Transactional Patterns

• Outbox — reliable publishing with IOutboxStore + OutboxWorker
• Inbox — idempotent processing with IInboxStore (deduplication by MessageId)
• Saga Orchestration — SagaCoordinator with steps, reverse-order compensation
• Scheduled Messages — absolute/relative scheduling with IMessageScheduler

Observability

• IActivityFactory + IMetricRecorder (no-op by default, zero overhead)
• IDiagnosticsContext with DiagnosticEntry centralized in ExecutionScope
• Distributed tracing with CorrelationId propagation

Execution Scope

• IExecutionScope — shared context per execution (CorrelationId, Items, Diagnostics, Metrics)
• AsyncLocal thread-safe, available without DI in Pipeline, Behaviors, Handlers and Providers

Plugin Model

• PluginModule — base class for plugins extending FlowCore
• Mandatory ModuleManifest with MinimumFlowCoreVersion for validation
• Plugins can register: Providers, Workers, Behaviors, Health Checks, Metrics
• Bootstrap treats plugins and official modules identically

AOT Compatibility

• Preferred path via Source Generators (zero runtime reflection)
• Reflection fallback annotated with [RequiresDynamicCode] and [RequiresUnreferencedCode]
• FlowMediator, HandlerDiscovery and DispatcherCache prioritize generated code
• Ready for Native AOT and Linker Trimming

Testing Infrastructure

• FlowCore.Testing — NuGet package with FakeEventBus, FakeClock, IFlowCoreTestBuilder
• Test environment setup without external infrastructure
• Isolation via ExecutionScope identical to runtime

📥 Installation

Core

dotnet add package FlowCore --version 2.2.3

Providers

dotnet add package FlowCore.RabbitMQ --version 2.2.3
dotnet add package FlowCore.Kafka --version 2.2.3

Testing

dotnet add package FlowCore.Testing --version 2.2.3

⚙️ Configuration

Basic

builder.Services.AddFlowCore();

With RabbitMQ

builder.Services
    .AddFlowCore()
    .AddRabbitMQ(options =>
    {
        options.Host = "localhost";
        options.Username = "guest";
        options.Password = "guest";
    });

With Kafka

builder.Services
    .AddFlowCore()
    .AddKafka(options =>
    {
        options.BootstrapServers = "localhost:9092";
        options.ConsumerGroup = "my-service";
    });

Optional modules

builder.Services
    .AddFlowCore()
    .AddFlowCoreTransactions()      // EF Core transaction scope
    .AddFlowCoreOutbox()             // Outbox Worker
    .AddFlowCoreDiagnostics()        // System.Diagnostics Activity + Metrics
    .AddFlowCoreSagaListener()       // Saga event listener
    .AddFlowCoreScheduler();         // Scheduled Messages Worker

Custom module with Manifest

public class MyModule : IFlowCoreModule
{
    public IModuleManifest Manifest { get; }
        = new ModuleManifest("MyModule", new Version(1, 0, 0),
            ["CustomProvider", "HealthCheck"]);

    public void Configure(IFlowCoreBuilder builder)
    {
        builder.AddHealthCheck<MyHealthCheck>();
        builder.AddHostedWorker<MyWorker>();
    }
}

builder.Services.AddFlowCore().AddModule<MyModule>();

Custom Health Check

public class MyHealthCheck : IHealthCheck
{
    public async ValueTask<HealthCheckResult> CheckAsync(CancellationToken ct)
    {
        return HealthCheckResult.Healthy("my-component", "All ok");
    }
}

Resilience policy

var pipeline = new PolicyComposer(
    new CircuitBreakerPolicy(failureThreshold: 3),
    new TimeoutPolicy(TimeSpan.FromSeconds(5)));

💡 Usage Examples

Commands and Queries

public record CreateUserCommand(string Name, string Email) : ICommand<Guid>;

public class CreateUserHandler : ICommandHandler<CreateUserCommand, Guid>
{
    public async Task<Guid> HandleAsync(CreateUserCommand command, CancellationToken ct)
    {
        var user = new User { Id = Guid.NewGuid(), Name = command.Name, Email = command.Email };
        return user.Id;
    }
}

var userId = await _mediator.SendAsync(new CreateUserCommand("John", "john@email.com"));

Events

public record UserCreatedEvent(Guid UserId) : IEvent;

public class UserCreatedHandler : IEventHandler<UserCreatedEvent>
{
    public Task HandleAsync(UserCreatedEvent @event, CancellationToken ct)
    {
        Console.WriteLine($"User created: {@event.UserId}");
        return Task.CompletedTask;
    }
}

await _eventBus.PublishAsync(new UserCreatedEvent(userId));

Scheduled Messages

await _scheduler.ScheduleAfterAsync(
    new OrderExpiredEvent(orderId),
    TimeSpan.FromHours(2));

Saga

public class OrderSaga : Saga
{
    public override Task DefineStepsAsync()
    {
        AddStep<OrderPlacedEvent>("ReserveInventory", async (evt, ct) =>
        {
            // execute
        }, compensate: async (evt, ct) =>
        {
            // compensate if later step fails
        });

        AddStep<PaymentProcessedEvent>("ProcessPayment", async (evt, ct) =>
        {
            // execute
        });

        return Task.CompletedTask;
    }
}

builder.Services.AddSaga<OrderSaga>();
builder.Services.AddFlowCoreSagaListener();

✅ Tests

131 unit tests covering Mediator, Behaviors, DI, EventBus, Serialization, Retry, DLQ, Outbox, Inbox, Tracing, Saga, Scheduling, DispatcherCache, Pipeline Integration, and Hosting.

dotnet test

For testing applications that use FlowCore, use the FlowCore.Testing package:

var services = new ServiceCollection();
var builder = services.CreateTestBuilder();
var provider = builder.Build();
var fakeBus = provider.GetFakeEventBus();

// Execute scenario...
Assert.Single(fakeBus.Published);

📄 License

MIT License