Event-Driven Programming

This page lives in the OOP track for a good historical reason, even though the idea outgrows any one paradigm. Alan Kay, who coined "object-oriented", later insisted the name pointed at the wrong thing: "The big idea is messaging" [1]. Objects were never meant to be bundles of getters — they were meant to be little computers that tell each other things. Event-driven programming takes that idea seriously, and this page follows it as far as it will go: from a method call, to messages between objects, to messages between machines, to a message bus running as a process in its own right.

Our running example is a shop. When an order is placed, three things must happen: stock is reserved, a receipt is emailed, and loyalty points are added.

Stage 1: Objects Calling Methods

The obvious design: OrderService knows its collaborators and calls them.

class Inventory:
    def reserve(self, order):
        print(f"Reserving stock for order {order['id']}")

class Receipts:
    def email(self, order):
        print(f"Emailing receipt for order {order['id']}")

class Loyalty:
    def add_points(self, order):
        print(f"Adding {int(order['total'])} points")

class OrderService:
    def __init__(self):
        self.inventory = Inventory()   # OrderService must know
        self.receipts = Receipts()     # every collaborator...
        self.loyalty = Loyalty()

    def place(self, order):
        self.inventory.reserve(order)   # ...call each one,
        self.receipts.email(order)      # in the right order,
        self.loyalty.add_points(order)  # and be edited for every new reaction

OrderService().place({"id": 42, "total": 99.90})

#include <iostream>

struct Order { int id; double total; };

class Inventory {
public:
    void reserve(const Order& order) {
        std::cout << "Reserving stock for order " << order.id << "\n";
    }
};

class Receipts {
public:
    void email(const Order& order) {
        std::cout << "Emailing receipt for order " << order.id << "\n";
    }
};

class Loyalty {
public:
    void addPoints(const Order& order) {
        std::cout << "Adding " << static_cast<int>(order.total) << " points\n";
    }
};

class OrderService {
    Inventory inventory;  // OrderService must know
    Receipts receipts;    // every collaborator...
    Loyalty loyalty;
public:
    void place(const Order& order) {
        inventory.reserve(order);   // ...call each one,
        receipts.email(order);      // in the right order,
        loyalty.addPoints(order);   // and be edited for every new reaction
    }
};

int main() {
    OrderService{}.place({42, 99.90});
}

record Order(int id, double total) {}

class Inventory {
    void reserve(Order order) {
        System.out.println("Reserving stock for order " + order.id());
    }
}

class Receipts {
    void email(Order order) {
        System.out.println("Emailing receipt for order " + order.id());
    }
}

class Loyalty {
    void addPoints(Order order) {
        System.out.println("Adding " + (int) order.total() + " points");
    }
}

class OrderService {
    private final Inventory inventory = new Inventory(); // OrderService must know
    private final Receipts receipts = new Receipts();    // every collaborator...
    private final Loyalty loyalty = new Loyalty();

    void place(Order order) {
        inventory.reserve(order);   // ...call each one,
        receipts.email(order);      // in the right order,
        loyalty.addPoints(order);   // and be edited for every new reaction
    }
}

// new OrderService().place(new Order(42, 99.90));

record Order(int Id, double Total);

class Inventory
{
    public void Reserve(Order order) =>
        Console.WriteLine($"Reserving stock for order {order.Id}");
}

class Receipts
{
    public void Email(Order order) =>
        Console.WriteLine($"Emailing receipt for order {order.Id}");
}

class Loyalty
{
    public void AddPoints(Order order) =>
        Console.WriteLine($"Adding {(int)order.Total} points");
}

class OrderService
{
    private readonly Inventory _inventory = new(); // OrderService must know
    private readonly Receipts _receipts = new();   // every collaborator...
    private readonly Loyalty _loyalty = new();

    public void Place(Order order)
    {
        _inventory.Reserve(order);  // ...call each one,
        _receipts.Email(order);     // in the right order,
        _loyalty.AddPoints(order);  // and be edited for every new reaction
    }
}

new OrderService().Place(new Order(42, 99.90));

class Inventory
  def reserve(order)
    puts "Reserving stock for order #{order[:id]}"
  end
end

class Receipts
  def email(order)
    puts "Emailing receipt for order #{order[:id]}"
  end
end

class Loyalty
  def add_points(order)
    puts "Adding #{order[:total].to_i} points"
  end
end

