Skip to content

Mediator

Intent

Move the interactions between a group of objects into one object they all talk to, so that none of them holds a reference to another. The colleagues stay small and reusable, the rules of the conversation live in one place you can read and test, and adding a colleague changes the mediator rather than every peer.

When to use and when not to

Use it when

  • Objects that should be simple are tangled: a car that knows every hall panel, a panel that knows every car. N colleagues talking to each other is N x (N - 1) connections; through a mediator it is N.
  • The rules of the interaction are a policy you expect to change: which car answers a call, who may speak to whom, what happens when two bidders bid at once.
  • You want to test the coordination without the colleagues: a controller fed by fake panels and cars that record what they were told.

Leave it out when

  • Two objects talk. A method call is the mediator.
  • The subject does not care who listens and nothing decides; that is Observer, or an event bus once topics are involved.
  • The mediator would only forward calls. A mediator with no policy is an indirection tax; the colleagues may as well call each other.
  • The coordination is a long-running workflow across services with persistent state; that is a saga or an orchestrator, not an in-process object.

Structure

Two kinds of colleague, the mediator interface they depend on, and one concrete controller that holds every decision and the state of the conversation.

classDiagram
    class DispatchMediator {
        <<interface>>
        +hall_call(floor: int, direction: Direction) None
        +car_arrived(car: Elevator, floor: int) None
    }
    class ElevatorController {
        -_cars: List~Elevator~
        -_panels: Dict~int,HallPanel~
        -_pending: Dict~tuple,Elevator~
        +log: List~str~
        +panel(floor: int) HallPanel
        +hall_call(floor: int, direction: Direction) None
        +car_arrived(car: Elevator, floor: int) None
        +tick() None
    }
    class HallPanel {
        +floor: int
        +lit: Set~Direction~
        +press(direction: Direction) None
        +clear(direction: Direction) None
    }
    class Elevator {
        +name: str
        +floor: int
        +direction: Direction
        +stops: FrozenSet~int~
        +attach(mediator: DispatchMediator) None
        +add_stop(floor: int) None
        +step() None
    }
    class Direction {
        <<enumeration>>
        UP
        DOWN
    }
    DispatchMediator <|.. ElevatorController
    HallPanel --> DispatchMediator : reports presses
    Elevator --> DispatchMediator : reports arrivals
    ElevatorController o-- "many" HallPanel : lights and clears
    ElevatorController o-- "many" Elevator : assigns stops
    Elevator ..> Direction

No arrow runs between HallPanel and Elevator; that missing edge is the pattern. Both depend on the two-method DispatchMediator interface, so a test can stand in a recorder for the controller. ElevatorController is the only class with more than one kind of collaborator.

Canonical example in Python

The colleagues come first (code/patterns/mediator.py, tested by code/patterns/tests/test_mediator.py):

code/patterns/mediator.py — the mediator interface and the two colleagues
class DispatchMediator(Protocol):
    """What a colleague may ask of the mediator. Colleagues never see each other, only this."""

    def hall_call(self, floor: int, direction: Direction) -> None: ...

    def car_arrived(self, car: Elevator, floor: int) -> None: ...


class HallPanel:
    """A colleague: the two buttons and lamps on one floor. It reports presses and lights lamps."""

    def __init__(self, floor: int, mediator: DispatchMediator) -> None:
        self.floor = floor
        self._mediator = mediator
        self.lit: set[Direction] = set()

    def press(self, direction: Direction) -> None:
        if direction in self.lit:
            return  # the call is already registered; pressing harder changes nothing
        self.lit.add(direction)
        self._mediator.hall_call(self.floor, direction)

    def clear(self, direction: Direction) -> None:
        self.lit.discard(direction)


