Table of Contents

Configuration

Bus Registration

BareWire separates the core engine from the transport adapter: the core (BareWire) contains the pipeline, flow control, and dispatch; a transport package (e.g. BareWire.Transport.RabbitMQ) implements the wire protocol. You always register both. There are three ways to do it.

1. Single call — bundle package (recommended)

Each transport ships a thin bundle package — BareWire.RabbitMQ, BareWire.Kafka, BareWire.AzureServiceBus, BareWire.AWS.SQS, BareWire.Google.PubSub — that depends on both the core and the matching transport and exposes a single AddBareWireWith{Transport} method:

builder.Services.AddBareWireWithRabbitMq(
    transport => transport.Host("amqp://guest:guest@localhost:5672/"),
    bus =>
    {
        bus.AddConsumer<MyConsumer>();
        // serializers, middleware, endpoints...
    });

The bus delegate is optional — omit it when transport defaults are enough:

builder.Services.AddBareWireWithRabbitMq(transport => transport.Host("amqp://localhost"));

This is the most ergonomic path for the common case of a single transport. Install one package (BareWire.RabbitMQ) instead of two, and register in one statement.

2. Two calls — core and transport registered separately

Register the transport adapter and the core explicitly. Use this when you reference the core and transport packages separately, or you want maximum control over package versions:

builder.Services.AddBareWireRabbitMq(transport =>
{
    transport.Host("amqp://guest:guest@localhost:5672/");
    transport.ReceiveEndpoint("my-queue", e => { /* ... */ });
});

builder.Services.AddBareWire(bus =>
{
    bus.AddConsumer<MyConsumer>();
});

AddBareWireWith{Transport} is exactly this pair behind one method, so the two paths are equivalent. The two-call form is fully supported and is the path to use when an application needs more than one transport — call AddBareWire once and register each transport on its own endpoints (a bundle call registers the core internally, so two bundle calls would register the core twice).

Deprecated: earlier versions configured the transport with cfg.UseRabbitMQ(...) inside the AddBareWire delegate. That marker is now an obsolete no-op — any host/credentials passed to it were silently ignored; the transport is configured only through AddBareWireRabbitMq (or the bundle). Calls still compile (with a warning) for one release; migrate to one of the forms above.

3. In-memory — tests

For unit and integration tests, BareWire.Testing provides an in-memory harness that needs no broker:

builder.Services.AddBareWireTestHarness(bus =>
{
    bus.AddConsumer<MyConsumer>();
});

See Custom Serializers and the testing guide for the harness API.

Why it is layered this way

  • The core never depends on a transport, and a transport never depends on the core. Both depend only on the abstractions. The transport implements a contract the core drives — the core calls the transport through an interface and is never called back into a specific broker. This keeps the engine swappable and the transports independently versionable.
  • The bundle is a separate layer on top of both. Rather than letting a transport reference the core (which would couple the two and let broker concerns leak into the engine), the single-call ergonomics live in a thin bundle package that references core + transport. The one-directional dependency rule is preserved; the convenience is additive. This is enforced by architecture tests, so the ergonomics can never silently erode the layering.

JSON Serializer

BareWire follows a raw-first approach — the default serializer produces raw JSON without an envelope. Register it with:

builder.Services.AddBareWireJsonSerializer();

This uses System.Text.Json internally with zero-copy IBufferWriter<byte> / ReadOnlySequence<byte> pipelines. No byte[] is allocated per-message in the hot path.

RabbitMQ Transport

Connection

The connection string is typically injected via Aspire or configuration:

// Via Aspire (automatic)
builder.AddRabbitMQClient("rabbitmq");

// Via connection string
rmq.Host("amqp://guest:guest@localhost:5672/");

Receive Endpoint Options

Each receive endpoint supports the following settings:

rmq.ReceiveEndpoint("my-queue", e =>
{
    e.PrefetchCount = 16;              // broker-level prefetch
    e.ConcurrentMessageLimit = 8;      // in-flight concurrency
    e.RetryCount = 3;                  // retry attempts before DLQ
    e.RetryInterval = TimeSpan.FromSeconds(1);

    e.Consumer<MyConsumer, MyMessage>();
});

Per-Key Consumer Ordering

A receive endpoint can preserve message order within a key while processing different keys in parallel. It is OFF by default — opt in per endpoint with one of:

rmq.ReceiveEndpoint("ordered-processing", e =>
{
    // Header-based (raw / cross-language) — leaves strategy at Auto
    e.OrderedByHeader("ordering-key");

    // Or the configurator block for full control
    e.OrderedBy(o =>
    {
        o.ByHeader("ordering-key");
        o.TransportAffinity(TransportAffinity.SingleActiveConsumer);
        o.MaxDeliveryAttempts(2);
    });

    e.Consumer<MyConsumer, MyMessage>();
});

Correlation-id key caveat. When no explicit key source is given, the ordering key falls back to the auto-stamped correlation-id. This only works when the correlation-id is stable per aggregate/entity and has appropriate cardinality: too few distinct values create a hot key that throttles parallelism; a value that changes per message gives no real affinity (each message is its own group); and the correlation-id is not stamped for plain PublishAsync/SendAsync, so under that traffic the message flows keyless (no ordering). Prefer an explicit ByHeader/By key source for predictable behavior.

See: Per-Key Consumer Ordering for strategies, transport affinity, fail-fast, and the end-to-end story.

Publish-Style Request Routing

By default request-response is send-style: the requester targets a fixed responder queue. Publish-style mode (opt-in, per request type) publishes the request to a per-type fanout exchange instead, so multiple responders can compete and the first response wins. Enable it with PublishRequest<T>(), optionally with an options block:

