Mediator Pattern: A Complete Guide with Examples in 11 Programming Languages

The Mediator Pattern is a behavioral design pattern that reduces chaotic dependencies between objects by introducing a central mediator object that handles all communication between them. Instead of components talking directly to each other — creating a web of tight couplings that grows quadratically with every new component — each component knows only the mediator. The mediator knows everyone and routes messages, coordinates workflows, and enforces interaction rules. The result is a system where components are loosely coupled, independently testable, and easy to extend without touching existing code.

In this comprehensive guide, we'll explore the Mediator Pattern with real-world examples in Python, TypeScript, Java, JavaScript, C#, PHP, Go, Rust, Dart, Swift, and Kotlin — complete with language-specific best practices, common pitfalls, and when to use this powerful pattern.

Table of Contents

• What is the Mediator Pattern?
• Why Use the Mediator Pattern?
• Mediator Pattern Comparison
• Mediator Pattern Explained
• Class Diagram
• Beginner-Friendly Example
• Production-Ready Example
• Real-World Use Cases
• Language-Specific Mistakes and Anti-Patterns
• Frequently Asked Questions
• Key Takeaways

What is the Mediator Pattern?

The Mediator Pattern replaces a tangle of direct peer-to-peer references with a single hub. Components (called Colleagues) send messages to the mediator rather than to each other. The mediator receives the message, decides what to do with it, and notifies whichever other components need to react.

Think of it like air traffic control. Planes don't talk directly to each other to coordinate landing and takeoff — that would be chaos with hundreds of aircraft. Instead, every plane talks to the control tower (the mediator). The tower knows the state of every runway, every plane's position, and all the coordination rules. It routes messages and issues instructions. Planes are simple; the tower is where the complexity lives.

Why Use the Mediator Pattern?

The Mediator Pattern offers several compelling benefits:

1. Reduced Coupling: Components don't know about each other — they only know the mediator interface. Adding or removing a component never requires changes to other components.
2. Centralized Control Logic: All interaction rules and workflow coordination live in one place — the mediator. This makes the rules easy to find, read, and change.
3. Simplified Components: Each colleague handles only its own concerns. It doesn't need to know how other components work, what their state is, or when to notify them.
4. Easier Reuse: A component with no direct dependencies on other components is trivially reusable in a different context with a different mediator.
5. Single Point of Change: When interaction rules change, only the mediator changes — not every component that participates in the interaction.
6. Testability: Components can be tested with a mock mediator. The mediator can be tested by injecting mock components.
7. Extensibility: New components are added by registering them with the mediator. Existing components are untouched.

Mediator Pattern Comparison

Let's compare the Mediator Pattern with related patterns:

Key Distinctions:

• Mediator vs. Observer: Both decouple senders from receivers. Observer is broadcast — a publisher fires an event and all interested subscribers receive it, but the publisher doesn't control who does what. Mediator is centralized coordination — the mediator decides what happens when a component notifies it, often calling specific methods on specific components based on business logic. Use Observer for loose event notification; use Mediator for complex multi-component workflow coordination.
• Mediator vs. Facade: Both introduce an intermediary. Facade simplifies a one-way interface to a subsystem (client → facade → subsystem). Mediator enables bidirectional coordination between peers (component ↔ mediator ↔ component). A facade is used by external clients; a mediator is used by the components themselves.

Mediator Pattern Explained

The Mediator Pattern typically involves:

Core Components:

1. Mediator Interface: Declares the communication interface that colleagues use to send messages — typically a notify(sender, event) method or typed message-sending methods.
2. Concrete Mediator: Implements the Mediator interface. Holds references to all colleague objects. Contains all the coordination and routing logic. This is where the "business rules of interaction" live.
3. Colleague Interface (optional): A base class or interface for components, declaring a reference to the mediator and common behavior.
4. Concrete Colleagues: Individual components that do their own work and communicate exclusively through the mediator. Each knows the mediator; none knows the others directly.

Communication Flow:

Component A                Mediator               Component B
    |                          |                       |
    |── notify("event_A") ────>|                       |
    |                          |── action_on_b() ─────>|
    |                          |                       |
    |                          |<── notify("event_B") ─|
    |<── action_on_a() ────────|                       |

Mediator Variants:

• Event-based Mediator: Components fire named events; mediator routes them to handlers (closest to Observer)
• Method-based Mediator: Components call typed methods on the mediator interface (most type-safe)
• Message-passing Mediator: Components post message objects; mediator dispatches based on message type
• Workflow Mediator: Mediator orchestrates a multi-step process, calling components in sequence
• Event Bus / Message Bus: A decentralized mediator where publishers post to topics and subscribers register interest — no single mediator object but the bus plays the same role

Class Diagram

Here's the UML class diagram showing the Mediator structure:

Beginner-Friendly Example

