Zonit.Messaging.Commands.Abstractions 1.0.2

Zonit.Messaging

A lightweight, high-performance .NET library for building event-driven and CQRS architectures with full AOT/Trimming support.

📦 NuGet Packages

Current Packages (Zonit.Messaging)

Package	Version	Downloads	Description
Zonit.Messaging.Commands			CQRS Commands & Queries
Zonit.Messaging.Commands.Abstractions			Command interfaces
Zonit.Messaging.Events			Pub/Sub Events
Zonit.Messaging.Events.Abstractions			Event interfaces
Zonit.Messaging.Tasks			Background Jobs
Zonit.Messaging.Tasks.Abstractions			Task interfaces

Legacy Packages (deprecated)

Package	Version	Downloads	Status
Zonit.Services.EventMessage			⚠️ Deprecated
Zonit.Services.EventMessage.Abstractions			⚠️ Deprecated

# Install current packages
dotnet add package Zonit.Messaging.Commands
dotnet add package Zonit.Messaging.Events
dotnet add package Zonit.Messaging.Tasks

# Or via NuGet Package Manager
Install-Package Zonit.Messaging.Commands
Install-Package Zonit.Messaging.Events
Install-Package Zonit.Messaging.Tasks

Features

• Commands (CQRS) - Request/Response pattern with strongly-typed handlers
• Events (Pub/Sub) - Publish events to multiple subscribers (fan-out)
• Tasks (Background Jobs) - Queue long-running operations with retry support
• Transaction Support - Group events into transactions to be processed sequentially
• AOT-Safe - Full Native AOT and trimming support via Source Generators
• Concurrent Processing - Control the number of concurrently executed handlers
• Timeout Handling - Configure timeouts for processing

Requirements

• .NET 8, .NET 9 or .NET 10

Quick Start

Recommended: Using Source Generators (AOT-Safe)

Auto-register handlers at compile time for Native AOT support:

using Zonit.Messaging.Commands;
using Zonit.Messaging.Events;
using Zonit.Messaging.Tasks;

services.AddCommandHandlers();
services.AddEventHandlers();
services.AddTaskHandlers();

Alternative: Manual Registration

Manually register providers and handlers:

using Zonit.Messaging.Commands;
using Zonit.Messaging.Events;
using Zonit.Messaging.Tasks;

// Commands (CQRS)
services.AddCommandProvider();
services.AddCommand<CreateUserHandler>();

// Events (Pub/Sub)
services.AddEventProvider();

// Tasks (Background Jobs)
services.AddTaskProvider();

Commands (CQRS)

Request/Response pattern - send a request, get a typed response.

1. Define a Command

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

2. Implement Handler

public class CreateUserHandler : IRequestHandler<CreateUserCommand, Guid>
{
    public async Task<Guid> HandleAsync(CreateUserCommand request, CancellationToken ct = default)
    {
        var userId = Guid.NewGuid();
        // Save to database...
        return userId;
    }
}

3. Send Command

var commandProvider = serviceProvider.GetRequiredService<ICommandProvider>();
var userId = await commandProvider.SendAsync(new CreateUserCommand("John", "john@example.com"));

Events (Pub/Sub)

Publish events to multiple subscribers asynchronously.

1. Subscribe to Events

var eventManager = serviceProvider.GetRequiredService<IEventManager>();

eventManager.Subscribe<UserCreatedEvent>(async payload =>
{
    Console.WriteLine($"User created: {payload.Data.Name}");
});

2. Publish Events

var eventProvider = serviceProvider.GetRequiredService<IEventProvider>();
eventProvider.Publish(new UserCreatedEvent { Name = "John" });

3. Using Transactions

Group events to be processed sequentially:

using (var transaction = eventProvider.CreateTransaction())
{
    eventProvider.Publish(new Event1());
    eventProvider.Publish(new Event2());
    // Events are queued until the transaction is completed
}
// Events are processed after the transaction is disposed

4. Awaiting Transaction Completion

Wait for all events in a transaction to be processed:

using (var transaction = eventProvider.CreateTransaction())
{
    eventProvider.Publish(new OrderCreatedEvent());
    eventProvider.Publish(new InventoryUpdatedEvent());
    
    // Wait for all events to be processed before continuing
    await transaction.WaitForCompletionAsync();
}

Tasks (Background Jobs)

Long-running operations with retry support and real-time progress tracking.

1. Simple Task Handler

var taskManager = serviceProvider.GetRequiredService<ITaskManager>();

taskManager.Subscribe<SendEmailTask>(async payload =>
{
    await SendEmailAsync(payload.Data.To, payload.Data.Subject);
}, new TaskSubscriptionOptions
{
    WorkerCount = 5,
    MaxRetries = 3,
    RetryDelay = TimeSpan.FromSeconds(10)
});

2. Task Handler with Progress Tracking

Create handlers with automatic progress reporting based on estimated step durations:

public class ImportDataHandler : TaskHandler<ImportDataTask>
{
    public override int WorkerCount => 2;
    public override TimeSpan Timeout => TimeSpan.FromMinutes(10);
    