class OrderService
  def initialize
    @inventory = Inventory.new  # OrderService must know
    @receipts = Receipts.new    # every collaborator...
    @loyalty = Loyalty.new
  end

  def place(order)
    @inventory.reserve(order)   # ...call each one,
    @receipts.email(order)      # in the right order,
    @loyalty.add_points(order)  # and be edited for every new reaction
  end
end

OrderService.new.place(id: 42, total: 99.90)

This is fine — genuinely fine — for a small program. But look at what OrderService.place has quietly promised: it knows every reaction to an order, it decides their order, and it must be edited (and re-tested, and re-deployed) every time a new department wants to react. The caller and the called are welded together: same call stack, same process, same moment in time.

Stage 2: Messages Between Objects

Now invert it. Instead of calling collaborators, OrderService announces what happened — a message — and anyone interested subscribes. This is the Observer pattern (see Design Patterns) grown into a tiny in-process message bus:

class MessageBus:
    def __init__(self):
        self.subscribers = {}

    def subscribe(self, topic, handler):
        self.subscribers.setdefault(topic, []).append(handler)

    def publish(self, topic, message):
        for handler in self.subscribers.get(topic, []):
            handler(message)

bus = MessageBus()
bus.subscribe("order.placed", lambda o: print(f"Reserving stock for order {o['id']}"))
bus.subscribe("order.placed", lambda o: print(f"Emailing receipt for order {o['id']}"))
bus.subscribe("order.placed", lambda o: print(f"Adding {int(o['total'])} points"))

class OrderService:
    def __init__(self, bus):
        self.bus = bus  # the ONLY dependency

    def place(self, order):
        # ...validate, save...
        self.bus.publish("order.placed", order)  # who reacts? not our problem

OrderService(bus).place({"id": 42, "total": 99.90})

#include <functional>
#include <iostream>
#include <map>
#include <string>
#include <vector>

struct Order { int id; double total; };

class MessageBus {
    std::map<std::string, std::vector<std::function<void(const Order&)>>> subscribers;
public:
    void subscribe(const std::string& topic, std::function<void(const Order&)> handler) {
        subscribers[topic].push_back(std::move(handler));
    }

    void publish(const std::string& topic, const Order& message) {
        for (auto& handler : subscribers[topic]) handler(message);
    }
};

class OrderService {
    MessageBus& bus; // the ONLY dependency
public:
    explicit OrderService(MessageBus& bus) : bus(bus) {}

    void place(const Order& order) {
        // ...validate, save...
        bus.publish("order.placed", order); // who reacts? not our problem
    }
};

int main() {
    MessageBus bus;
    bus.subscribe("order.placed", [](const Order& o) {
        std::cout << "Reserving stock for order " << o.id << "\n"; });
    bus.subscribe("order.placed", [](const Order& o) {
        std::cout << "Emailing receipt for order " << o.id << "\n"; });
    bus.subscribe("order.placed", [](const Order& o) {
        std::cout << "Adding " << static_cast<int>(o.total) << " points\n"; });

    OrderService service(bus);
    service.place({42, 99.90});
}

import java.util.*;
import java.util.function.Consumer;

class MessageBus {
    private final Map<String, List<Consumer<Order>>> subscribers = new HashMap<>();

    void subscribe(String topic, Consumer<Order> handler) {
        subscribers.computeIfAbsent(topic, t -> new ArrayList<>()).add(handler);
    }

    void publish(String topic, Order message) {
        subscribers.getOrDefault(topic, List.of()).forEach(h -> h.accept(message));
    }
}

class OrderService {
    private final MessageBus bus; // the ONLY dependency

    OrderService(MessageBus bus) { this.bus = bus; }

    void place(Order order) {
        // ...validate, save...
        bus.publish("order.placed", order); // who reacts? not our problem
    }
}

var bus = new MessageBus();
bus.subscribe("order.placed",
    o -> System.out.println("Reserving stock for order " + o.id()));
bus.subscribe("order.placed",
    o -> System.out.println("Emailing receipt for order " + o.id()));
bus.subscribe("order.placed",
    o -> System.out.println("Adding " + (int) o.total() + " points"));

new OrderService(bus).place(new Order(42, 99.90));

using System;
using System.Collections.Generic;

class MessageBus
{
    private readonly Dictionary<string, List<Action<Order>>> _subscribers = new();

    public void Subscribe(string topic, Action<Order> handler)
    {
        if (!_subscribers.TryGetValue(topic, out var list))
            _subscribers[topic] = list = new List<Action<Order>>();
        list.Add(handler);
    }