Let's start with the most intuitive Mediator example: a chat room. Users join a room and send messages. No user knows the others directly — they all communicate through the chat room (the mediator), which routes messages and enforces rules like blocking banned users.

from __future__ import annotations
from abc import ABC, abstractmethod
from datetime import datetime
from typing import Dict, List, Optional


# ── Mediator Interface ────────────────────────────────────────────
class ChatMediator(ABC):
    @abstractmethod
    def send_message(self, message: str, sender: "User") -> None:
        pass

    @abstractmethod
    def add_user(self, user: "User") -> None:
        pass

    @abstractmethod
    def remove_user(self, user: "User") -> None:
        pass


# ── Colleague ─────────────────────────────────────────────────────
class User:
    def __init__(self, name: str, mediator: ChatMediator) -> None:
        self.name     = name
        self._mediator = mediator
        self._mediator.add_user(self)

    def send(self, message: str) -> None:
        print(f"[{self.name}] → sending: \"{message}\"")
        self._mediator.send_message(message, self)

    def receive(self, message: str, sender: "User") -> None:
        ts = datetime.now().strftime("%H:%M:%S")
        print(f"  [{ts}] {self.name} received from {sender.name}: \"{message}\"")

    def leave(self) -> None:
        print(f"[{self.name}] leaving the room")
        self._mediator.remove_user(self)

    def __repr__(self) -> str:
        return f"User({self.name})"


# ── Concrete Mediator ─────────────────────────────────────────────
class ChatRoom(ChatMediator):
    def __init__(self, name: str) -> None:
        self.name       = name
        self._users:    Dict[str, User] = {}
        self._banned:   List[str]       = []
        self._log:      List[str]       = []

    def add_user(self, user: User) -> None:
        if user.name in self._banned:
            print(f"[ChatRoom:{self.name}] '{user.name}' is banned and cannot join")
            return
        self._users[user.name] = user
        self._log_event(f"{user.name} joined")
        self._broadcast_system(f"👋 {user.name} has joined the room", exclude=user)

    def remove_user(self, user: User) -> None:
        self._users.pop(user.name, None)
        self._log_event(f"{user.name} left")
        self._broadcast_system(f"👋 {user.name} has left the room")

    def ban_user(self, name: str) -> None:
        self._banned.append(name)
        if name in self._users:
            self._broadcast_system(f"🚫 {name} has been banned")
            self._users.pop(name)
        self._log_event(f"{name} banned")

    def send_message(self, message: str, sender: User) -> None:
        if sender.name not in self._users:
            print(f"  [ChatRoom] {sender.name} is not in the room — message rejected")
            return
        entry = f"{sender.name}: {message}"
        self._log.append(entry)
        # Mediator routes message to all OTHER users
        for user in self._users.values():
            if user is not sender:
                user.receive(message, sender)

    def _broadcast_system(self, notice: str, exclude: Optional[User] = None) -> None:
        for user in self._users.values():
            if user is not exclude:
                print(f"  [System → {user.name}] {notice}")

    def _log_event(self, event: str) -> None:
        ts = datetime.now().strftime("%H:%M:%S")
        self._log.append(f"[{ts}] {event}")

    def print_log(self) -> None:
        print(f"\n[ChatRoom:{self.name}] Event log:")
        for entry in self._log:
            print(f"  {entry}")

    def roster(self) -> None:
        print(f"[ChatRoom:{self.name}] Users online: {list(self._users.keys())}")


# ── Client ────────────────────────────────────────────────────────
if __name__ == "__main__":
    room = ChatRoom("General")

    alice = User("Alice", room)
    bob   = User("Bob",   room)
    carol = User("Carol", room)

    print()
    room.roster()

    print("\n--- Messages ---")
    alice.send("Hey everyone!")
    bob.send("Hi Alice!")
    carol.send("Hello all 👋")

    print("\n--- Bob leaves ---")
    bob.leave()
    alice.send("Anyone still here?")

    print("\n--- Carol is banned ---")
    room.ban_user("Carol")
    alice.send("Just me now?")

    room.print_log()

Production-Ready Example

Now let's tackle a realistic scenario: an air traffic control system. Multiple aircraft share the same airspace. No aircraft communicates with another directly — all coordination (landing permission, runway assignment, conflict alerts, position broadcasting) goes through the ATC tower (the mediator). This demonstrates multi-component coordination with state tracking, conflict detection, and typed events.

🐍 Python (Production Example)

from __future__ import annotations
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from enum import Enum, auto
from typing import Dict, List, Optional, Set
import math


# ─── Domain Types ────────────────────────────────────────────────
class AircraftStatus(Enum):
    AIRBORNE    = auto()
    APPROACHING = auto()
    HOLDING     = auto()
    LANDING     = auto()
    LANDED      = auto()
    DEPARTING   = auto()
    DEPARTED    = auto()


