Conveyo.Storage.Postgres 0.1.0-preview.2

Conveyo

Conveyo helps you build event-driven .NET applications without tying your application code to a broker client. Define messages, write consumers, and use one bus abstraction to send commands, publish events, and pass large payloads by reference.

The bits you interact with is minimalistic by design:

• Send<T> routes a command to one queue.
• Publish<T> emits an event to zero, one, or many queues.
• IConsumer<T> handles messages inside normal .NET dependency-injection scopes.

For transport and blob storage there are additional packages:

Quick Start

Install the core package and one transport:

dotnet add package Conveyo
dotnet add package Conveyo.RabbitMQ

Define messages as plain reference types:

public sealed record SubmitWeatherObservationCommand
{
    public Guid ObservationId { get; init; } = Guid.NewGuid();
    public required Guid StationId { get; init; }
    public required string Location { get; init; }
    public required int HumidityPercent { get; init; }
    public required float WindSpeedMs { get; init; }
    public required double PressureHpa { get; init; }
    public required bool IsPrecipitating { get; init; }
    public DateTime ObservedAt { get; init; } = DateTime.UtcNow;
}

public sealed record WeatherObservationRecordedEvent
{
    public required Guid ObservationId { get; init; }
    public required Guid StationId { get; init; }
    public required string Location { get; init; }
    public required double FeelsLikeC { get; init; }
    public required string Summary { get; init; }
    public DateTime RecordedAt { get; init; } = DateTime.UtcNow;
}

Register the bus in your host:

services.AddConveyo(bus =>
{
    bus.Map<SubmitWeatherObservationCommand>("weather:SubmitWeatherObservationCommand.v1");
    bus.Map<WeatherObservationRecordedEvent>("weather:WeatherObservationRecordedEvent.v1");

    bus.MapEndpointConvention<SubmitWeatherObservationCommand>(new Uri("queue:weather-stations"));
    bus.AddConsumer<SubmitWeatherObservationConsumer>();

    bus.UsingRabbitMq((ctx, rabbit) =>
    {
        rabbit.Host("localhost", "/", host =>
        {
            host.Username("guest");
            host.Password("guest");
        });

        rabbit.ReceiveEndpoint("weather-stations", endpoint =>
        {
            endpoint.ConfigureConsumer<SubmitWeatherObservationConsumer>(ctx);
        });
    });
});

Handle messages with IConsumer<T>:

public sealed class SubmitWeatherObservationConsumer(IWeatherService weather)
    : IConsumer<SubmitWeatherObservationCommand>
{
    public async Task Consume(ConsumeContext<SubmitWeatherObservationCommand> context)
    {
        var observation = await weather.RecordAsync(
            context.Message,
            context.CancellationToken);

        await context.Publish(new WeatherObservationRecordedEvent
        {
            ObservationId = observation.ObservationId,
            StationId = observation.StationId,
            Location = observation.Location,
            FeelsLikeC = observation.FeelsLikeC,
            Summary = observation.Summary
        }, context.CancellationToken);
    }
}

Send or publish from application services by injecting IBus:

public sealed class WeatherStationClient(IBus bus)
{
    public Task SubmitObservationAsync(
        Guid stationId,
        string location,
        CancellationToken cancellationToken)
    {
        return bus.Send(new SubmitWeatherObservationCommand
        {
            StationId = stationId,
            Location = location,
            HumidityPercent = 91,
            WindSpeedMs = 8.4f,
            PressureHpa = 1007.2,
            IsPrecipitating = true
        }, cancellationToken);
    }
}

Documentation

• Getting started: registration, consumers, command routing, and events.
• RabbitMQ transport: host options, TLS, topology, retries, and failure queues.
• MessageData: storing large payloads outside the envelope.
• Wire contract: JSON envelope and RabbitMQ wire shape for non-.NET clients.
• MessageData URI schemes: data: and pgbin:// resolver contracts.

The sample producer and consumer in examples/ show RabbitMQ with optional Postgres MessageData storage.