Skip to content
Prague
Improve this page

React to Changes

Goal: maintain a derived structure (here, open order total per customer) that follows the cache as messages arrive, without polling.

Prerequisite: a cache registered through AddKafkaCaches. The code is in ReactToChanges.cs; the registration block is compiled-only, the rest runs as tests.

1. Implement the handler

ICacheAfterHandler<TKey, TValue>.Handle is called after the cache applied a message in the live phase. Every branch of UpdateType is reachable; the arguments differ per case.

public sealed class OrderProjector : ICacheAfterHandler<string, Order> {
    private readonly Dictionary<int, decimal> _openTotalByCustomer = new();   // only this worker writes it

    public IReadOnlyDictionary<int, decimal> OpenTotalByCustomer => _openTotalByCustomer;

    public ValueTask Handle(UpdateType updateType, string key, Order? newValue, Order? oldValue) {
        switch (updateType) {
            case UpdateType.Add:                                  // oldValue is null
                Apply(newValue!, +1);
                break;
            case UpdateType.Update:                               // both values set
                Apply(oldValue!, -1);
                Apply(newValue!, +1);
                break;
            case UpdateType.Delete:                               // newValue is null; oldValue is what was removed
                Apply(oldValue!, -1);
                break;
            case UpdateType.Same:                                 // a re-emit of the resident value: nothing to do
            case UpdateType.Filtered:                             // carries NO key and NO values — never read `key` here
                break;
        }

        return ValueTask.CompletedTask;                           // keep it short: it runs on the cache's live worker
    }

    private void Apply(Order order, int sign) {
        if (order.Status is not (OrderStatus.Pending or OrderStatus.Paid))
            return;
        _openTotalByCustomer[order.CustomerId] = _openTotalByCustomer.GetValueOrDefault(order.CustomerId) + sign * order.Total;
    }
}
UpdateType newValue oldValue When
Add set null key was not resident
Update set set key was resident and CacheEquals said different
Same set null key was resident and CacheEquals said equal
Delete null set tombstone or a treatAsDelete filter, key was resident
Filtered null null a filter skipped the message; key is default too

The handler runs on the cache's own live worker, sequentially with the other handlers of that cache; exceptions are caught and logged, not propagated. The initial replay never calls it: when Handle runs for the first time, the cache already holds the topic.

2. Register it

Type registration makes it a singleton ICacheAfterHandler<string, Order>; the factory form is for handlers with constructor arguments DI cannot supply on its own.

public static void Register(KafkaCacheHandlersBuilder b) {
    b.AddCache<OrderCache, string, Order>()
        .WithAfterHandler<OrderProjector>()                 // singleton ICacheAfterHandler<string, Order>
        .WithAfterHandler(sp => new OrderAuditLog(sp.GetRequiredService<TimeProvider>()));   // factory form
}

3. Test it by hand

The consumer is the only caller in production; a test is the other one. Drive the sequence a topic would produce.

var projector = new OrderProjector();
var pending = new Order { OrderId = "ORD-001", CustomerId = 42, CreatedAtUnixMs = 1, Status = OrderStatus.Pending, Total = 100m };
var paid = new Order { OrderId = "ORD-001", CustomerId = 42, CreatedAtUnixMs = 1, Status = OrderStatus.Paid, Total = 100m };
var shipped = new Order { OrderId = "ORD-001", CustomerId = 42, CreatedAtUnixMs = 1, Status = OrderStatus.Shipped, Total = 100m };

await projector.Handle(UpdateType.Add, "ORD-001", pending, null);        // +100
await projector.Handle(UpdateType.Same, "ORD-001", pending, null);       // re-emit: no change
await projector.Handle(UpdateType.Update, "ORD-001", paid, pending);     // still open: -100 +100
await projector.Handle(UpdateType.Update, "ORD-001", shipped, paid);     // leaves the open set: -100
await projector.Handle(UpdateType.Filtered, default!, null, null);       // what a filtered message looks like

4. Same versus Update

A replayed or duplicated message is common on Kafka. The cache compares the incoming value with the resident one using the generated CacheEquals (deep, property by property, collections included); an equal value is reported as Same and the resident instance is not swapped.

var first = NewProduct(name: "Widget");
products.AddOrUpdate(first);

var reEmit = NewProduct(name: "Widget");                               // a new instance, same content (a replayed message)
var changed = products.AddOrUpdate(reEmit, out var previous);          // false → the live phase reports Same
var renamed = products.AddOrUpdate(NewProduct(name: "Gadget"), out previous);   // true → Update, previous = the Widget

var equal = first.CacheEquals(reEmit);                                 // the generated deep equality, callable directly

AddOrUpdate returns false for Same, which is exactly the signal the live phase uses. Conditional Updates covers equality in depth.

Pitfalls

  • Reading key on Filtered. It is default, so for a string key it is null. Return early.
  • Awaiting I/O in Handle. Every message of that cache waits behind it, and the liveness check reports HandlerProcessingTimeout after 5 s by default. Enqueue to a channel and return.
  • Expecting the replay. If you need the derived structure to reflect the initial load, build it from the cache after LoadingCompletion and use the handler only for deltas.
  • Mutating newValue. It is the instance the cache holds and every query returns. Treat it as read-only; Clone() it if you must change it (see Map, Clone, Batch).
Prague documentation