implementing-pubsub-pattern

.NET Pub-Sub Pattern

A guide for Pub-Sub patterns for event-based asynchronous communication.

Quick Reference: See QUICKREF.md for essential patterns at a glance.

1. Core APIs

API	Purpose	NuGet
System.Reactive (Rx.NET)	Reactive event streams	System.Reactive
System.Threading.Channels	Async Producer-Consumer	BCL
IObservable<T>	Observable sequence	BCL

---

2. System.Threading.Channels

2.1 Basic Usage

using System.Threading.Channels;

public sealed class MessageProcessor
{
    private readonly Channel<Message> _channel =
        Channel.CreateUnbounded<Message>();

    // Producer - Send message
    public async Task SendAsync(Message message)
    {
        await _channel.Writer.WriteAsync(message);
    }

    // Consumer - Process message
    public async Task ProcessAsync(CancellationToken ct)
    {
        await foreach (var message in _channel.Reader.ReadAllAsync(ct))
        {
            await HandleMessage(message);
        }
    }

    // Channel completion signal
    public void Complete() => _channel.Writer.Complete();
}

2.2 Bounded Channel (Backpressure Control)

// Backpressure control with buffer size limit
var options = new BoundedChannelOptions(capacity: 100)
{
    FullMode = BoundedChannelFullMode.Wait, // Wait when full
    SingleReader = true,
    SingleWriter = false
};

var channel = Channel.CreateBounded<Message>(options);

// Writer waits until space is available
await channel.Writer.WriteAsync(message);

2.3 Multiple Consumer Pattern

public sealed class WorkerPool
{
    private readonly Channel<WorkItem> _channel;
    private readonly int _workerCount;

    public WorkerPool(int workerCount = 4)
    {
        _workerCount = workerCount;
        _channel = Channel.CreateUnbounded<WorkItem>();
    }

    public async Task StartAsync(CancellationToken ct)
    {
        var workers = Enumerable.Range(0, _workerCount)
            .Select(_ => ProcessAsync(ct));

        await Task.WhenAll(workers);
    }

    private async Task ProcessAsync(CancellationToken ct)
    {
        await foreach (var item in _channel.Reader.ReadAllAsync(ct))
        {
            await ProcessItem(item);
        }
    }

    public ValueTask EnqueueAsync(WorkItem item) =>
        _channel.Writer.WriteAsync(item);
}