@dataclass
class Position:
    x:        float  # Nautical miles east of tower
    y:        float  # Nautical miles north of tower
    altitude: float  # Feet

    def distance_to(self, other: "Position") -> float:
        return math.sqrt((self.x - other.x) ** 2 + (self.y - other.y) ** 2)

    def __str__(self) -> str:
        return f"({self.x:.1f}nm, {self.y:.1f}nm, {self.altitude:.0f}ft)"


# ─── Events ──────────────────────────────────────────────────────
class ATCEvent(Enum):
    REQUEST_LANDING   = auto()
    POSITION_UPDATE   = auto()
    EMERGENCY         = auto()
    READY_FOR_TAKEOFF = auto()
    AIRBORNE          = auto()
    LANDED            = auto()


@dataclass
class ATCMessage:
    event:    ATCEvent
    callsign: str
    data:     dict = field(default_factory=dict)


# ─── Mediator Interface ──────────────────────────────────────────
class ATCMediator(ABC):
    @abstractmethod
    def notify(self, message: ATCMessage) -> Optional[str]:
        """Returns an ATC instruction string, or None."""
        pass

    @abstractmethod
    def register(self, aircraft: "Aircraft") -> None:
        pass

    @abstractmethod
    def deregister(self, callsign: str) -> None:
        pass


# ─── Colleague ───────────────────────────────────────────────────
class Aircraft:
    def __init__(self, callsign: str, atc: ATCMediator) -> None:
        self.callsign = callsign
        self._atc     = atc
        self._status  = AircraftStatus.AIRBORNE
        self._position: Optional[Position] = None
        atc.register(self)

    # ── Aircraft actions → all go via ATC ──
    def request_landing(self) -> None:
        print(f"[{self.callsign}] Requesting landing clearance")
        instruction = self._atc.notify(ATCMessage(
            event=ATCEvent.REQUEST_LANDING, callsign=self.callsign
        ))
        if instruction:
            print(f"[{self.callsign}] ATC: {instruction}")
            self._status = AircraftStatus.APPROACHING

    def update_position(self, pos: Position) -> None:
        self._position = pos
        instruction = self._atc.notify(ATCMessage(
            event=ATCEvent.POSITION_UPDATE, callsign=self.callsign,
            data={"position": pos}
        ))
        if instruction:
            print(f"[{self.callsign}] ATC: {instruction}")

    def declare_emergency(self, reason: str) -> None:
        print(f"[{self.callsign}] ⚠ DECLARING EMERGENCY: {reason}")
        self._atc.notify(ATCMessage(
            event=ATCEvent.EMERGENCY, callsign=self.callsign,
            data={"reason": reason}
        ))

    def ready_for_takeoff(self) -> None:
        print(f"[{self.callsign}] Ready for departure")
        instruction = self._atc.notify(ATCMessage(
            event=ATCEvent.READY_FOR_TAKEOFF, callsign=self.callsign
        ))
        if instruction:
            print(f"[{self.callsign}] ATC: {instruction}")
            self._status = AircraftStatus.DEPARTING

    def confirm_landed(self) -> None:
        self._status = AircraftStatus.LANDED
        self._atc.notify(ATCMessage(event=ATCEvent.LANDED, callsign=self.callsign))

    def confirm_airborne(self) -> None:
        self._status = AircraftStatus.DEPARTED
        self._atc.notify(ATCMessage(event=ATCEvent.AIRBORNE, callsign=self.callsign))

    @property
    def status(self) -> AircraftStatus:
        return self._status

    @property
    def position(self) -> Optional[Position]:
        return self._position

    def __str__(self) -> str:
        pos_str = str(self._position) if self._position else "unknown"
        return f"Aircraft({self.callsign}, {self._status.name}, pos={pos_str})"