class Elevator:
    """A colleague: moves one floor per tick towards its stops and reports every arrival.

    Cabin buttons call ``add_stop`` directly, because a car owns its own stops; hall
    calls reach it only through the mediator. The car keeps its direction while it
    has stops ahead (the LOOK rule) and otherwise turns around or goes idle.
    """

    def __init__(self, name: str, floor: int = 0) -> None:
        self.name = name
        self.floor = floor
        self.direction: Direction | None = None
        self._stops: set[int] = set()
        self._mediator: DispatchMediator | None = None

    def attach(self, mediator: DispatchMediator) -> None:
        self._mediator = mediator

    @property
    def stops(self) -> frozenset[int]:
        return frozenset(self._stops)

    @property
    def idle(self) -> bool:
        return not self._stops

    def add_stop(self, floor: int) -> None:
        self._stops.add(floor)

    def step(self) -> None:
        """One tick: open at the current floor if it is a stop, otherwise move one floor."""
        if not self._stops:
            self.direction = None
            return
        if self.floor in self._stops:
            self._arrive()
            return
        if self.direction is Direction.UP and not any(s > self.floor for s in self._stops):
            self.direction = Direction.DOWN
        elif self.direction is Direction.DOWN and not any(s < self.floor for s in self._stops):
            self.direction = Direction.UP
        elif self.direction is None:
            nearest = min(self._stops, key=lambda s: abs(s - self.floor))
            self.direction = Direction.UP if nearest > self.floor else Direction.DOWN
        self.floor += 1 if self.direction is Direction.UP else -1
        if self.floor in self._stops:
            self._arrive()

    def _arrive(self) -> None:
        self._stops.remove(self.floor)
        if self._mediator is not None:
            self._mediator.car_arrived(self, self.floor)

Three decisions to say out loud:

  • Colleagues depend on a Protocol, not on the controller. A panel knows how to light a lamp and whom to tell about a press; a car knows how to move and whom to tell about an arrival. Cabin buttons call add_stop directly, because a car's own stops are not an interaction between colleagues. The mediator is for what happens between objects, not for everything.
  • Per-colleague behaviour stays in the colleague. The LOOK rule (keep going while stops remain ahead, then turn) is the car's business. Pulling it into the controller is the first step towards the god object in the warning below.
  • Pressing twice is absorbed at the edge. The panel ignores a press for a lamp that is already lit, so the mediator never sees duplicate calls and never has to deduplicate them.

The mediator owns the two decisions that would otherwise be smeared across every colleague:

code/patterns/mediator.py — the controller
class ElevatorController:
    """The Mediator: the only object that knows both the panels and the cars.

    It owns the two decisions that would otherwise be smeared across every
    colleague: which car answers a call, and which lamps an arrival clears.
    ``_pending`` remembers the assignment so an arrival of a *different* car at the
    same floor does not clear a lamp that car is not serving.
    """

    def __init__(self, cars: Iterable[Elevator], floors: int) -> None:
        self._cars = list(cars)
        if not self._cars or floors < 2:
            raise ValidationError("a controller needs at least one car and two floors")
        for car in self._cars:
            car.attach(self)
        self._panels = {floor: HallPanel(floor, self) for floor in range(floors)}
        self._pending: dict[tuple[int, Direction], Elevator] = {}
        self.log: list[str] = []

    def panel(self, floor: int) -> HallPanel:
        if floor not in self._panels:
            raise NotFoundError(f"there is no floor {floor}")
        return self._panels[floor]

    @property
    def cars(self) -> tuple[Elevator, ...]:
        return tuple(self._cars)

    def hall_call(self, floor: int, direction: Direction) -> None:
        car = min(self._cars, key=lambda candidate: self._cost(candidate, floor, direction))
        car.add_stop(floor)
        self._pending[(floor, direction)] = car
        self.log.append(f"call {floor} {direction}: assigned to {car.name}")

    def car_arrived(self, car: Elevator, floor: int) -> None:
        served = [key for key, assigned in self._pending.items() if assigned is car and key[0] == floor]
        for key in served:
            del self._pending[key]
            self._panels[floor].clear(key[1])
        cleared = ", ".join(f"{key[1]}" for key in served) or "nothing"
        self.log.append(f"{car.name} arrived at {floor}: cleared {cleared}")

    def tick(self) -> None:
        for car in self._cars:
            car.step()

    @staticmethod
    def _cost(car: Elevator, floor: int, direction: Direction) -> tuple[int, int, int]:
        """Nearest car that is idle or already heading that way wins; otherwise the least busy."""
        distance = abs(car.floor - floor)
        heading = car.direction is direction and (
            (direction is Direction.UP and car.floor <= floor)
            or (direction is Direction.DOWN and car.floor >= floor)
        )
        if car.idle or heading:
            return (0, distance, 0)
        return (1, len(car.stops), distance)

