Observer and Pub/Sub Pattern

At the Jurassic Park control center, Robert Muldoon observes dozens of monitors simultaneously. "The problem is," he explains, "that every system - fences, cameras, motion sensors - must immediately react to changes in a dinosaur's status. But they can't keep polling (checking) the database all the time!" The solution is the Observer pattern - a mechanism where objects automatically notify interested parties about changes.

Problem: Tight Component Coupling

Without the Observer pattern, components must directly know and call each other. The function that reacts to an escape then becomes a switchboard that phones every system by hand:

1// Problem: the alarm system knows ALL other systems
2function onDinosaurEscaped(dinosaur) {
3  // Must manually call each system
4  fenceSystem.activateEmergency(dinosaur);
5  cameraSystem.trackDinosaur(dinosaur);
6  alarmSystem.soundAlert(dinosaur);
7  gateSystem.lockAllGates();
8  evacuationSystem.startEvacuation();
9  // What if we add a new system? We have to modify this function!
10}

Each new system requires modifying existing code. This violates the Open-Closed Principle, which says code should be open for extension but closed for modification.

Observer Pattern - Basics

In the Observer pattern we have two types of objects:

• Subject (Observable) - an object that emits events (notifies about changes)
• Observer - an object that listens for events and reacts

The subject is the park siren, and the observers are the rangers who signed up to hear it. The simplest implementation is an EventEmitter, which keeps a list of callback functions for every event name:

1// Simple EventEmitter - the heart of the Observer pattern
2class EventEmitter {
3  constructor() {
4    this.listeners = {};
5  }
6
7  // Register an observer (subscribe)
8  on(event, callback) {
9    if (!this.listeners[event]) {
10      this.listeners[event] = [];
11    }
12    this.listeners[event].push(callback);
13    return this; // Enables chaining
14  }
15
16  // Emit an event (notify observers)
17  emit(event, ...args) {
18    const callbacks = this.listeners[event];
19    if (callbacks) {
20      callbacks.forEach(callback => callback(...args));
21    }
22    return this;
23  }
24
25  // Remove an observer
26  off(event, callback) {
27    if (this.listeners[event]) {
28      this.listeners[event] = this.listeners[event].filter(
29        cb => cb !== callback
30      );
31    }
32    return this;
33  }
34
35  // One-time subscription
36  once(event, callback) {
37    const wrapper = (...args) => {
38      callback(...args);
39      this.off(event, wrapper);
40    };
41    this.on(event, wrapper);
42    return this;
43  }
44}

The listeners object is a registry: an event name leads to an array of functions. on() adds a function, and emit() calls all of them synchronously, in registration order. off() compares functions by reference, so you cannot detach an anonymous arrow function passed straight to on(). once() wraps the callback in a function that unregisters itself after the first call.

Application in Jurassic Park

DinosaurMonitor extends EventEmitter, so it gets on(), emit() and off() for free. It only takes care of dinosaur data and announcing changes:

1// Dinosaur monitoring system based on EventEmitter
2class DinosaurMonitor extends EventEmitter {
3  constructor() {
4    super();
5    this.dinosaurs = new Map();
6  }
7
8  registerDinosaur(id, data) {
9    this.dinosaurs.set(id, { ...data, status: "contained" });
10    this.emit("dinosaur:registered", { id, ...data });
11  }
12
13  updateStatus(id, newStatus) {
14    const dino = this.dinosaurs.get(id);
15    if (!dino) return;
16
17    const oldStatus = dino.status;
18    dino.status = newStatus;
19
20    this.emit("status:changed", { id, name: dino.name, oldStatus, newStatus });
21
22    // Emit a special event for escapes
23    if (newStatus === "escaped") {
24      this.emit("dinosaur:escaped", { id, name: dino.name, species: dino.species });
25    }
26  }
27}

The monitor knows nothing about fences, cameras or evacuation - it only knows it has to announce an event. On an escape it emits two: the general status:changed and the specific dinosaur:escaped.