# ─── Concrete Mediator: ATC Tower ───────────────────────────────
class ATCTower(ATCMediator):
    SEPARATION_NM      = 5.0   # Minimum horizontal separation (nautical miles)
    APPROACH_RADIUS_NM = 30.0  # Aircraft within this range are managed closely

    def __init__(self, airport_code: str, runways: List[str]) -> None:
        self._code      = airport_code
        self._aircraft: Dict[str, Aircraft] = {}
        self._runways:  Dict[str, Optional[str]] = {r: None for r in runways}  # runway → callsign
        self._queue:    List[str] = []  # Landing queue
        self._log:      List[str] = []

    # ── Registration ──
    def register(self, aircraft: Aircraft) -> None:
        self._aircraft[aircraft.callsign] = aircraft
        self._log_event(f"Registered {aircraft.callsign}")
        print(f"[ATC:{self._code}] Registered {aircraft.callsign}")

    def deregister(self, callsign: str) -> None:
        self._aircraft.pop(callsign, None)
        self._queue = [c for c in self._queue if c != callsign]
        self._log_event(f"Deregistered {callsign}")

    # ── Central Coordination Logic ────────────────────────────────
    def notify(self, message: ATCMessage) -> Optional[str]:
        callsign = message.callsign
        event    = message.event

        if event == ATCEvent.REQUEST_LANDING:
            return self._handle_landing_request(callsign)

        elif event == ATCEvent.POSITION_UPDATE:
            pos = message.data.get("position")
            if pos:
                return self._handle_position_update(callsign, pos)

        elif event == ATCEvent.EMERGENCY:
            reason = message.data.get("reason", "unknown")
            self._handle_emergency(callsign, reason)

        elif event == ATCEvent.READY_FOR_TAKEOFF:
            return self._handle_takeoff_request(callsign)

        elif event == ATCEvent.LANDED:
            self._handle_landed(callsign)

        elif event == ATCEvent.AIRBORNE:
            self._handle_airborne(callsign)

        return None

    def _handle_landing_request(self, callsign: str) -> str:
        free_runway = self._find_free_runway()
        if free_runway:
            self._runways[free_runway] = callsign
            self._log_event(f"{callsign} cleared to land on {free_runway}")
            # Notify all other aircraft of new traffic
            self._alert_others(callsign, f"Traffic: {callsign} on approach to {free_runway}")
            return f"Cleared to land on runway {free_runway}. Wind 270 at 10."
        else:
            self._queue.append(callsign)
            pos = len(self._queue)
            self._log_event(f"{callsign} entering holding pattern (queue pos {pos})")
            return f"All runways occupied. Enter holding pattern. Position {pos} in queue."

    def _handle_position_update(self, callsign: str, pos: Position) -> Optional[str]:
        # Check separation with other aircraft
        conflicts = self._check_separation(callsign, pos)
        if conflicts:
            warning = f"TRAFFIC ALERT: {callsign} within {self.SEPARATION_NM}nm of {', '.join(conflicts)}"
            self._log_event(warning)
            print(f"[ATC:{self._code}] ⚠ {warning}")
            return f"Traffic alert: {', '.join(conflicts)} at your proximity. Adjust heading."
        return None

    def _handle_emergency(self, callsign: str, reason: str) -> None:
        self._log_event(f"EMERGENCY: {callsign} — {reason}")
        print(f"[ATC:{self._code}] 🚨 EMERGENCY DECLARED by {callsign}: {reason}")
        # Clear a runway immediately
        free_runway = self._find_free_runway()
        if not free_runway:
            # Bump first non-emergency aircraft
            for runway, occupant in self._runways.items():
                if occupant and occupant != callsign:
                    self._log_event(f"Vacating {runway} from {occupant} for emergency")
                    occ = self._aircraft.get(occupant)
                    if occ:
                        print(f"[ATC:{self._code}] INSTRUCTING {occupant}: Go around immediately. Emergency traffic.")
                    self._runways[runway] = callsign
                    break
        else:
            self._runways[free_runway] = callsign
        # Alert all aircraft
        self._alert_all(f"EMERGENCY TRAFFIC: {callsign}. All aircraft maintain positions.")
        aircraft = self._aircraft.get(callsign)
        if aircraft:
            print(f"[ATC:{self._code}] [{callsign}] ATC: Emergency services dispatched. Runway cleared. Land immediately.")

    def _handle_takeoff_request(self, callsign: str) -> str:
        free_runway = self._find_free_runway()
        if free_runway:
            self._runways[free_runway] = callsign
            self._log_event(f"{callsign} cleared for takeoff on {free_runway}")
            self._alert_others(callsign, f"Departing traffic: {callsign} on {free_runway}")
            return f"Cleared for takeoff on runway {free_runway}. Wind 270 at 10. Good day."
        return "Hold position. Runway not available."

    def _handle_landed(self, callsign: str) -> None:
        for runway, occupant in self._runways.items():
            if occupant == callsign:
                self._runways[runway] = None
                self._log_event(f"{callsign} vacated {runway}")
                print(f"[ATC:{self._code}] {callsign} vacated runway {runway}")
                # Clear next in queue
                if self._queue:
                    next_cs = self._queue.pop(0)
                    next_ac = self._aircraft.get(next_cs)
                    if next_ac:
                        self._runways[runway] = next_cs
                        print(f"[ATC:{self._code}] [{next_cs}] ATC: Cleared to land on runway {runway}. Runway now available.")
                break

    def _handle_airborne(self, callsign: str) -> None:
        for runway, occupant in self._runways.items():
            if occupant == callsign:
                self._runways[runway] = None
                self._log_event(f"{callsign} airborne, vacated {runway}")
                break
        self.deregister(callsign)
        print(f"[ATC:{self._code}] {callsign} departed. Safe travels.")

    # ── Helpers ───────────────────────────────────────────────────
    def _find_free_runway(self) -> Optional[str]:
        for runway, occupant in self._runways.items():
            if occupant is None:
                return runway
        return None

    def _check_separation(self, callsign: str, pos: Position) -> List[str]:
        conflicts = []
        for cs, aircraft in self._aircraft.items():
            if cs == callsign or aircraft.position is None:
                continue
            dist = pos.distance_to(aircraft.position)
            if dist < self.SEPARATION_NM:
                conflicts.append(cs)
        return conflicts

    def _alert_others(self, exclude_callsign: str, notice: str) -> None:
        for cs, aircraft in self._aircraft.items():
            if cs != exclude_callsign:
                print(f"[ATC:{self._code}] [{cs}] ATC: {notice}")

    def _alert_all(self, notice: str) -> None:
        for cs in self._aircraft:
            print(f"[ATC:{self._code}] [{cs}] ATC: {notice}")

    def _log_event(self, event: str) -> None:
        self._log.append(event)

    def status_board(self) -> None:
        print(f"\n[ATC:{self._code}] ── STATUS BOARD ──────────────────────")
        print(f"  Runways: { {r: (v or 'FREE') for r, v in self._runways.items()} }")
        print(f"  Queue:   {self._queue or '(empty)'}")
        print(f"  Aircraft: {list(self._aircraft.keys())}")
        print(f"  Log ({len(self._log)} entries): last 5:")
        for entry in self._log[-5:]:
            print(f"    • {entry}")