    // Optional: Display title and description in UI
    public override string? Title => "Import Data";
    public override string? Description => "Importing data from external source";

    // Define steps with estimated durations for smooth progress calculation
    public override TaskProgressStep[]? ProgressSteps =>
    [
        new(TimeSpan.FromSeconds(5), "Connecting to source..."),
        new(TimeSpan.FromSeconds(10), "Downloading data..."),
        new(TimeSpan.FromSeconds(15), "Processing records..."),
        new(TimeSpan.FromSeconds(5), "Saving to database...")
    ];

    protected override async Task HandleAsync(
        ImportDataTask data,
        ITaskProgressContext progress,
        CancellationToken cancellationToken)
    {
        // Step 1: Connect (0% -> 14%)
        await progress.NextAsync();
        await ConnectAsync(cancellationToken);

        // Step 2: Download (14% -> 43%)
        await progress.NextAsync();
        await DownloadAsync(data.Url, cancellationToken);

        // Step 3: Process (43% -> 86%)
        await progress.NextAsync();
        for (int i = 0; i < data.RecordCount; i++)
        {
            await ProcessRecordAsync(i, cancellationToken);
            // Update message without changing step
            await progress.SetMessageAsync($"Processing {i + 1}/{data.RecordCount}...");
        }

        // Step 4: Save (86% -> 100%)
        await progress.NextAsync();
        await SaveAsync(cancellationToken);
    }
}

Register the handler:

var handler = new ImportDataHandler();
taskManager.Subscribe<ImportDataTask>(
    async payload => await ((ITaskHandler<ImportDataTask>)handler).HandleAsync(payload),
    new TaskSubscriptionOptions
    {
        WorkerCount = handler.WorkerCount,
        Timeout = handler.Timeout,
        ProgressSteps = handler.ProgressSteps
    });

3. Publish Tasks

var taskProvider = serviceProvider.GetRequiredService<ITaskProvider>();

// Simple publish
taskProvider.Publish(new SendEmailTask { To = "user@example.com", Subject = "Welcome!" });

// Publish with ExtensionId for filtering
var organizationId = Guid.NewGuid();
taskProvider.Publish(new ImportDataTask("data.csv", 1000), organizationId);

4. Monitoring Task Progress

Subscribe to real-time progress updates:

var taskManager = serviceProvider.GetRequiredService<ITaskManager>();

// Monitor all tasks
taskManager.OnChange(state =>
{
    Console.WriteLine($"Task {state.TaskType}: {state.Progress}% - {state.Message}");
    Console.WriteLine($"Duration: {state.Duration}");
});

// Monitor specific task type with typed data access
taskManager.OnChange<ImportDataTask>(state =>
{
    Console.WriteLine($"Import from {state.Data.Source}: {state.Progress}%");
    Console.WriteLine($"Step {state.CurrentStep}/{state.TotalSteps}: {state.Message}");
    Console.WriteLine($"Running for: {state.Duration?.TotalSeconds:F1}s");
});

// Monitor tasks for specific ExtensionId (e.g., organization)
taskManager.OnChange(organizationId, state =>
{
    // Only tasks published with this ExtensionId
    UpdateProgressBar(state.Progress ?? 0);
});

// Monitor specific type for specific ExtensionId
taskManager.OnChange<ImportDataTask>(organizationId, state =>
{
    // Typed access + filtered by ExtensionId
    UpdateUI(state.Data, state.Progress);
});

5. TaskState Properties

6. Viewing Active Tasks

// Get all active tasks
var activeTasks = taskManager.GetActiveTasks();

// Get active tasks for specific ExtensionId
var orgTasks = taskManager.GetActiveTasks(organizationId);

foreach (var task in activeTasks)
{
    Console.WriteLine($"{task.TaskType}: {task.Status} ({task.Progress}%) - {task.Duration}");
}

// Get specific task state
var state = taskManager.GetTaskState(taskId);

7. Task Lifecycle

Tasks go through various states during their lifecycle:

• Pending - Task is queued but not yet started
• Processing - Task is currently being processed
• Completed - Task finished successfully
• Failed - Task failed during processing
• Cancelled - Task was cancelled before completion

8. Progress Tracking Features

• Time-based smooth progress: Progress is automatically calculated based on estimated step durations
• Automatic updates: Progress updates are sent automatically (max 100 times, when % changes)
• Step tracking: Know which step is currently executing and how many remain
• Duration tracking: Real-time duration available via Duration property
• Typed access: Use OnChange<T> to get typed access to task data
• Efficient filtering: Filter by ExtensionId at the system level for better performance

Migration from Legacy API

If upgrading from Zonit.Services.EventMessage:

Legacy code continues to work but shows deprecation warnings.

Example Use Cases

• Decoupling logic between independent application components
• Implementing event-driven workflows
• Handling system notifications and real-time updates
• Creating robust background job queues
• CQRS architecture implementation

Contributing & Support

Found a bug or have a feature request? Open an issue on GitHub!

License

MIT