Now each system connects to the monitor on its own, and we simulate a T-Rex escape:

1// Creating the system
2const monitor = new DinosaurMonitor();
3
4// Registering observers - each system connects itself
5monitor.on("dinosaur:escaped", (data) => {
6  console.log(`FENCE ALARM: ${data.name} escaped! Activating electric fence.`);
7});
8
9monitor.on("dinosaur:escaped", (data) => {
10  console.log(`CAMERA ALARM: Tracking ${data.name} (${data.species}) on cameras.`);
11});
12
13monitor.on("dinosaur:escaped", (data) => {
14  console.log(`EVACUATION ALARM: Starting evacuation due to ${data.name} escape!`);
15});
16
17monitor.on("status:changed", (data) => {
18  console.log(`LOG: ${data.name} changed status from "${data.oldStatus}" to "${data.newStatus}"`);
19});
20
21// Register and simulate
22monitor.registerDinosaur("TREX-001", { name: "Rexy", species: "T-Rex" });
23monitor.registerDinosaur("RAPTOR-01", { name: "Blue", species: "Velociraptor" });
24
25// Simulate escape
26monitor.updateStatus("TREX-001", "escaped");
27// LOG: Rexy changed status from "contained" to "escaped"
28// FENCE ALARM: Rexy escaped! Activating electric fence.
29// CAMERA ALARM: Tracking Rexy (T-Rex) on cameras.
30// EVACUATION ALARM: Starting evacuation due to Rexy escape!

The order of the printouts follows the order of emit() calls in updateStatus(): first the log, then the three alarms in registration order. A new system, drones for example, takes a single monitor.on(...) call, without touching the monitor class. That is the main advantage of the pattern: lower coupling between components.

Publish/Subscribe (Pub/Sub) Pattern

Pub/Sub extends the Observer pattern where communication happens through channels/topics instead of direct references. The sender publishes to a topic, and receivers subscribe to topics without knowing anything about each other:

1class PubSub {
2  constructor() {
3    this.topics = {};
4    this.subscriptionId = 0;
5  }
6
7  // Subscribe to a topic
8  subscribe(topic, callback) {
9    if (!this.topics[topic]) {
10      this.topics[topic] = {};
11    }
12
13    const id = ++this.subscriptionId;
14    this.topics[topic][id] = callback;
15
16    // Return an unsubscribe function
17    return () => {
18      delete this.topics[topic][id];
19    };
20  }
21
22  // Publish a message
23  publish(topic, data) {
24    if (!this.topics[topic]) return;
25
26    Object.values(this.topics[topic]).forEach(callback => {
27      callback(data);
28    });
29  }
30}

Each subscription gets its own number, and subscribe() returns a function that cancels it. So there is no need to keep a reference to the callback, as there is with off().

The park's central bus handles two topics - security alerts and the feeding schedule:

1// Central park communication system
2const parkBus = new PubSub();
3
4// Systems subscribe to topics they care about
5const unsubFence = parkBus.subscribe("security:alert", (data) => {
6  console.log(`Fence: Alert received - ${data.message}`);
7});
8
9parkBus.subscribe("security:alert", (data) => {
10  console.log(`Command center: Level ${data.level} alert - ${data.message}`);
11});
12
13parkBus.subscribe("feeding:schedule", (data) => {
14  console.log(`Kitchen: Prepare ${data.food} for ${data.dinosaur}`);
15});
16
17// Publishing events - the sender doesn't know who is listening
18parkBus.publish("security:alert", {
19  level: "HIGH",
20  message: "Movement detected near sector C fence"
21});
22
23parkBus.publish("feeding:schedule", {
24  dinosaur: "Rexy",
25  food: "200kg beef",
26  time: "14:00"
27});
28
29// Unsubscribe
30unsubFence(); // Fence stops listening