# ─── Client ──────────────────────────────────────────────────────
if __name__ == "__main__":
    tower = ATCTower("EGLL", runways=["27L", "27R"])

    ua123  = Aircraft("UA123",  tower)
    ba456  = Aircraft("BA456",  tower)
    dl789  = Aircraft("DL789",  tower)
    medic1 = Aircraft("MEDIC1", tower)

    print("\n=== Phase 1: Normal operations ===")
    ua123.request_landing()
    ba456.request_landing()
    dl789.request_landing()   # All runways occupied — goes to queue

    print("\n=== Phase 2: Position updates / separation check ===")
    ua123.update_position(Position(10.0, 5.0,  3000))
    ba456.update_position(Position(10.3, 5.1,  3100))  # Close to UA123 — conflict!
    dl789.update_position(Position(25.0, 12.0, 8000))

    print("\n=== Phase 3: UA123 lands, DL789 promoted from queue ===")
    ua123.confirm_landed()

    print("\n=== Phase 4: Emergency ===")
    medic1.declare_emergency("Engine failure")

    print("\n=== Phase 5: BA456 departs ===")
    ba456.confirm_landed()
    ba456.ready_for_takeoff()
    ba456.confirm_airborne()

    tower.status_board()

Real-World Use Cases

1. GUI Component Coordination

In graphical UIs, components (text fields, buttons, dropdowns, checkboxes) often affect each other: selecting "international shipping" enables a country dropdown and hides a domestic ZIP code field. Without a mediator, every component needs references to every other component. With a DialogMediator, each component just notifies the mediator when it changes, and the mediator handles all the cascading updates.

class ShippingDialogMediator:
    def notify(self, sender, event):
        if event == "shipping_type_changed":
            is_intl = sender.value == "international"
            self.country_field.visible  = is_intl
            self.zip_field.visible      = not is_intl
            self.submit_button.enabled  = self._form_valid()

2. Flight Booking / Reservation Systems

Online travel agencies coordinate between a flight service, a hotel service, a car rental service, and a payment gateway. A booking mediator orchestrates the workflow: reserve flight → reserve hotel → charge payment → send confirmation emails. If hotel reservation fails, the mediator cancels the flight reservation and processes a refund — all without the individual services knowing about each other.

3. Multiplayer Game Servers

In a multiplayer game, a game room (the mediator) coordinates all players. When a player fires a weapon, the server (mediator) checks hit detection against all other players, updates scores, broadcasts the event to all clients, and checks win conditions — none of which the individual player objects handle directly.

4. Event Bus / Message Bus

Frameworks like Spring's ApplicationEventPublisher, Android's EventBus, and Redux's store are all mediator implementations. Components publish events to the bus; the bus routes them to registered handlers. Publishers and handlers are completely decoupled.

5. Workflow Engines

Business process management systems (Camunda, Activiti) use a mediator to orchestrate workflow steps. When a loan application moves from "submitted" to "under review", the workflow mediator notifies the underwriting service, schedules a credit check, and sends an email to the applicant — all coordinated from a central workflow definition.

6. Smart Home Automation

A smart home hub (the mediator) coordinates between sensors, switches, lights, thermostats, and speakers. When a motion sensor fires, the hub decides: if it's night, turn on the hallway light; if the alarm is armed, trigger the siren and send a notification. The sensor doesn't know about lights; the light doesn't know about the alarm.

Language-Specific Mistakes and Anti-Patterns

