Catga.Transport.Redis 0.1.1

<div align="center">

<img src="docs/web/favicon.svg" width="120" height="120" alt="Catga Logo"/>

Catga

High-Performance .NET CQRS/Event Sourcing Framework

👁 .NET 9
👁 Native AOT

Zero Reflection · Source Generated · Native AOT · Distributed Ready

Quick Start · Performance · Examples · Documentation

</div>

⚡ Performance

BenchmarkDotNet on AMD Ryzen 7 5800H, .NET 9.0.8

✨ Features

• Zero Reflection - Source Generator, compile-time handler discovery
• Native AOT - Full support, trimming safe
• Distributed - Redis Streams, NATS JetStream
• Event Sourcing - Event Store, Snapshots, Projections, Time Travel
• Flow DSL - Distributed workflows, Sagas, ForEach parallel processing
• Reliability - Outbox/Inbox, Idempotency, Dead Letter Queue
• Observability - OpenTelemetry tracing, Metrics

🚀 Quick Start

dotnet add package Catga
dotnet add package Catga.Transport.InMemory
dotnet add package Catga.Persistence.InMemory
dotnet add package Catga.Serialization.MemoryPack

// Define command
[MemoryPackable]
public partial record CreateOrder(string ProductId, int Quantity) : IRequest<Order>;

// Define handler
public class CreateOrderHandler : IRequestHandler<CreateOrder, Order>
{
 public ValueTask<CatgaResult<Order>> HandleAsync(CreateOrder cmd, CancellationToken ct = default)
 => new(CatgaResult<Order>.Success(new Order(cmd.ProductId, cmd.Quantity)));
}

// Configure
builder.Services.AddCatga().UseMemoryPack();
builder.Services.AddInMemoryTransport();
builder.Services.AddInMemoryPersistence();

// Use
var result = await mediator.SendAsync<CreateOrder, Order>(new("PROD-001", 5));

📦 Packages

🛒 OrderSystem Example

A complete e-commerce system demonstrating best practices. Focus on your business logic, not framework boilerplate.

examples/OrderSystem/
├── Commands/ # Command definitions
├── Queries/ # Query definitions
├── Events/ # Event definitions
├── Handlers/ # Business logic
├── Flows/ # Distributed workflows
├── Models/ # Domain models
└── Program.cs # Minimal setup

Run

cd examples/OrderSystem
dotnet run

# Run tests
.\test.ps1

Key Patterns

1. Commands & Queries - Clean separation of write/read operations

// Command - changes state
public record CreateOrder(string CustomerId, List<OrderItem> Items) : IRequest<Order>;

// Query - reads state
public record GetOrder(string OrderId) : IRequest<Order>;

2. Event Sourcing - Full audit trail

public record OrderCreated(string OrderId, string CustomerId) : IEvent;
public record OrderShipped(string OrderId, string TrackingNumber) : IEvent;

3. Flow DSL - Distributed workflows

public class OrderFlow : FlowConfig<OrderState>
{
 protected override void Configure(IFlowBuilder<OrderState> flow)
 {
 flow.Send(s => new ReserveInventory(s.Items))
 .IfFail(s => new ReleaseInventory(s.ReservationId));
 
 flow.Send(s => new ProcessPayment(s.OrderId, s.Total))
 .IfFail(s => new RefundPayment(s.PaymentId));
 
 flow.Publish(s => new OrderCompleted(s.OrderId));
 }
}

🗄️ Event Sourcing

// Append events
await eventStore.AppendAsync("Order-123", new[] { orderCreated, itemAdded });

// Read stream
var stream = await eventStore.ReadAsync("Order-123");

// Snapshots
await snapshotStore.SaveAsync("Order-123", aggregate, version);

// Time Travel
var stateAtV5 = await timeTravelService.GetStateAtVersionAsync("order-1", 5);

🔄 Flow DSL

public class ProcessOrderFlow : FlowConfig<OrderState>
{
 protected override void Configure(IFlowBuilder<OrderState> flow)
 {
 // Sequential steps with compensation
 flow.Send(s => new ReserveInventory(s.OrderId))
 .Into(s => s.ReservationId)
 .IfFail(s => new ReleaseInventory(s.ReservationId));

 // Parallel processing
 flow.ForEach<OrderItem>(s => s.Items)
 .Configure((item, f) => f.Send(s => new ProcessItem(item.Id)))
 .WithParallelism(4)
 .ContinueOnFailure()
 .EndForEach();

 // Conditional logic
 flow.If(s => s.AllItemsProcessed)
 .Send(s => new CompleteOrder(s.OrderId))
 .EndIf();
 }
}

🔧 Configuration

// OpenTelemetry
builder.Services.AddOpenTelemetry()
 .WithTracing(t => t.AddSource(CatgaOpenTelemetryExtensions.ActivitySourceName))
 .WithMetrics(m => m.AddMeter(CatgaOpenTelemetryExtensions.MeterName));

// Resilience
builder.Services.AddCatga()
 .UseResilience(o => o.TransportRetryCount = 3);

// Reliability
builder.Services.AddCatga()
 .UseInbox()
 .UseOutbox();

📚 Documentation

• API Reference

📄 License