rmq.PublishRequest<CheckOrderStatus>();          // default name formatter

rmq.PublishRequest<CheckOrderStatus>(o =>
{
    o.ExchangeName = "Orders.Api:CheckOrderStatus"; // override the default name
    o.Strict       = true;                          // mandatory:true → fast "no responder bound"
    o.AutoDeclare  = true;                          // auto-declare the per-type fanout exchange
});
Option Type Default Meaning
ExchangeName string? null Overrides the default per-type exchange name. null uses the formatter (Namespace:TypeName, a literal colon, PascalCase). Required for generic / nested request types.
Strict bool false Publishes with mandatory: true; a return surfaces synchronously as a publish exception, turning a silent "no responder bound" into an immediate explicit error instead of a timeout. Opt-in because a brief "zero responders" window during a migration is expected.
AutoDeclare bool false Auto-declares the per-type fanout exchange. Off by default so no broker entity is created without consent; otherwise the exchange must be declared in ConfigureTopology.

The default exchange name follows the MassTransit convention Namespace:TypeName with a literal colon — it must match the responder's exchange exactly, or the request publishes to an exchange no responder listens on and times out.

To declare the per-type fanout exchange and bind a responder queue without spelling out the Namespace:TypeName name by hand, ITopologyConfigurator offers two opt-in convenience helpers — sugar over DeclareExchange + BindExchangeToQueue:

rmq.ConfigureTopology(t =>
{
    t.DeclareRequestExchange<CheckOrderStatus>();              // fanout Namespace:TypeName, durable
    t.BindRequestExchangeToQueue<CheckOrderStatus>("orders");  // bind a responder queue, no routing key
});

DeclareRequestExchange<T>() declares the fanout exchange (durable: true, autoDelete: false); BindRequestExchangeToQueue<T>(queue) binds the queue with an empty routing key (fanout ignores the key). They change nothing else — the same fail-fast validation and default-OFF posture apply. When AutoDeclare = true is set on PublishRequest<T>, the per-type fanout exchange is declared automatically at topology deploy (idempotently — declaring it explicitly with the helper as well does not create a duplicate).

See: Publishing and Consuming for the full competing-responders scenario, topology, and the first-in-wins caveats.

Topology Configuration

Use ConfigureTopology to declare exchanges, queues, and bindings. Queue arguments can be configured using the fluent IQueueConfigurator API:

rmq.ConfigureTopology(topology =>
{
    topology.DeclareExchange("orders", ExchangeType.Topic, durable: true);
    topology.DeclareQueue("orders", durable: true, autoDelete: false, configure: q =>
    {
        q.SetQueueType(QueueType.Quorum)
         .DeadLetterExchange("orders.dlx")
         .MessageTtl(TimeSpan.FromDays(7));
    });
    topology.BindExchangeToQueue("orders", "orders", routingKey: "#");
});

See: Topology for full details and all available IQueueConfigurator methods.

Flow Control Options

BareWire provides both consume-side and publish-side flow control. Register options via DI:

// Consume-side: credit-based flow control
builder.Services.AddSingleton(new FlowControlOptions
{
    MaxInFlightMessages = 50,
    MaxInFlightBytes = 1_048_576  // 1 MiB
});

// Publish-side: bounded outgoing channel
builder.Services.AddSingleton(new PublishFlowControlOptions
{
    MaxPendingPublishes = 500
});

See: samples/BareWire.Samples.BackpressureDemo/Program.cs

SAGA Persistence

A saga needs two registrations — the repository (persistence) and the state machine (which wires the dispatcher into the consume pipeline) — plus a receive endpoint to host it:

// Repository (EF Core) — BareWire.Saga.EntityFramework
builder.Services.AddBareWireSaga<OrderSagaState>(
    options => options.UseNpgsql(connectionString));
// Or with SQLite:
// builder.Services.AddBareWireSaga<OrderSagaState>(options => options.UseSqlite("Data Source=saga.db"));

// State machine — BareWire.Saga
builder.Services.AddBareWireSagaStateMachine<OrderSagaStateMachine, OrderSagaState>();

// Host it on an endpoint (pass the state-machine type):
rmq.ReceiveEndpoint("order-saga", e => e.StateMachineSaga<OrderSagaStateMachine>());

See Saga State Machines for the full DSL, and samples/BareWire.Samples.SagaOrderFlow/Program.cs for the complete setup.

Transactional Outbox

Configure the outbox with a database provider and polling settings:

builder.Services.AddBareWireOutbox(
    configureDbContext: options => options.UseNpgsql(connectionString),
    configureOutbox: outbox =>
    {
        outbox.PollingInterval = TimeSpan.FromSeconds(1);
        outbox.DispatchBatchSize = 100;
    });

See: samples/BareWire.Samples.TransactionalOutbox/Program.cs

Observability

Enable OpenTelemetry integration:

builder.Services.AddBareWireObservability(cfg =>
{
    cfg.EnableOpenTelemetry = true;
});

See: samples/BareWire.Samples.ObservabilityShowcase/Program.cs

Service Defaults

For consistent observability and health check setup across multiple services, use the shared AddServiceDefaults() extension:

builder.AddServiceDefaults();  // during DI setup
// ...
app.MapServiceDefaults();      // after app.Build()

This registers OpenTelemetry tracing/metrics, OTLP exporter, and health check endpoints:

  • /health — combined liveness + readiness
  • /health/live — liveness only
  • /health/ready — full readiness including dependencies

See: samples/BareWire.Samples.ServiceDefaults/ServiceDefaultsExtensions.cs

Edit this page