The alert reaches the fence and the command center, while the feeding message goes only to the kitchen. After unsubFence() the fence stops receiving alerts, while the command center keeps listening. Names like security:alert are just a convention that groups topics together.

Practical Example: Notification System

Both ideas can be combined. ParkNotificationSystem extends EventEmitter with a history and a * channel that hears everything:

1class ParkNotificationSystem extends EventEmitter {
2  constructor() {
3    super();
4    this.history = [];
5  }
6
7  notify(channel, message, priority = "normal") {
8    const notification = {
9      channel,
10      message,
11      priority,
12      timestamp: new Date().toISOString()
13    };
14
15    this.history.push(notification);
16    this.emit(channel, notification);
17    this.emit("*", notification); // Wildcard - all notifications
18  }
19
20  getHistory(channel) {
21    if (channel) {
22      return this.history.filter(n => n.channel === channel);
23    }
24    return [...this.history];
25  }
26}

The notify() method emits every notification twice: on the target channel and on the * channel. getHistory() without an argument returns a copy of the history, so nobody outside can clear it by accident.

The admin panel subscribes to the asterisk, and the security and maintenance systems to their own channels:

1const notifications = new ParkNotificationSystem();
2
3// Admin panel - listens to everything
4notifications.on("*", (n) => {
5  console.log(`[ADMIN] [${n.priority.toUpperCase()}] ${n.channel}: ${n.message}`);
6});
7
8// Security system - only alerts
9notifications.on("security", (n) => {
10  if (n.priority === "critical") {
11    console.log(`RED ALARM: ${n.message}`);
12  }
13});
14
15// Logging
16notifications.on("maintenance", (n) => {
17  console.log(`Maintenance: ${n.message}`);
18});
19
20// Usage
21notifications.notify("security", "Camera failure in sector B", "normal");
22notifications.notify("security", "T-Rex broke through fence!", "critical");
23notifications.notify("maintenance", "Replace lab light bulbs", "low");
24
25console.log("Alert history:", notifications.getHistory("security"));

The admin panel prints all three notifications, the red alarm appears only for the critical priority, and the alert history contains two entries from the security channel.

Observer vs Pub/Sub - When to Use Which?

The key difference is the intermediary: in Observer the subject keeps the list of observers itself, while in Pub/Sub the parties meet only on the channel. Neither pattern is synchronous or asynchronous by definition - our implementations call callbacks synchronously, while in distributed systems Pub/Sub messages usually travel asynchronously through a broker.

Summary

Muldoon summarizes: "The Observer pattern is like the park's alarm system - you don't have to constantly check every sensor. Sensors will notify you themselves:"

1. Observer - an object (Subject) notifies registered observers about changes
2. EventEmitter - Observer implementation with on(), emit(), off() methods
3. Pub/Sub - looser coupling through channels/topics, sender doesn't know receivers
4. Decoupling - components don't have to directly know each other, they communicate through events

These patterns are the foundation of many libraries and frameworks - from Node.js (EventEmitter) through React (state management) to microservice architectures.

In a real project don't write an emitter from scratch: Node.js has a ready-made EventEmitter in the node:events module, and the browser has EventTarget with addEventListener(). You will meet observers built into the browser, such as IntersectionObserver, later in this location. In the lab below you will extend PubSub with a once() method.

Remember: with Observer, the sensors call the rangers - the rangers don't patrol the enclosures with a flashlight.

