Every Tenth Tick Becomes an Order

Part 4 ended with a warning label: FeedR's Redis pub/sub tier delivers messages to whoever happens to be listening and forgets them instantly. That's the right economics for price ticks. It would be malpractice for an order. The moment a system says “an order has been placed”, somebody downstream is going to bill, notify, or ship against that statement — and “the notifier was restarting, so the order never happened” is not an acceptable sentence in that world.

This is where FeedR — the open-source DevMentors sample this series reads — grows its second messaging tier. Same publish/subscribe verbs, entirely different guarantees, and a different broker: Apache Pulsar. This part follows a currency tick into the aggregator, watches it become an OrderPlaced integration event, and rides it through Pulsar to the notifier. Along the way: envelopes, message metadata, producer caching, and an acknowledgement loop — the vocabulary of durable messaging, each item earning its place.

The aggregator: consume, decide, promote

The aggregator subscribes to the pricing channel like any other stream consumer (part 4's seam), and hands each tick to a handler whose job is the decision:

internal sealed class PricingHandler : IPricingHandler
{
    private int _counter;

    public async Task HandleAsync(CurrencyPair currencyPair)
    {
        if (ShouldPlaceOrder())
        {
            var orderId = Guid.NewGuid().ToString("N");
            var integrationEvent = new OrderPlaced(orderId, currencyPair.Symbol);
            await _messagePublisher.PublishAsync("orders", integrationEvent);
        }
    }

    private bool ShouldPlaceOrder() => Interlocked.Increment(ref _counter) % 10 == 0;
}

The business logic is a placeholder and says so in a source comment — every tenth tick, place an order. Fine; the shape is what matters. Three things in this handler are the real curriculum:

1. The promotion is explicit. A tick arrives on the transient tier (IStreamSubscriber), a fact departs on the durable tier (IMessagePublisher). The handler is the checkpoint where data changes citizenship, and it's one readable method — not a config flag on a shared bus.
2. OrderPlaced is an integration event, and it's tiny: record OrderPlaced(string OrderId, string Symbol) : IMessage. Just the fact and its identifiers, no fat payload. Consumers who need more can ask (if there were an API to ask — in the sample there isn't, which is honest about its scope).
3. The IMessage marker interface gates the durable tier at compile time: PublishAsync<T> requires T : IMessage. You cannot accidentally shove a raw CurrencyPair tick onto the orders topic — the type system remembers which tier things belong to. Cheap trick, real fence.

The second seam: same verbs, heavier envelope

The messaging abstraction in FeedR.Shared mirrors the streaming one — with two telling differences:

public interface IMessagePublisher
{
    Task PublishAsync<T>(string topic, T message) where T : class, IMessage;
}

public interface IMessageSubscriber
{
    Task SubscribeAsync<T>(string topic, Action<MessageEnvelope<T>> handler) where T : class, IMessage;
}

public record MessageEnvelope<T>(T Message, string CorrelationId) where T : IMessage;

Difference one is the marker constraint. Difference two is that subscribers don't receive naked payloads — they receive an envelope carrying the message plus its correlation ID. Transient ticks travel bare; durable facts travel with paperwork. That asymmetry, deliberate or not, encodes something true: the further a message outlives its moment, the more context it has to carry with it, because the context won't be reconstructible later.

And exactly like the streaming seam, the messaging seam ships no-op defaults (AddMessaging()) that the Pulsar registration (AddPulsar()) overrides via DI last-wins. The notifier can boot and serve HTTP with no broker anywhere in sight. Same swappability design, applied twice — by the second occurrence it's a house style.

Pulsar, and the publisher's anatomy

Why Pulsar and not the RabbitMQ or Kafka you'd expect in a .NET sample? Partly the DevMentors team's stated mission — the repo exists to “explore and play with different tools” — but it's a defensible pick on merits: Pulsar is a log-based broker like Kafka (messages persist, consumers track positions, subscriptions replay) with a built-in per-subscription acknowledgement model that feels closer to a traditional queue, multi-tenancy out of the box, and a one-container standalone mode that makes local dev painless. For this workload — durable facts, named subscriptions, competing consumers later if you need them — it slots in cleanly. The client library is DotPulsar, the official native .NET client.

The publisher (PulsarMessagePublisher, condensed from the actual source) is where the practical patterns live:

private readonly ConcurrentDictionary<string, IProducer<ReadOnlySequence<byte>>> _producers = new();