    public void Publish(string topic, Order message)
    {
        if (_subscribers.TryGetValue(topic, out var handlers))
            handlers.ForEach(h => h(message));
    }
}

class OrderService
{
    private readonly MessageBus _bus; // the ONLY dependency
    public OrderService(MessageBus bus) => _bus = bus;

    public void Place(Order order)
    {
        // ...validate, save...
        _bus.Publish("order.placed", order); // who reacts? not our problem
    }
}

var bus = new MessageBus();
bus.Subscribe("order.placed", o => Console.WriteLine($"Reserving stock for order {o.Id}"));
bus.Subscribe("order.placed", o => Console.WriteLine($"Emailing receipt for order {o.Id}"));
bus.Subscribe("order.placed", o => Console.WriteLine($"Adding {(int)o.Total} points"));

new OrderService(bus).Place(new Order(42, 99.90));

class MessageBus
  def initialize
    @subscribers = Hash.new { |h, k| h[k] = [] }
  end

  def subscribe(topic, &handler)
    @subscribers[topic] << handler
  end

  def publish(topic, message)
    @subscribers[topic].each { |handler| handler.call(message) }
  end
end

bus = MessageBus.new
bus.subscribe("order.placed") { |o| puts "Reserving stock for order #{o[:id]}" }
bus.subscribe("order.placed") { |o| puts "Emailing receipt for order #{o[:id]}" }
bus.subscribe("order.placed") { |o| puts "Adding #{o[:total].to_i} points" }

class OrderService
  def initialize(bus)
    @bus = bus # the ONLY dependency
  end

  def place(order)
    # ...validate, save...
    @bus.publish("order.placed", order) # who reacts? not our problem
  end
end

OrderService.new(bus).place(id: 42, total: 99.90)

Three things just changed, and they are the essence of event-driven design:

The dependency arrow flipped. OrderService no longer knows Inventory, Receipts or Loyalty exist. Adding a fourth reaction (fraud checks, analytics) touches no existing code.
The message is named after the past, not the future. order.placed reports a fact; it doesn't command anyone. Subscribers decide what the fact means to them.
You paid for it with traceability. "Who handles this event?" now has no answer at the call site. Tooling, naming discipline and logging matter more from here on.

Stage 3: Crossing the Machine Boundary

The moment the loyalty team wants their own service on their own machine, a method call cannot reach it. But a message — a value, not a call — travels happily: serialise it (here as JSON, one message per line) and send it over TCP:

# loyalty_service.py — runs on machine B
import json
import socketserver

class Handler(socketserver.StreamRequestHandler):
    def handle(self):
        for line in self.rfile:
            event = json.loads(line)
            if event["topic"] == "order.placed":
                order = event["payload"]
                print(f"Adding {int(order['total'])} points for order {order['id']}")

socketserver.ThreadingTCPServer(("0.0.0.0", 5000), Handler).serve_forever()

# ---------------------------------------------------------------
# order_service.py — runs on machine A
import json
import socket

class EventSender:
    def __init__(self, host, port):
        self.sock = socket.create_connection((host, port))

    def publish(self, topic, payload):
        line = json.dumps({"topic": topic, "payload": payload}) + "\n"
        self.sock.sendall(line.encode())

sender = EventSender("loyalty.internal", 5000)
sender.publish("order.placed", {"id": 42, "total": 99.90})

// loyalty_service.cpp — runs on machine B (POSIX sockets)
#include <arpa/inet.h>
#include <sys/socket.h>
#include <unistd.h>
#include <cstdio>
#include <cstring>
#include <thread>

void handle(int client) {
    FILE* in = fdopen(client, "r");
    char line[4096];
    while (fgets(line, sizeof line, in)) {
        // parse the JSON with your library of choice (nlohmann::json, etc.)
        if (strstr(line, "\"order.placed\"")) {
            printf("Adding points: %s", line);
        }
    }
    fclose(in);
}

int main() {
    int server = socket(AF_INET, SOCK_STREAM, 0);
    sockaddr_in addr{};
    addr.sin_family = AF_INET;
    addr.sin_port = htons(5000);
    addr.sin_addr.s_addr = INADDR_ANY;
    bind(server, reinterpret_cast<sockaddr*>(&addr), sizeof addr);
    listen(server, 16);
    while (true) {
        std::thread(handle, accept(server, nullptr, nullptr)).detach();
    }
}