# BAD: Mediator contains all business logic for every component
class GodMediator:
    def notify(self, sender, event):
        if event == "button_clicked":
            self._validate_input()
            self._save_to_db()
            self._send_email()
            self._update_ui()
            self._log_to_analytics()
            self._refresh_cache()
            # ... 200 more lines of entangled logic

# The mediator has become the monolith the pattern was meant to prevent.
# Every new feature requires editing this one massive class.

# GOOD: Each mediator method does one specific coordination job.
# Extract complex logic into domain services; the mediator just orchestrates.
class DialogMediator:
    def notify(self, sender, event):
        if event == "submit_clicked":
            if self._validator.validate(self._form.data):
                self._submission_service.submit(self._form.data)
                self._dialog.close()
        elif event == "cancel_clicked":
            self._dialog.close()

# BAD: Components know each other despite having a mediator
class TextBox:
    def __init__(self, mediator, submit_button):
        self._mediator      = mediator
        self._submit_button = submit_button  # Direct reference — defeats the pattern!

    def on_text_change(self):
        self._mediator.notify(self, "text_changed")
        self._submit_button.enabled = bool(self.text)  # Bypassing the mediator!

# GOOD: Colleagues ONLY communicate through the mediator
class TextBox:
    def __init__(self, mediator):
        self._mediator = mediator

    def on_text_change(self):
        self._mediator.notify(self, "text_changed")  # Only via mediator

# BAD: ComponentA notifies mediator → mediator updates ComponentB
#       → ComponentB's update triggers another notification → infinite loop!
class BrokenMediator:
    def notify(self, sender, event):
        if event == "value_changed":
            self.component_b.set_value(self.component_a.value)  # Updates B
            # BUT: set_value triggers component_b.on_change()
            #      which calls mediator.notify() again → infinite recursion!

# GOOD: Guard against re-entrant notifications
class SafeMediator:
    def __init__(self):
        self._handling = False

    def notify(self, sender, event):
        if self._handling:
            return  # Prevent re-entrant loop
        self._handling = True
        try:
            if event == "value_changed":
                # Update B without triggering its change notification
                self.component_b.set_value_silent(self.component_a.value)
        finally:
            self._handling = False

Frequently Asked Questions

Q1: What is the difference between the Mediator and Observer patterns?

Both decouple senders from receivers, but in different ways:

Observer is broadcast-and-forget. A publisher fires an event ("userLoggedIn") and every subscriber receives it. The publisher doesn't control what subscribers do, doesn't know their count, and doesn't coordinate between them. Subscribers react independently.

Mediator is centralized coordination. The mediator receives a notification from one component and actively decides which other components to call, in what order, with what arguments. The mediator knows all participants and controls the interaction flow. It enforces business rules about how components interact.

Use Observer when you want loose pub/sub event notification with no coordination logic. Use Mediator when multiple components must coordinate their behavior based on shared rules — when the mediator needs to look at the state of multiple components before deciding what to do.

Q2: When does the Mediator pattern become an anti-pattern?

The Mediator becomes an anti-pattern when it turns into a God Object — a class that knows and does too much. Signs of a mediator gone wrong:

• The mediator is the largest class in the codebase
• Every new feature requires editing the mediator
• The mediator contains domain business logic that belongs in services
• The mediator is impossible to test without setting up the entire system
• Developers avoid touching the mediator because it's too complex

The fix is to decompose the mediator: extract domain logic into services, use an event bus for loose coupling, or split one large mediator into several focused ones.

Q3: How does the Mediator relate to event buses and message brokers?

An event bus (RxJS, EventEmitter, Spring ApplicationEventPublisher) is a generalized, decentralized Mediator. Instead of one concrete mediator class, there is a bus that routes typed events to registered handlers. Publishers don't reference the mediator directly — they publish to a topic. Subscribers register interest in topics.

Message brokers (RabbitMQ, Kafka, AWS SNS/SQS) extend this to distributed systems — the broker is a Mediator that spans processes and machines, providing durability, retry, and fan-out.

Q4: Should the Mediator or the Colleague trigger the initial action?

Colleagues trigger their own actions (the user types, the sensor fires, the button is clicked). They notify the mediator of what happened. The mediator decides what to do next and calls methods on other colleagues. The colleague's notify() call is always about something that happened to the sender (past tense), not an order to another component.

This distinction matters: if a component says "mediator, TELL component B to do X", it's coupled to B's API. If it says "mediator, I just did X", the mediator decides how other components should respond.

Q5: Can I have multiple mediators in the same system?

Yes — and you often should. Rather than one mega-mediator for the whole application, decompose it into focused mediators:

• A LoginDialogMediator coordinates username/password/submit within the login dialog
• A CheckoutMediator coordinates cart, shipping, payment, and confirmation within checkout
• An ATCTower mediates aircraft in its sector; another tower mediates aircraft in an adjacent sector

Each mediator is small, focused, and independently testable. They can communicate with each other through events if cross-mediator coordination is needed.