public async Task PublishAsync<T>(string topic, T message) where T : class, IMessage
{
    var producer = _producers.GetOrAdd(topic, _client.NewProducer()
        .ProducerName(_producerName)
        .Topic($"persistent://public/default/{topic}")
        .Create());

    var correlationId = _contextAccessor.HttpContext?.GetCorrelationId()
                        ?? Guid.NewGuid().ToString("N");
    var metadata = new MessageMetadata
    {
        ["custom_id"] = Guid.NewGuid().ToString("N"),
        ["producer"] = _producerName,
        ["correlationId"] = correlationId,
    };
    var messageId = await producer.Send(metadata, _serializer.SerializeBytes(message));
}

Reading list, top to bottom:

• Producer-per-topic, cached in a ConcurrentDictionary. Pulsar producers are stateful, connection-holding objects — creating one per publish would be brutal. GetOrAdd amortizes creation across the process lifetime. (Pedantic footnote: under a race, GetOrAdd's value factory can run twice and one producer leaks unclosed. Lazy<T> inside the dictionary is the textbook fix. It will never matter here; it has mattered in systems I've run.)
• persistent://public/default/{topic} — Pulsar's fully qualified topic naming: persistence mode, tenant, namespace, topic. The sample hard-codes tenant and namespace; the persistent:// prefix is the entire durability request. One string prefix is the difference between this tier and part 4's.
• Metadata rides outside the payload. Producer name (derived from the entry assembly), a unique message ID, and the correlation ID travel as Pulsar message properties — consumers can read them without deserializing the body, and infrastructure (dead-letter tooling, tracing) can see them without knowing your types. This is the correct layer for that data, and part 8 leans on it hard.
• The correlation fallback is a quiet bug. HttpContext?.GetCorrelationId() is null in a background-service call path — which is precisely where the aggregator's handler runs — so the publisher mints a fresh GUID. The order's correlation ID has no relationship to the request that started the pricing feed. The trace breaks silently, right here, and part 8 dissects why.
• The client is built in the constructor with defaults (PulsarClient.Builder().Build() — localhost:6650), with a TODO in the source acknowledging the missing options type. The Redis options pattern from part 4 is the template it should copy.

The notifier: an await foreach with a receipt

On the consuming end, the notifier registers a BackgroundService that delegates to PulsarMessageSubscriber, whose core is this loop (condensed):

var consumer = _client.NewConsumer()
    .SubscriptionName($"{_consumerName}_{topic}")
    .Topic($"persistent://public/default/{topic}")
    .Create();

await foreach (var message in consumer.Messages())
{
    var payload = _serializer.DeserializeBytes<T>(message.Data.FirstSpan.ToArray());
    if (payload is not null)
    {
        handler(new MessageEnvelope<T>(payload, message.Properties["correlationId"]));
    }

    await consumer.Acknowledge(message);
}

Two lines carry the semantics. The subscription name — feedr.notifier_orders — is durable server-side state: Pulsar remembers, per subscription, what's been acknowledged. Kill the notifier, publish ten orders, restart it: the subscription's backlog delivers all ten. That is the whole difference from part 4 in one observable behavior, and it's worth actually performing on your machine — stop the notifier, generate orders, start it, and watch the backlog drain into the log.

The Acknowledge call is the receipt. Note it runs after the handler and unconditionally — if the handler throws, the message is still… well, actually, no: an exception would skip the acknowledge and kill the loop, taking the whole consumer down with no retry, no dead-letter, no supervision. And since the handler is a synchronous Action<> receiving fire-and-forgotten async work (the same signature critique as part 4), a handler that queues async work gets its message acknowledged before the work runs — at-least-once delivery downgraded to at-most-once processing by an API signature. The production checklist writes itself: Func<T, Task> handlers, acknowledge only after successful handling, negative-acknowledge on failure, a retry/DLQ policy, and a supervisor that restarts a faulted consume loop.

The two-tier map, complete

With this part, FeedR's full eventing economics are on the table:

Most architectures I meet have exactly one bus and force both kinds of traffic through it — paying log storage for telemetry, or gambling business facts on fire-and-forget. FeedR, a teaching sample under two thousand lines, keeps two tiers and makes the type system police the border. That's the idea I'd steal even if I never used Pulsar or Redis at all.

Next, part 8 pulls the thread that's been fraying through every part so far: the correlation ID. It's born in the YARP gateway, rides HTTP headers and Pulsar metadata — and dies, silently, in the middle of the system. Tracing exactly where is the best observability lesson the sample has to offer.