// order_service.cpp — runs on machine A
// int sock = socket(AF_INET, SOCK_STREAM, 0);
// connect(sock, /* loyalty.internal:5000 */);
// const char* msg =
//     "{\"topic\": \"order.placed\", \"payload\": {\"id\": 42, \"total\": 99.9}}\n";
// send(sock, msg, strlen(msg), 0);

// LoyaltyService.java — runs on machine B
import java.io.*;
import java.net.*;

public class LoyaltyService {
    public static void main(String[] args) throws IOException {
        try (var server = new ServerSocket(5000)) {
            while (true) {
                Socket client = server.accept();
                new Thread(() -> handle(client)).start();
            }
        }
    }

    static void handle(Socket client) {
        try (var in = new BufferedReader(new InputStreamReader(client.getInputStream()))) {
            String line;
            while ((line = in.readLine()) != null) {
                Event event = Json.parse(line, Event.class); // Jackson/Gson etc.
                if (event.topic().equals("order.placed")) {
                    System.out.println("Adding " + (int) event.payload().total()
                                       + " points for order " + event.payload().id());
                }
            }
        } catch (IOException ignored) {}
    }
}

record Event(String topic, Order payload) {}

// OrderService.java — runs on machine A
// var sock = new Socket("loyalty.internal", 5000);
// var out = new PrintWriter(sock.getOutputStream(), true);
// out.println(Json.stringify(new Event("order.placed", new Order(42, 99.90))));

// LoyaltyService — runs on machine B
using System.Net;
using System.Net.Sockets;
using System.Text.Json;

var listener = new TcpListener(IPAddress.Any, 5000);
listener.Start();
while (true)
{
    var client = await listener.AcceptTcpClientAsync();
    _ = Task.Run(async () =>
    {
        using var reader = new StreamReader(client.GetStream());
        while (await reader.ReadLineAsync() is { } line)
        {
            var evt = JsonSerializer.Deserialize<Event>(line)!;
            if (evt.Topic == "order.placed")
                Console.WriteLine(
                    $"Adding {(int)evt.Payload.Total} points for order {evt.Payload.Id}");
        }
    });
}

record Event(string Topic, Order Payload);
record Order(int Id, double Total);

// OrderService — runs on machine A
// using var sock = new TcpClient("loyalty.internal", 5000);
// var writer = new StreamWriter(sock.GetStream()) { AutoFlush = true };
// await writer.WriteLineAsync(JsonSerializer.Serialize(
//     new Event("order.placed", new Order(42, 99.90))));

# loyalty_service.rb — runs on machine B
require "socket"
require "json"

server = TCPServer.new("0.0.0.0", 5000)
loop do
  Thread.start(server.accept) do |client|
    client.each_line do |line|
      event = JSON.parse(line, symbolize_names: true)
      next unless event[:topic] == "order.placed"

      order = event[:payload]
      puts "Adding #{order[:total].to_i} points for order #{order[:id]}"
    end
  end
end

# ---------------------------------------------------------------
# order_service.rb — runs on machine A
require "socket"
require "json"

class EventSender
  def initialize(host, port)
    @sock = TCPSocket.new(host, port)
  end

  def publish(topic, payload)
    @sock.puts JSON.generate(topic: topic, payload: payload)
  end
end

sender = EventSender.new("loyalty.internal", 5000)
sender.publish("order.placed", id: 42, total: 99.90)

Notice what stayed the same: the event is still order.placed with the same payload. What changed is everything around it — the network can fail, the receiver can be down, delivery takes real time, and the sender needs the receiver's address. That last one is the new coupling: point-to-point connections mean every producer must know every consumer's host and port, and sending to three services means three connections and three chances to fail differently.

Stage 4: A Message Bus of Its Own

The fix is the same trick as Stage 2, applied to infrastructure: introduce a middleman so that producers and consumers only know one address — the bus. Here is a complete (if minimal) broker: a separate process speaking a tiny text protocol — SUB topic to subscribe, PUB topic payload to publish, and the broker fans out MSG topic payload lines to every subscriber:

# broker.py — its own process; everyone else just knows its address
import socket
import threading

subscribers = {}   # topic -> list of client sockets
lock = threading.Lock()

def serve(client):
    for line in client.makefile("r"):
        command, _, rest = line.strip().partition(" ")
        if command == "SUB":
            with lock:
                subscribers.setdefault(rest, []).append(client)
        elif command == "PUB":
            topic, _, payload = rest.partition(" ")
            with lock:
                targets = list(subscribers.get(topic, []))
            for sock in targets:
                try:
                    sock.sendall(f"MSG {topic} {payload}\n".encode())
                except OSError:
                    with lock:
                        subscribers[topic].remove(sock)

