Event-driven

In an event-driven system, parts announce what happened and others react, without the announcer knowing who's listening. This page publishes Pedal's dock statuses to three subscribers, adds a fourth without touching the docks, sends a failing consumer's events to dead letters, and shows how a morning rush turns into consumer lag.

1. In one picture

In an event-driven architecture, parts announce events (facts about what happened: "dock D12 now has 1 bike", "ride R1 ended") and other parts react to them. The publisher doesn't call anyone and doesn't know who's listening; a broker or log (Kafka, RabbitMQ, a cloud event bus) carries events to subscribers.

Events also differ in how much they carry: a thin notification ("dock D12 changed", go and look) or a fat event with the full new state, so consumers never need to call back. See Architecture module 14 for the architect's view and System design module 18 for delivery guarantees.

2. Pedal built this way

Every dock publishes its status every few seconds. Follow one status update to three subscribers, one of which is broken.

The price of this freedom: the map is eventually consistent (it shows what the last processed event said, which may be seconds old), events can arrive twice or out of order, and nobody can read the whole flow in one place. Consumers must be idempotent and handle order (by sequence numbers or versions).

3. The rules

4. How it goes wrong

5. The code

A small broker with subscribers for dock.status: an availability map, a rebalancing planner and a broken analytics consumer. A city feed subscribes later without any change to the publisher. Failed deliveries are retried three times, then dead-lettered.

from __future__ import annotations
from collections import defaultdict
from typing import Callable

Handler = Callable[[dict], None]


class EventBroker:
    """Broker topology: publishers and subscribers know only topics, never each other."""

    def __init__(self, max_attempts: int = 3) -> None:
        self.subs: dict[str, list[tuple[str, Handler]]] = defaultdict(list)
        self.dead_letters: list[str] = []
        self.max_attempts = max_attempts

    def subscribe(self, topic: str, name: str, handler: Handler) -> None:
        self.subs[topic].append((name, handler))

    def publish(self, topic: str, event: dict) -> None:
        for name, handler in self.subs[topic]:
            for attempt in range(1, self.max_attempts + 1):
                try:
                    handler(event)
                    break
                except Exception:
                    if attempt == self.max_attempts:
                        self.dead_letters.append(f"{name} <- {topic} {event['dock']}")


# ---- Subscribers: each keeps its own view of the world. ----
class AvailabilityMap:
    def __init__(self) -> None:
        self.bikes: dict[str, int] = {}

    def on_status(self, e: dict) -> None:
        self.bikes[e["dock"]] = e["bikes"]


class Rebalancer:
    def __init__(self) -> None:
        self.jobs: list[str] = []

    def on_status(self, e: dict) -> None:
        if e["bikes"] < 2:
            self.jobs.append(f"bring bikes to {e['dock']}")
        elif e["bikes"] > e["slots"] - 2:
            self.jobs.append(f"take bikes from {e['dock']}")


class CityFeed:                               # added later; the docks' code never changed
    def __init__(self) -> None:
        self.sent = 0

    def on_status(self, e: dict) -> None:
        self.sent += 1


def flaky_analytics(e: dict) -> None:         # always fails: shows retries and dead letters
    raise ConnectionError("analytics store down")


if __name__ == "__main__":
    broker = EventBroker()
    amap, crew = AvailabilityMap(), Rebalancer()
    broker.subscribe("dock.status", "map", amap.on_status)
    broker.subscribe("dock.status", "rebalancer", crew.on_status)
    broker.subscribe("dock.status", "analytics", flaky_analytics)
    # The docks publish facts; they don't know who listens.
    broker.publish("dock.status", {"dock": "D12", "bikes": 1, "slots": 20})
    broker.publish("dock.status", {"dock": "D40", "bikes": 19, "slots": 20})
    feed = CityFeed()
    broker.subscribe("dock.status", "city-feed", feed.on_status)
    broker.publish("dock.status", {"dock": "D41", "bikes": 8, "slots": 20})
    print("map:", amap.bikes)
    print("crew jobs:", crew.jobs)
    print("city feed events:", feed.sent)
    print("dead letters:", broker.dead_letters)

interface StatusEvent { dock: string; bikes: number; slots: number; }
type Handler = (e: StatusEvent) => void;

// Broker topology: publishers and subscribers know only topics, never each other.
class EventBroker {
  private subs = new Map<string, [string, Handler][]>();
  deadLetters: string[] = [];
  constructor(private maxAttempts = 3) {}
  subscribe(topic: string, name: string, handler: Handler): void { this.subs.set(topic, [...(this.subs.get(topic) ?? []), [name, handler]]); }
  publish(topic: string, e: StatusEvent): void {
    for (const [name, handler] of this.subs.get(topic) ?? []) {
      for (let attempt = 1; attempt <= this.maxAttempts; attempt++) {
        try { handler(e); break; } catch { if (attempt === this.maxAttempts) this.deadLetters.push(`${name} <- ${topic} ${e.dock}`); }
      }
    }
  }
}

// ---- Subscribers: each keeps its own view of the world. ----
class AvailabilityMap { bikes = new Map<string, number>(); onStatus = (e: StatusEvent): void => { this.bikes.set(e.dock, e.bikes); }; }
class Rebalancer {
  jobs: string[] = [];
  onStatus = (e: StatusEvent): void => {
    if (e.bikes < 2) this.jobs.push(`bring bikes to ${e.dock}`);
    else if (e.bikes > e.slots - 2) this.jobs.push(`take bikes from ${e.dock}`);
  };
}
class CityFeed { sent = 0; onStatus = (): void => { this.sent++; }; } // added later; the docks' code never changed
const flakyAnalytics: Handler = () => { throw new Error("analytics store down"); }; // always fails: retries and dead letters