_cost is the dispatch policy in one place: nearest car that is idle or already heading that way, else the least busy. Swap the method and every colleague is untouched, which is where a Strategy plugs in. _pending is the state of the conversation: which car serves which call. Without it, any car stopping at floor 3 would clear a lamp it was not serving. State about the relationship belongs in the mediator; state about the car stays in the car.

A hall call and its answer: the panel and the car never address each other.

sequenceDiagram
    autonumber
    actor P as Passenger
    participant HP as Hall panel on 3
    participant C as ElevatorController
    participant A as Elevator A
    participant B as Elevator B
    P->>HP: press(UP)
    HP->>C: hall_call(3, UP)
    Note over C: cost: A idle at 0 is 3 away, B idle at 8 is 5 away
    C->>A: add_stop(3)
    loop one tick per floor
        C->>A: step()
    end
    A->>C: car_arrived(A, 3)
    C->>HP: clear(UP)
    Note over B: B never heard about the call

Running python -m patterns.mediator prints:

--- two cars, ten floors; panels and cars only ever talk to the controller ---
start:  A@0  B@8                                 lamps: none
calls:  A@0 [3]  B@8 [7]                         lamps: 3up, 7down
tick 1: A@1 up [3]  B@7 down                     lamps: 3up
tick 2: A@2 up [3]  B@7                          lamps: 3up
tick 3: A@3 up [6]  B@7                          lamps: none
tick 4: A@4 up [6]  B@7                          lamps: none
controller log:
  call 3 up: assigned to A
  call 7 down: assigned to B
  B arrived at 7: cleared down
  A arrived at 3: cleared up
--- the chat room: the same shape with callables and a routing policy ---
alice broadcasts -> delivered to 2
bob broadcasts   -> delivered to 1 (carol blocked bob)
bob to alice     -> delivered to 1
  alice  ['bob: hi', 'bob: psst']
  bob    ['alice: hello all']
  carol  ['alice: hello all']
rejected: 'dave' is not in the room

Pythonic variant

When colleagues only need to receive, they do not need a class or a set_mediator call: a colleague is a callable, and the mediator is a dict plus a policy. The chat room is the textbook example in that form:

code/patterns/mediator.py — a chat room over plain callables
# (sender, text) -> None: a member is nothing more than something that can receive.
type Receiver = Callable[[str, str], None]