server = socket.create_server(("0.0.0.0", 7000))
while True:
    client, _ = server.accept()
    threading.Thread(target=serve, args=(client,), daemon=True).start()

# --- any service, on any machine, in any language: ---------------
# bus = socket.create_connection(("bus.internal", 7000))
# bus.sendall(b"SUB order.placed\n")                    # consumer side
# bus.sendall(b'PUB order.placed {"id": 42}\n')         # producer side
# for line in bus.makefile("r"):                         # consumer loop
#     print("received:", line.strip())

// broker.cpp — its own process; everyone else just knows its address
#include <arpa/inet.h>
#include <sys/socket.h>
#include <unistd.h>
#include <cstdio>
#include <cstring>
#include <map>
#include <mutex>
#include <string>
#include <thread>
#include <vector>

std::map<std::string, std::vector<int>> subscribers; // topic -> sockets
std::mutex lock_;

void serve(int client) {
    FILE* in = fdopen(client, "r");
    char line[4096];
    while (fgets(line, sizeof line, in)) {
        std::string s(line);
        while (!s.empty() && (s.back() == '\n' || s.back() == '\r')) s.pop_back();
        auto space = s.find(' ');
        std::string command = s.substr(0, space), rest = s.substr(space + 1);
        if (command == "SUB") {
            std::lock_guard<std::mutex> g(lock_);
            subscribers[rest].push_back(client);
        } else if (command == "PUB") {
            auto sp = rest.find(' ');
            std::string topic = rest.substr(0, sp), payload = rest.substr(sp + 1);
            std::vector<int> targets;
            {
                std::lock_guard<std::mutex> g(lock_);
                targets = subscribers[topic];
            }
            std::string msg = "MSG " + topic + " " + payload + "\n";
            for (int sock : targets) send(sock, msg.data(), msg.size(), 0);
        }
    }
    fclose(in);
}

int main() {
    int server = socket(AF_INET, SOCK_STREAM, 0);
    sockaddr_in addr{};
    addr.sin_family = AF_INET;
    addr.sin_port = htons(7000);
    addr.sin_addr.s_addr = INADDR_ANY;
    bind(server, reinterpret_cast<sockaddr*>(&addr), sizeof addr);
    listen(server, 64);
    while (true) {
        std::thread(serve, accept(server, nullptr, nullptr)).detach();
    }
}

// any service, on any machine, in any language:
//   send "SUB order.placed\n"           consumer side
//   send "PUB order.placed {...}\n"     producer side
//   read "MSG order.placed {...}\n"     consumer loop

// Broker.java — its own process; everyone else just knows its address
import java.io.*;
import java.net.*;
import java.util.*;

public class Broker {
    private static final Map<String, List<PrintWriter>> subscribers = new HashMap<>();

    public static void main(String[] args) throws IOException {
        try (var server = new ServerSocket(7000)) {
            while (true) {
                Socket client = server.accept();
                new Thread(() -> serve(client)).start();
            }
        }
    }

    static void serve(Socket client) {
        try (var in = new BufferedReader(new InputStreamReader(client.getInputStream()))) {
            var out = new PrintWriter(client.getOutputStream(), true);
            String line;
            while ((line = in.readLine()) != null) {
                String[] parts = line.split(" ", 2);
                switch (parts[0]) {
                    case "SUB" -> {
                        synchronized (subscribers) {
                            subscribers.computeIfAbsent(parts[1], t -> new ArrayList<>())
                                       .add(out);
                        }
                    }
                    case "PUB" -> {
                        String[] msg = parts[1].split(" ", 2); // topic, payload
                        List<PrintWriter> targets;
                        synchronized (subscribers) {
                            targets = new ArrayList<>(
                                subscribers.getOrDefault(msg[0], List.of()));
                        }
                        targets.forEach(w -> w.println("MSG " + msg[0] + " " + msg[1]));
                    }
                }
            }
        } catch (IOException ignored) {}
    }
}

// any service, on any machine, in any language:
//   out.println("SUB order.placed");                 consumer side
//   out.println("PUB order.placed {\"id\": 42}");    producer side
//   in.readLine()  ->  "MSG order.placed {...}"      consumer loop

// Broker — its own process; everyone else just knows its address
using System.Net;
using System.Net.Sockets;