Q6: How do I test a mediator?

Test the mediator with mock colleagues. Inject mock components, trigger events on the mediator, and assert that the correct methods were called on the correct mocks with the correct arguments.

from unittest.mock import Mock

def test_mediator_enables_submit_when_both_fields_filled():
    mediator    = DialogMediator()
    email_field = Mock(spec=TextField)
    name_field  = Mock(spec=TextField)
    submit_btn  = Mock(spec=Button)

    mediator.register("email",  email_field)
    mediator.register("name",   name_field)
    mediator.register("submit", submit_btn)

    email_field.value = "alice@example.com"
    name_field.value  = "Alice"
    mediator.notify(email_field, "text_changed")

    submit_btn.set_enabled.assert_called_once_with(True)

Q7: How does Mediator work with dependency injection?

Inject the mediator into colleagues via constructor injection (preferred) or setter injection. The mediator itself can be created by a DI container and injected wherever needed. This allows swapping the concrete mediator with a mock in tests.

# Production
room  = ChatRoom("General")
alice = User("Alice", room)

# Test — inject a mock mediator
mock_mediator = Mock(spec=ChatMediator)
alice = User("Alice", mock_mediator)
alice.send("Hello")
mock_mediator.send_message.assert_called_once_with("Hello", alice)

Q8: How does the Mediator pattern differ from a Service Layer?

A Service Layer (application service, use-case handler) orchestrates business operations by calling domain objects and repositories. It's called by the UI/API layer and orchestrates from outside the domain.

A Mediator orchestrates components from within a component group — it's called by the components themselves when they change state. The mediator IS a participant in the group.

In practice, the distinction blurs: CQRS command handlers and MediatR (the .NET library) call themselves mediators but are essentially service/command handlers.

Q9: What is MediatR in .NET and how does it relate to the Mediator pattern?

MediatR is a popular .NET library that implements a mediator for CQRS (Command Query Responsibility Segregation). IMediator.Send(command) routes a command object to the single registered handler for that command type. IMediator.Publish(notification) broadcasts to all registered handlers.

It's the Mediator pattern applied to the application layer — handlers are colleagues, the MediatR instance is the mediator. It decouples controllers from services without direct constructor injection chains.

Q10: When should I NOT use the Mediator pattern?

Avoid the Mediator pattern when:

• Only two components interact: Adding a mediator for A ↔ B adds overhead with no benefit. Just let them reference each other or use an event.
• Interactions are simple and static: If the coordination logic never changes and is trivially understood, a direct reference is clearer.
• The mediator would be a pass-through: If the mediator just calls component B when it hears from component A with no logic, a direct call or Observer is simpler.
• Performance is critical: Every interaction goes through an extra method call. For very high-frequency interactions (thousands per second), the indirection may matter.
• The team isn't familiar with the pattern: A misapplied Mediator (God Object) is worse than direct references. Apply it only when the component coupling problem is clearly present.

Key Takeaways

🎯 Core Concept: The Mediator Pattern introduces a central hub that all components (colleagues) talk to instead of talking to each other. The mediator receives notifications from colleagues, applies coordination logic, and calls the appropriate methods on other colleagues. Components are decoupled from each other; all coupling goes through the mediator.

🔑 Key Benefits:

• Eliminated spaghetti dependencies: N components with direct references = N² possible couplings. With a mediator: N components × 1 mediator reference = N couplings
• Centralized rules: Interaction logic lives in one place — easy to find, change, and audit
• Loose coupling: Components can be added, removed, or replaced without touching each other
• Single Responsibility: Each component does its own job; the mediator does coordination
• Testability: Mock the mediator to test each component; mock the components to test the mediator

🌐 Language-Specific Highlights:

