Tyto.Transports.RabbitMq 0.0.1-alpha.106

Tyto.Transports.RabbitMq

RabbitMQ transport for Tyto. Publishes and consumes over AMQP with a pooled channel model, automatic topology declaration, retry-then-dead-letter, and OpenTelemetry.

Install

dotnet add package Tyto.Transports.RabbitMq

Usage

builder.AddTyto(tyto => {
    tyto.MessageDefinitions(d => d.Add<PlaceOrder>("place-order", 1));

    tyto.Transports(t => t.AddRabbitMq("rabbitmq", cfg => {
        cfg.Connection("amqp://guest:guest@localhost:5672/");

        cfg.Topology(topo => {
            topo.Exchange("orders-exchange", "fanout");
            topo.Queue("orders");
            topo.Bind("orders", "orders-exchange");
        });

        cfg.Consumers(c => {
            c.DefaultPrefetchCount = 100;
            c.DefaultConcurrencyLimit = 2;   // consumers per queue
            c.MaxRetryCount = 5;             // retries before dead-letter
            c.PerQueueSettings["orders"] = new() { ConcurrencyLimit = 8, MaxRetryCount = 3 };
        });
    }));

    tyto.Endpoints(e => e.Add("orders-endpoint", ep => {
        TransportEndpoint queue = new() { TransportName = "rabbitmq", Address = "orders" };
        ep.ListenOn(queue);
        ep.Routing.Send<PlaceOrder>().To(new TransportEndpoint {
            TransportName = "rabbitmq", Address = "orders-exchange", RoutingKey = "place-order"
        });
        ep.AddHandlersFromAssembly(typeof(PlaceOrder).Assembly);
    }));
});

The declared topology (exchanges/queues/bindings) is created at startup by the topology manager. For sends, Address is the exchange and RoutingKey the routing key.

Retries & dead-lettering

On a handler failure the message is republished to the same queue with an incremented Tyto-Retry-Count header (and the original is acked), until MaxRetryCount is reached. At the limit it is dropped — and, if Tyto.DeadLettering is registered, reported to your dead-letter store first.

The dead-letter record captures the full failure context — reason, attempts, and the exception (type / message / stack).

Ordering caveat: in-place retry republishes to the same queue, so a failing message may be reordered relative to newer messages. This is the trade-off for transport-agnostic retry-then-dead-letter (vs. relying on RabbitMQ DLX + TTL).

Observability (OpenTelemetry)

Meter and ActivitySource are both named Tyto.Transports.RabbitMq.

tracing.AddSource("Tyto.Transports.RabbitMq");
metrics.AddMeter("Tyto.Transports.RabbitMq");

Traces: rabbitmq.send, rabbitmq.receive. Counters: published, received, completed, rejected; histogram: processing.duration. Logs are source-generated.