class ChatRoom:
    """The lightweight mediator: members are callables, the room owns the routing policy.

    Members never hold references to each other; broadcast, direct messages and
    blocking are decided here, in one place. That policy is what tells a mediator
    apart from an event bus, which delivers by topic and decides nothing.
    """

    def __init__(self) -> None:
        self._members: dict[str, Receiver] = {}
        self._blocked: set[tuple[str, str]] = set()  # (who blocked, whom)

    def join(self, name: str, receive: Receiver) -> None:
        if name in self._members:
            raise ConflictError(f"{name!r} is already in the room")
        self._members[name] = receive

    def leave(self, name: str) -> None:
        self._members.pop(name, None)

    def block(self, blocker: str, blocked: str) -> None:
        self._blocked.add((blocker, blocked))

    def say(self, sender: str, text: str, to: str | None = None) -> int:
        """Broadcast, or a direct message when ``to`` is given. Returns how many members received it."""
        if sender not in self._members:
            raise NotFoundError(f"{sender!r} is not in the room")
        if to is not None and to not in self._members:
            raise NotFoundError(f"{to!r} is not in the room")
        recipients = [to] if to is not None else [name for name in self._members if name != sender]
        delivered = 0
        for name in recipients:
            if (name, sender) in self._blocked:
                continue
            self._members[name](sender, text)
            delivered += 1
        return delivered
  • Members are callables. A lambda appending to a list, a bound method on an Inbox, a function that posts to a socket: the room cannot tell them apart and never holds one member's reference to another.
  • The policy is the point. Blocking, direct messages and excluding the sender from a broadcast live in say. Take the policy away and ChatRoom degrades into an event bus, which is the right tool when there is no policy.
  • Return values close the loop. say reports how many members received the message, so the caller can tell a blocked delivery from a silent one without knowing the rule.
Reach for When
A direct method call Two objects, one direction
Observer One subject, many listeners, no decision about who hears what
An event bus Many publishers and subscribers decoupled by topic, still no decision
A mediator over callables Many-to-many with a routing policy, colleagues that only receive
A mediator with a Protocol Colleagues that both report and are commanded, state about the relationship, a policy you will swap

Real-world usage

  • The asyncio event loop: coroutines never schedule each other. They hand control to the loop, which decides who runs next, wires futures to callbacks and owns timers and I/O readiness. loop.call_soon and Future.add_done_callback are colleagues talking to the mediator.
  • GUI toolkits: tkinter's mainloop and Qt's event dispatch route events to widgets, and a dialog controller that enables OK only when every field validates is the Gang of Four's own example.
  • Django Form.clean(): fields validate themselves; cross-field rules (end date after start date) live in the form, the mediator between fields.
  • Board games: a Board that validates a chess move against every other piece is a mediator; pieces never consult each other.
Looks like Mediator How to tell them apart
Observer A subject emits and observers react; nobody decides who should hear what. A mediator receives, decides and directs. Mediators often use Observer to hear from colleagues, which is why the two get conflated.
Event Bus Delivery by topic with no policy; publishers and subscribers are unknown to each other. A bus that grows routing rules is a mediator in disguise, and should be named as one.
Facade A facade simplifies a subsystem for outsiders, and the subsystem does not know it exists. A mediator coordinates peers that know it and report to it.
Command The hall call is a request object; the controller is who receives it and decides. Commands are the messages, the mediator is the router.
Strategy The dispatch rule inside the controller is a strategy; the controller is the mediator that applies it.
Singleton Controllers are often made singletons out of habit. One instance created at the composition root is enough.

Where it appears in LLD problems

  • Design an elevator system — ElevatorController between hall calls, cars and displays, with the dispatch strategy plugged into it and per-car stop sets kept in the cars.
  • Design an online auction — the bid service between bidders: validate against the current high bid, resolve proxy bids, notify the outbid bidder. Bidders never see each other.
  • Design Uber (LLD) with driver matching — the matching service between riders, drivers and the location index, owning the offer, accept and timeout cascade.
  • Design a traffic signal controller — the intersection controller that makes conflicting greens impossible; lights never consult each other.

Interview tips

Interview tip

Draw the star, not the mesh: every colleague points at one box. Then say what the box decides (the dispatch rule, the lamp rule) and what state it keeps (pending calls), because a mediator with no decision is plumbing. Close with how you would test it: fake colleagues that record what they were told, driven by a tick.

Common mistake

The god object. A mediator that absorbs the colleagues' own behaviour (moving the car, lighting the lamp, validating a field) until nobody can read it. Only the interaction moves into the mediator; per-colleague behaviour stays where it was. Runner-up: calling it Observer. If the object decides who hears what, it is a mediator, and hiding that policy in subscriber order is a bug waiting to happen.