• Python: Guard against re-entrant notification loops with an _handling flag; always validate sender membership in send_message(); avoid the mediator becoming a God Object by extracting domain logic to services
• TypeScript: Type the mediator interface explicitly — never any; use a component registry map rather than hardcoded concrete type fields; return typed response objects when components need clearance/responses
• Java: Avoid instanceof checks — route by event string or typed message; use ConcurrentHashMap for thread-safe colleague registries; avoid re-entrant notification by copying the user list before iteration
• JavaScript: Validate sender membership at the start of every send; share one mediator instance across all colleagues in a group — never create per-colleague mediators; use private class fields (#) to prevent external access to the colleague registry
• C#: Use WeakReference<T> for colleague references in long-lived mediators (prevents memory leaks); return string? or typed response records from Notify() when components need responses; never async void — use async Task for async mediator methods
• PHP: Always exclude the sender from broadcast with $user !== $sender; prefer IteratorAggregate for the colleague collection; use constructor promotion for clean mediator injection
• Go: Never hold a mutex during a mediator call (deadlock risk); use context.Context for goroutine lifecycle in async mediators; use sync.RWMutex for read-heavy colleague registries; guard map access with mutexes
• Rust: Use Weak<RefCell<>> for colleague references to the mediator to break reference cycles; use Rc<RefCell<>> for the mediator so it can be shared; consider message-passing channels as an alternative to shared state
• Dart: Never cache references to peers discovered via the mediator; use abstract class for the mediator interface; prefer named parameters in broadcastSystem() calls for clarity
• Swift: Always use weak var mediator in colleagues to break retain cycles; use optional chaining mediator?.sendMessage() for safety after the mediator might have been deallocated; protocol-based mediator interface for testability
• Kotlin: Keep the colleague registry private — never expose it publicly; use class instances (not object singletons) so multiple mediators can coexist; use LinkedHashMap for predictable iteration order in broadcast

📋 When to Use Mediator:

• Multiple components interact in complex ways and direct references create a web of coupling
• Component interaction rules need to be centralized so they're easy to change
• Components should be reusable independently — they can't be if they hold references to each other
• You're building a GUI dialog where multiple form elements affect each other
• You need a chat room, game room, or collaboration session where participants communicate through a hub
• You're orchestrating a workflow involving multiple services that should not know about each other

⚠️ When NOT to Use Mediator:

• Only two components interact — a direct reference or Observer is simpler
• The coordination logic is trivially simple — a mediator adds indirection without benefit
• The "mediator" would just be a pass-through with no logic — use Observer or a direct call instead
• Performance of very high-frequency component interactions is critical — each call goes through an extra level of indirection
• The mediator is already becoming a God Object — decompose it into focused mediators or extract logic to services

🛠️ Best Practices Across Languages:

1. Colleagues own only a mediator reference: Never let colleagues store references to each other — all communication through the hub
2. Notify about what happened, not what should happen: Colleagues say "I changed" (past tense event); the mediator decides consequences
3. Keep the mediator focused: One mediator per cohesive group of components. Multiple mediators are better than one God Object
4. Extract business logic from the mediator: The mediator coordinates; domain services execute. Don't put billing logic in the chat room mediator
5. Use weak references for colleague references (where the language requires it): Prevent memory leaks in long-lived mediators
6. Make the mediator interface, not the concrete class: Colleagues depend on the abstraction — easy to mock and test
7. Guard against re-entrant calls: Component A → mediator → component B → mediator → A can cause infinite loops if not guarded
8. Return responses when needed: Not all notify() calls are fire-and-forget; sometimes components need clearance back (landing permission, form validation result)

💡 Common Pitfalls:

• God Object Mediator: Mediator grows to contain all application logic — defeat the entire point of the pattern. Extract logic to services.
• Direct peer references despite a mediator: Colleagues hold references to each other "just for convenience" — the mediator is bypassed
• Circular notification loops: A→mediator→B→mediator→A with no loop guard causes infinite recursion
• One mediator per colleague: Every component has its own private mediator — components can't communicate at all
• Thread-unsafe mediator: Colleagues registered from multiple threads; mediator's colleague map modified and iterated concurrently without synchronization
• Retain cycles (Swift/Rust/ObjC): Strong mediator → strong colleagues → strong mediator = memory that's never freed
• Mediator knows concrete types: Using instanceof / is checks instead of event strings or typed messages — every new component type breaks the mediator

🎁 Advanced Techniques:

• Event Bus variant: Decouple further by having colleagues publish typed event objects to a bus rather than calling mediator.notify(self, "event"). The bus routes events to registered handlers. Publishers and handlers never share a reference.
• Async mediator with coroutines: In Kotlin, use Flow or Channel for async event passing through the mediator. In Python, use asyncio.Queue. Colleagues send events asynchronously; the mediator consumes and dispatches without blocking.
• State machine mediator: The mediator implements a finite state machine. Each state determines which components are active and how notifications are routed. State transitions happen inside the mediator based on events received.
• CQRS command bus: The MediatR library pattern — typed command objects are published to the mediator; a single registered handler processes each command type. Decouples controllers from service/domain logic.
• Hierarchical mediators: A local mediator manages a widget group; a parent mediator manages widget groups; a root mediator manages the whole page. Messages can bubble up or down the hierarchy, similar to DOM event bubbling.
• Mediator with history/replay: The mediator logs all events. On startup or after a crash, it replays the event log to restore state — essential for collaborative real-time applications (Google Docs-style co-editing).

The Mediator Pattern is the architecture that makes complex multi-component systems manageable. The moment you find yourself adding the fifth direct reference between UI components, or the third service that needs to know about four others, the Mediator Pattern turns a web of dependencies into a star topology with a single, understandable hub at the center.

Want to dive deeper into design patterns? Check out our comprehensive guides on:

• Observer Pattern
• Command Pattern
• Chain of Responsibility Pattern
• Facade Pattern
• Iterator Pattern Each guide includes examples in all 11 programming languages with language-specific best practices.