Bsft.Messaging 2.2.0

Bsft.Messaging

The core of the Bsft.Messaging family: integration event staging, the declared event vocabulary and its payload encoding, the transactional-outbox dispatcher, and the consuming pipeline that puts an inbox in front of the handlers.

Install

dotnet add package Bsft.Messaging

Requires Bsft.Messaging.Abstractions (pulled transitively), Bsft.Core for the polling loop the dispatcher runs on, and Bsft.Instrumentation for the consumer's instrumentation source. It holds nothing durable itself: the outbox and the durable inbox live in Bsft.Messaging.EFCore, and a broker lives in Bsft.Messaging.RabbitMQ. Without either, the in-memory bus and the in-memory inbox are what a host gets.

Register

services.AddBsftMessaging(options =>
{
    options.AddEvent<OrderPlaced>("orders.placed.v1");
    options.Consumer.ConsumerName = "orders";
});

// or bound from configuration, section "Bsft:Messaging"
services.AddBsftMessaging(configuration, options => options.AddEvent<OrderPlaced>("orders.placed.v1"));

The event vocabulary is declared in code because it names CLR types; everything else binds from the section. Consuming is opt-in and needs three more calls: AddBsftIntegrationEventHandler<TEvent, THandler>() per handler, AddBsftConsumerHost() for the background service that reads the transport, and either a broker adapter or AddBsftMessagingLoopback() for the transport itself — both of them before AddBsftMessaging, which registers the in-memory bus with TryAdd. AddBsftInbox() registers the duplicate guard on its own where the host runs the delivery itself.

Configure

Validation: ValidateOnStart() rejects an attempt budget below one on either side, a non-positive receive-restart delay, poll interval or lease duration, an inbox-ledger or parked-message capacity below one, a negative retry delay or retry backoff, a retry ceiling that is not positive on either side, and an inbox retention that is set and not positive — all of them reported in one message rather than one restart at a time. ConsumerName is deliberately not among them: a host that publishes without consuming has no use for it, so the consumer host rejects an absent name instead.

Extend

Every seam of the family is declared in Bsft.Messaging.Abstractions and listed there, so a replacement can be written without referencing this package. What this package supplies is the default implementation of each: IntegrationEventTypeRegistry and JsonIntegrationEventSerializer for the vocabulary and its encoding, IntegrationEventCollector for both sides of the staging contract, InMemoryMessageBus and LoopbackMessageTransport for the transport, InMemoryInboxStore over the process-wide InMemoryInboxLedger for the inbox, IncomingMessageDecoder, IntegrationEventHandlerInvoker and InboxGuardedDelivery for the consuming steps, and IntegrationEventConsumer for the three transport outcomes. Two of those defaults expect replacing in anything durable: the in-memory bus handles inside the publish and has no inbox at all, and the in-memory inbox forgets every event when the process stops.

IOutboxProcessor is not a replacement seam but a way in: ProcessOnceAsync runs one drain pass of the transactional outbox instead of waiting for the polling loop. AddBsftMessagingEFCore registers it on the same OutboxDispatcher the host runs as a background service, so a test drains what the application would have published. A pass takes one batch, so repeat until it returns zero.

IConsumerTelemetry is the one seam this package declares itself. It stays here rather than in the abstractions because it names MessageParkReason against a meter and an activity source that only the consuming pipeline has; replace it to record deliveries somewhere other than the shipped Bsft.Messaging instrumentation source.

Documentation

Topic page: docs/messaging.md. Conventions: docs/package-conventions.md.