An Event Bus in Seven Files

Message brokers feel like infrastructure you buy, not code you read. RabbitMQ, Azure Service Bus, Kafka — each hides the essential moves behind a connection string. The DevMentors ModularMonolith estate does something more instructive: it builds a working in-process event bus out of System.Threading.Channels in seven files small enough to read in one sitting, and in doing so gives every enterprise-integration pattern a face. Part 4 covered how this machinery boots; this part reads the mechanism itself.

The seven files live in Shared.Infrastructure/Messaging/: MessageChannel, IMessageChannel, AsynchronousDispatcher, IAsynchronousDispatcher, AsynchronousDispatcherJob, InMemoryMessageBroker, MessagingOptions — plus an Extensions.cs that registers them. All of it arrived in one commit, 60840b2, the estate's last.

The channel

The heart is barely a class:

internal sealed class MessageChannel : IMessageChannel
{
    private readonly Channel<IMessage> _messages = Channel.CreateUnbounded<IMessage>();

    public ChannelReader<IMessage> Reader => _messages.Reader;
    public ChannelWriter<IMessage> Writer => _messages.Writer;
}

Registered as a singleton, this is a literal EIP Message Channel: one shared conduit that producers write to and a consumer reads from, decoupling the two in time. Two design decisions hide in its two meaningful lines. It is untyped — a single channel of IMessage, not a channel per message type — so in Hohpe and Woolf's vocabulary it is a plain channel rather than a Datatype Channel; routing by type happens downstream. And it is unbounded: CreateUnbounded means writers never wait, which also means there is no backpressure anywhere in the system. A publisher that outruns the consumer grows the queue without limit; memory is the only regulator. Channel.CreateBounded<IMessage>(capacity) with a wait-on-full policy is one line away, and choosing it would force the interesting question — block the HTTP request or shed events? — that unbounded lets you defer indefinitely.

The interface exposes the raw ChannelReader/ChannelWriter rather than wrapping them. That is unusually honest plumbing: no abstraction tax, at the cost of any consumer being able to complete the writer or steal reads. In a sample, transparency wins.

The producer side

AsynchronousDispatcher is enqueue-and-return:

public async Task PublishAsync<TMessage>(TMessage message) where TMessage : class, IMessage
    => await _messageChannel.Writer.WriteAsync(message);

On an unbounded channel WriteAsync completes synchronously in practice, so publishing costs the caller almost nothing — fire-and-forget with the forgetting made structural. The HTTP request that publishes ConferenceCreated resumes immediately; whatever happens next is the consumer's problem, in every sense that part 8 will make precise.

The pump

The consumer is a stock BackgroundService, and it is the file to read slowly:

protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
    _logger.LogInformation("Running the async dispatcher job.");

    await foreach (var message in _messageChannel.Reader.ReadAllAsync(stoppingToken))
    {
        try
        {
            await _moduleClient.PublishAsync(message);
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, ex.Message);
        }
    }

    _logger.LogInformation("Finished running the async dispatcher job.");
}

• await foreach over ReadAllAsync is the idiomatic channel consumer in 2021 C# and still today — no polling, no manual WaitToReadAsync loop; the iterator suspends until a message arrives.
• One consumer, sequential dispatch. There is exactly one pump, and it awaits each PublishAsync before reading the next message. That is Fowler's Singular Update Queue: all cross-module traffic serialises through a single point, which incidentally buys global event ordering — a property distributed systems pay dearly for, obtained here as a side effect of being one process.
• The catch logs and drops. A message whose handler throws is written to the log at Error level and never seen again. No retry, no dead-letter channel, no poison-message counter. This is at-most-once delivery, chosen implicitly by the shape of a try/catch.
• The last line never runs. When the host stops, ReadAllAsync honours stoppingToken by throwing OperationCanceledException, which exits ExecuteAsync past the final log statement. “Finished running the async dispatcher job.” is unreachable in the normal shutdown path — a two-line tell that this code was never watched during a graceful stop. Harmless, and exactly the kind of detail an estate with tests or an ops runbook would have caught.

In EIP terms the pump is an Event-Driven Consumer feeding the publish-subscribe distribution that the module registry performs (part 6 follows a message through it).

The bus as a checklist of absences

Line the estate up against the Message Construction and reliability columns of the EIP catalogue and the reading gets sharper, because IMessage — the currency every file above trades in — is this:

// Marker
public interface IMessage
{
}

No message id, no correlation id, no timestamp, no version. Each absent field is an absent pattern, and each becomes concrete work the day any module needs it:

That table is not a demolition; it is the syllabus. The seven files implement precisely the subset of a message broker that demonstrates the architecture, and omit precisely the subset that makes one operable — and having both lists explicit is worth more to a learner than either a toy that hides the gaps or a product that hides the mechanism. The same authors' production-grade answer to this table is Convey, whose outbox and inbox fill several rows at once.

What the channel does not do is decide who receives a message or what type it arrives as. That is the delivery half of the spine — a registry keyed on class names and a translator made of JSON — and it is where the estate's cleverest and most dangerous idea lives. Serialize, deserialize, deliver is next.