TCIS.EventBus 1.0.0-rc.18

TCIS.EventBus

Transactional messaging for the TCIS ecosystem: an outbox that writes the message in the same transaction as your business data, a background sender, and a consumer side with retries and a poison-message path.

Transport bindings live in separate packages — see TCIS.EventBus.Kafka. For choosing between mediator.Publish and the EventBus, see TCIS.Mediator.

Table of contents

1. Registration

builder.Services.AddTEventBus(options =>
{
    builder.Configuration.GetSection("TCIS:EventBus").Bind(options);

    options.UseSqlServer(connectionString);      // outbox lives next to your data
    options.UseKafka(kafka =>
        builder.Configuration.GetSection("TCIS:EventBus:Kafka").Bind(kafka));
});

Three things must be registered or startup throws: the bus itself, one storage provider, and one transport provider.

{
  "TCIS": {
    "EventBus": {
      "DefaultGroupName": "tcis.queue.vessel-service",
      "Version": "v1",
      "FailedRetryCount": 10,
      "FailedRetryInterval": 30,
      "SucceedMessageExpiredAfter": 86400,
      "ConsumerShutdownTimeoutSeconds": 5,
      "Kafka": {
        "Servers": "broker:9092",
        "MainConfig": { "group.id": "tcis.subscriber.vessel-group" }
      }
    }
  }
}

2. Publishing

public sealed class BerthVesselHandler(IEventPublisher bus, VesselDbContext db)
{
    public async Task Handle(BerthVessel cmd, CancellationToken ct)
    {
        // The bus joins the EF transaction: the outbox row and the business row commit together.
        using var tx = db.Database.BeginTransaction(bus, autoCommit: false);

        visit.Status = "BERTHED";
        await db.SaveChangesAsync(ct);

        await bus.PublishAsync("vessel.berthed", new VesselBerthed(visit.Id, visit.BerthCode));

        await tx.CommitAsync(ct);
    }
}

BeginTransaction(publisher, autoCommit) is an extension shipped by the storage package (TCIS.EventBus.SqlServer, TCIS.EventBus.PostgreSql); overloads exist for DatabaseFacade and for a raw IDbConnection.

The point of the outbox is that PublishAsync does not talk to the broker. It writes a row in the same transaction. If the transaction rolls back, the message was never published. If the process dies after commit but before send, a background processor picks the row up and sends it.

Publishing outside a transaction is allowed and sends immediately after persisting — use it only when there is no business data to keep in step.

3. Subscribing

public sealed class VesselSubscriber(IVesselService service) : IEventSubscribe
{
    [EventSubscribe("vessel.berthed")]
    public async Task OnBerthed(VesselBerthed evt, CancellationToken ct)
    {
        await service.RecordBerthingAsync(evt.VisitId, evt.BerthCode, ct);
    }
}

builder.Services.AddScoped<VesselSubscriber>();   // discovery reads DI registrations

Subscribers are discovered from the service collection, so the class must be registered — an unregistered subscriber is not an error, it simply never runs.

Each subscriber runs in its own DI scope, not the publisher's. A throwing subscriber does not affect the publisher; it affects only its own message.

4. Headers and context propagation

Ambient context travels in headers, so subscribers do not need it in the payload:

Keep the DTO clean — do not copy TenantId or SiteCode into the payload.

5. ⚠️ What happens to a message that fails

Three distinct outcomes. Confusing them is the usual reason an incident takes hours instead of minutes.

The subscriber throws

Retried in place up to 3 times, then handed to the background retry processor, which retries every FailedRetryInterval seconds up to FailedRetryCount times. After that the message stays Failed in storage and FailedThresholdCallback fires.

The message cannot be routed or deserialized

No subscriber matches the topic+group, or the payload does not parse. Retrying cannot help — the next attempt fails identically. The message is stored as an exception message, the offset is committed, and the stream continues. One Error log line records the reason:

Message parked and skipped: it could not be routed or deserialized, so the offset was committed
to let the stream continue. Id:…, Name:…, Group:…, Reason:SubscriberNotFoundException-->…

The payload parses but does not match the DTO

Nothing is reported. System.Text.Json ignores unknown JSON properties and leaves absent ones at their default, so the subscriber receives a half-empty object and the message is marked Succeeded. Property names are matched case-insensitively — but a genuinely different name (vessel_name versus VesselName) still will not bind.

This one has no library-level defence. Validate required fields inside the subscriber:

if (string.IsNullOrWhiteSpace(evt.VesselName))
    throw new TValidationException("VALIDATION_MISSING_VESSEL_NAME", "vessel_name is empty");

That converts a silent empty row into a parked message with a log line.

6. Consuming from a third-party producer

External producers do not emit TCIS headers. The transport fills in what is missing — see the Kafka README for how t-msg-id is derived and why it stays stable across redeliveries.

Two things to get right on the TCIS side:

1. Do not set TopicNamePrefix. It would rewrite the partner's topic name into prefix.partner.topic, which does not exist.
2. Match the payload explicitly. Declare [JsonPropertyName("...")] on the DTO rather than hoping the partner's naming matches yours. When the schema is not yet known, take a JsonElement first and log raw.GetRawText() to see exactly what arrives.

7. Storage choices

InMemory is for development and for stateless gateways where losing an un-sent message is acceptable. Anything with business meaning belongs in a durable store.

Two InMemory specifics worth knowing:

• Messages are held for SucceedMessageExpiredAfter (24 h by default) before the collector removes them. That is 24 hours of messages resident in RAM — lower it for a high-volume consumer.
• The dictionaries are static, so every host in the same process shares one store. This is fine for a single application, but it means two hosts side by side (or parallel integration tests) see each other's messages.

8. Delivery guarantees

At-least-once, with one condition. The offset is committed only after the subscriber finishes, so a crash mid-processing means the broker redelivers.

That condition breaks when EnableSubscriberParallelExecute = true: messages are then handed to an in-memory channel and the offset is committed straight away. A crash loses whatever is still in the channel. Enable it only when the throughput matters more than those messages do.

Because delivery is at-least-once, subscribers must be idempotent. Use t-msg-id as the deduplication key.

9. Options reference

10. Pitfalls