var subscribers = new Dictionary<string, List<StreamWriter>>();
var gate = new object();

var listener = new TcpListener(IPAddress.Any, 7000);
listener.Start();
while (true)
{
    var client = await listener.AcceptTcpClientAsync();
    _ = Task.Run(async () =>
    {
        var stream = client.GetStream();
        using var reader = new StreamReader(stream);
        var writer = new StreamWriter(stream) { AutoFlush = true };
        while (await reader.ReadLineAsync() is { } line)
        {
            var parts = line.Split(' ', 2);
            if (parts[0] == "SUB")
            {
                lock (gate)
                {
                    if (!subscribers.TryGetValue(parts[1], out var list))
                        subscribers[parts[1]] = list = new List<StreamWriter>();
                    list.Add(writer);
                }
            }
            else if (parts[0] == "PUB")
            {
                var msg = parts[1].Split(' ', 2); // topic, payload
                List<StreamWriter> targets;
                lock (gate)
                    targets = subscribers.TryGetValue(msg[0], out var l)
                        ? new List<StreamWriter>(l) : new();
                foreach (var w in targets)
                    try { await w.WriteLineAsync($"MSG {msg[0]} {msg[1]}"); }
                    catch { /* subscriber gone */ }
            }
        }
    });
}

// any service, on any machine, in any language:
//   writer.WriteLine("SUB order.placed");               consumer side
//   writer.WriteLine("PUB order.placed {\"id\": 42}");  producer side
//   reader.ReadLine()  ->  "MSG order.placed {...}"     consumer loop

# broker.rb — its own process; everyone else just knows its address
require "socket"

subscribers = Hash.new { |h, k| h[k] = [] }
lock = Mutex.new

server = TCPServer.new("0.0.0.0", 7000)
loop do
  Thread.start(server.accept) do |client|
    client.each_line do |line|
      command, rest = line.strip.split(" ", 2)
      case command
      when "SUB"
        lock.synchronize { subscribers[rest] << client }
      when "PUB"
        topic, payload = rest.split(" ", 2)
        targets = lock.synchronize { subscribers[topic].dup }
        targets.each do |sock|
          sock.puts "MSG #{topic} #{payload}"
        rescue IOError, Errno::EPIPE
          lock.synchronize { subscribers[topic].delete(sock) }
        end
      end
    end
  end
end

# --- any service, on any machine, in any language: ---------------
# bus = TCPSocket.new("bus.internal", 7000)
# bus.puts "SUB order.placed"                    # consumer side
# bus.puts 'PUB order.placed {"id": 42}'         # producer side
# bus.each_line { |l| puts "received: #{l}" }    # consumer loop

Our toy broker fits on a page, which is exactly why you should not ship it: it loses messages when subscribers are offline, has no acknowledgements, no persistence, no replay, no security. Production systems reach for a hardened broker instead — Redis pub/sub or NATS at the lightweight end, RabbitMQ or MQTT for routing and IoT, Kafka when events must be stored and replayed as a log [2]. The programming model, though, is the one you just built: publishers name facts, the bus fans them out, subscribers react.

The Trade-Off Ledger

You gain	You pay with
Loose coupling — producers and consumers evolve, deploy and scale independently	Harder end-to-end reasoning: no stack trace crosses the bus
Extensibility — new subscribers, zero changes to publishers	Hidden flows: the system's behaviour is the sum of subscriptions
Resilience — one slow consumer no longer blocks the others	Eventual consistency: the points arrive after the order, not with it
A natural audit trail — events are facts worth logging	Operational surface: the bus itself must be run, monitored and secured

A good rule of thumb: start at Stage 1, and let real forces — team boundaries, deployment independence, genuinely asynchronous reactions — push you rightward. Fowler's advice applies: "event-driven" covers several distinct patterns (notification, state transfer, event sourcing), and it pays to say which one you mean [3].

References

Kay, A. (1998). "prototypes vs classes was: Re: Sun's HotSpot." Email to the Squeak mailing list, 10 October 1998 (“The big idea is messaging”). http://lists.squeakfoundation.org/pipermail/squeak-dev/1998-October/017019.html
Hohpe, G. & Woolf, B. (2003). Enterprise Integration Patterns: Designing, Building, and Deploying Messaging Solutions. Addison-Wesley. https://www.enterpriseintegrationpatterns.com/
Fowler, M. (2017). "What do you mean by 'Event-Driven'?" https://martinfowler.com/articles/201701-event-driven.html