const list = (xs: string[]): string => "[" + xs.map((x) => `'${x}'`).join(", ") + "]";
const broker = new EventBroker();
const amap = new AvailabilityMap(), crew = new Rebalancer();
broker.subscribe("dock.status", "map", amap.onStatus);
broker.subscribe("dock.status", "rebalancer", crew.onStatus);
broker.subscribe("dock.status", "analytics", flakyAnalytics);
// The docks publish facts; they don't know who listens.
broker.publish("dock.status", { dock: "D12", bikes: 1, slots: 20 });
broker.publish("dock.status", { dock: "D40", bikes: 19, slots: 20 });
const feed = new CityFeed();
broker.subscribe("dock.status", "city-feed", feed.onStatus);
broker.publish("dock.status", { dock: "D41", bikes: 8, slots: 20 });
console.log("map: {" + [...amap.bikes].map(([k, v]) => `'${k}': ${v}`).join(", ") + "}");
console.log("crew jobs: " + list(crew.jobs));
console.log("city feed events: " + feed.sent);
console.log("dead letters: " + list(broker.deadLetters));

import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.function.Consumer;
import java.util.stream.Collectors;

record StatusEvent(String dock, int bikes, int slots) {}

// Broker topology: publishers and subscribers know only topics, never each other.
class EventBroker {
    private final Map<String, List<Map.Entry<String, Consumer<StatusEvent>>>> subs = new LinkedHashMap<>();
    final List<String> deadLetters = new ArrayList<>();
    private final int maxAttempts = 3;
    void subscribe(String topic, String name, Consumer<StatusEvent> handler) { subs.computeIfAbsent(topic, k -> new ArrayList<>()).add(Map.entry(name, handler)); }
    void publish(String topic, StatusEvent e) {
        for (Map.Entry<String, Consumer<StatusEvent>> s : subs.getOrDefault(topic, List.of())) {
            for (int attempt = 1; attempt <= maxAttempts; attempt++) {
                try { s.getValue().accept(e); break; }
                catch (RuntimeException ex) { if (attempt == maxAttempts) deadLetters.add(s.getKey() + " <- " + topic + " " + e.dock()); }
            }
        }
    }
}

// ---- Subscribers: each keeps its own view of the world. ----
class AvailabilityMap { final Map<String, Integer> bikes = new LinkedHashMap<>(); void onStatus(StatusEvent e) { bikes.put(e.dock(), e.bikes()); } }
class Rebalancer {
    final List<String> jobs = new ArrayList<>();
    void onStatus(StatusEvent e) {
        if (e.bikes() < 2) jobs.add("bring bikes to " + e.dock());
        else if (e.bikes() > e.slots() - 2) jobs.add("take bikes from " + e.dock());
    }
}
class CityFeed { int sent = 0; void onStatus(StatusEvent e) { sent++; } } // added later; the docks' code never changed

public class Main {
    static String list(List<String> xs) { return "[" + xs.stream().map(x -> "'" + x + "'").collect(Collectors.joining(", ")) + "]"; }
    public static void main(String[] args) {
        EventBroker broker = new EventBroker();
        AvailabilityMap amap = new AvailabilityMap(); Rebalancer crew = new Rebalancer();
        broker.subscribe("dock.status", "map", amap::onStatus);
        broker.subscribe("dock.status", "rebalancer", crew::onStatus);
        broker.subscribe("dock.status", "analytics", e -> { throw new IllegalStateException("analytics store down"); }); // always fails
        // The docks publish facts; they don't know who listens.
        broker.publish("dock.status", new StatusEvent("D12", 1, 20));
        broker.publish("dock.status", new StatusEvent("D40", 19, 20));
        CityFeed feed = new CityFeed();
        broker.subscribe("dock.status", "city-feed", feed::onStatus);
        broker.publish("dock.status", new StatusEvent("D41", 8, 20));
        System.out.println("map: {" + amap.bikes.entrySet().stream().map(en -> "'" + en.getKey() + "': " + en.getValue()).collect(Collectors.joining(", ")) + "}");
        System.out.println("crew jobs: " + list(crew.jobs));
        System.out.println("city feed events: " + feed.sent);
        System.out.println("dead letters: " + list(broker.deadLetters));
    }
}

Output (the same in all three languages)

map: {'D12': 1, 'D40': 19, 'D41': 8}
crew jobs: ['bring bikes to D12', 'take bikes from D40']
city feed events: 1
dead letters: ['analytics <- dock.status D12', 'analytics <- dock.status D40', 'analytics <- dock.status D41']

6. Try it

Consumer lag

400 docks each send a status every 5 seconds: 80 events a second, and three times that during the morning rush (08:00–09:00). The map's consumers each handle some events a second. Watch the backlog over a morning.

7. What it costs, and when to pick another style

8. Signs you've outgrown it

• Nobody can explain what happens after a given event without reading five codebases.
• Most events are really commands with a reply expected.
• Consumer lag is a daily alert.
• Event schemas change every release and break consumers.

9. Remember it and say it

In an interview or design review

• Name the events and who consumes them: "dock.status every 5 s, consumed by the map, rebalancing and the city feed; ride.ended by billing."
• State the guarantees: at-least-once delivery, idempotent consumers, ordering per dock (partition by dock ID).
• Size consumers against the rate: 80 events a second, 240 at the rush: how many copies, how much lag is acceptable.
• Say broker vs mediator and why: independent reactions vs an ordered process with compensation.

10. Check yourself

11. Go deeper