1// The Observer and Pub/Sub patterns
2// Jurassic Park - Notification system
3
4console.log("=== EVENT EMITTER ===");
5
6// A simple EventEmitter implementation
7class EventEmitter {
8  constructor() {
9    this.listeners = {};
10  }
11
12  on(event, callback) {
13    if (!this.listeners[event]) {
14      this.listeners[event] = [];
15    }
16    this.listeners[event].push(callback);
17    return this;
18  }
19
20  emit(event, ...args) {
21    const callbacks = this.listeners[event];
22    if (callbacks) {
23      callbacks.forEach(callback => callback(...args));
24    }
25    return this;
26  }
27
28  off(event, callback) {
29    if (this.listeners[event]) {
30      this.listeners[event] = this.listeners[event].filter(
31        cb => cb !== callback
32      );
33    }
34    return this;
35  }
36
37  once(event, callback) {
38    const wrapper = (...args) => {
39      callback(...args);
40      this.off(event, wrapper);
41    };
42    this.on(event, wrapper);
43    return this;
44  }
45}
46
47console.log("\n=== MONITORING SYSTEM ===");
48
49// Dinosaur monitoring system
50class DinosaurMonitor extends EventEmitter {
51  constructor() {
52    super();
53    this.dinosaurs = new Map();
54  }
55
56  register(id, data) {
57    this.dinosaurs.set(id, { ...data, status: "contained" });
58    this.emit("registered", { id, ...data });
59  }
60
61  updateStatus(id, newStatus) {
62    const dino = this.dinosaurs.get(id);
63    if (!dino) return;
64    const oldStatus = dino.status;
65    dino.status = newStatus;
66    this.emit("status:changed", { id, name: dino.name, oldStatus, newStatus });
67    if (newStatus === "escaped") {
68      this.emit("dinosaur:escaped", { id, name: dino.name, species: dino.species });
69    }
70  }
71}
72
73const monitor = new DinosaurMonitor();
74
75// Registering observers
76monitor.on("dinosaur:escaped", (data) => {
77  console.log(`ALARM: ${data.name} escaped! Activating the fence.`);
78});
79
80monitor.on("dinosaur:escaped", (data) => {
81  console.log(`CAMERAS: Tracking ${data.name} (${data.species}).`);
82});
83
84monitor.on("status:changed", (data) => {
85  console.log(`LOG: ${data.name}: "${data.oldStatus}" -> "${data.newStatus}"`);
86});
87
88monitor.register("TREX-001", { name: "Rexy", species: "T-Rex" });
89monitor.register("RAPTOR-01", { name: "Blue", species: "Velociraptor" });
90
91console.log("\n--- Escape simulation ---");
92monitor.updateStatus("TREX-001", "escaped");
93
94console.log("\n=== PUB/SUB ===");
95
96// The Publish/Subscribe pattern
97class PubSub {
98  constructor() {
99    this.topics = {};
100    this.subId = 0;
101  }
102
103  subscribe(topic, callback) {
104    if (!this.topics[topic]) this.topics[topic] = {};
105    const id = ++this.subId;
106    this.topics[topic][id] = callback;
107    return () => { delete this.topics[topic][id]; };
108  }
109
110  publish(topic, data) {
111    if (!this.topics[topic]) return;
112    Object.values(this.topics[topic]).forEach(cb => cb(data));
113  }
114}
115
116const parkBus = new PubSub();
117
118parkBus.subscribe("security:alert", (data) => {
119  console.log(`Fence: Alert - ${data.message}`);
120});
121
122parkBus.subscribe("security:alert", (data) => {
123  console.log(`Control center: Level ${data.level} - ${data.message}`);
124});
125
126const unsubFeeding = parkBus.subscribe("feeding", (data) => {
127  console.log(`Kitchen: ${data.food} for ${data.dinosaur}`);
128});
129
130parkBus.publish("security:alert", {
131  level: "HIGH",
132  message: "Movement at the sector C fence"
133});
134
135parkBus.publish("feeding", {
136  dinosaur: "Rexy",
137  food: "200kg of beef"
138});
139
140// Unsubscribing
141unsubFeeding();
142console.log("\nAfter unsubscribing from feeding:");
143parkBus.publish("feeding", { dinosaur: "Blue", food: "50kg of meat" });
144// Nothing will be displayed
145
146// TODO: Add a once() method to PubSub
147// TODO: Create a notification system with an event history
148console.log("\n=== Exercise ===");
149console.log("Extend PubSub with a once() method - see the TODO");