Wzorzec Observer i Pub/Sub

W centrum kontroli Parku Jurajskiego, Robert Muldoon obserwuje dziesiątki monitorów jednocześnie. "Problem polega na tym," wyjaśnia, "że każdy system - ogrodzenia, kamery, czujniki ruchu - musi natychmiast reagować na zmiany statusu dinozaura. Ale nie mogą ciągle sprawdzać (pollować) stanu bazy danych!" Rozwiązaniem jest wzorzec Observer - mechanizm, w którym obiekty automatycznie powiadamiają zainteresowanych o zmianach.

Problem: Ścisłe powiązanie komponentów

Bez wzorca Observer komponenty muszą się bezpośrednio znać i wywoływać. Funkcja reagująca na ucieczkę staje się wtedy centralą, która ręcznie obdzwania każdy system:

1// Problem: system alarmowy zna WSZYSTKIE inne systemy
2function onDinosaurEscaped(dinosaur) {
3  // Musi ręcznie wywołać każdy system
4  fenceSystem.activateEmergency(dinosaur);
5  cameraSystem.trackDinosaur(dinosaur);
6  alarmSystem.soundAlert(dinosaur);
7  gateSystem.lockAllGates();
8  evacuationSystem.startEvacuation();
9  // Co jeśli dodamy nowy system? Trzeba zmodyfikować tę funkcję!
10}

Każdy nowy system wymaga modyfikacji istniejącego kodu. To łamie zasadę otwarte-zamknięte (Open-Closed Principle), według której kod powinien być otwarty na rozszerzenia, ale zamknięty na modyfikacje.

Wzorzec Observer - podstawy

We wzorcu Observer mamy dwa typy obiektów:

• Subject (Observable) - obiekt, który emituje zdarzenia (powiadamia o zmianach)
• Observer - obiekt, który nasłuchuje na zdarzenia i reaguje

Subject to syrena parku, a obserwatorzy to strażnicy, którzy sami zgłosili, że chcą ją słyszeć. Najprostsza implementacja to EventEmitter, który trzyma listę funkcji zwrotnych dla każdej nazwy zdarzenia:

1// Prosty EventEmitter - serce wzorca Observer
2class EventEmitter {
3  constructor() {
4    this.listeners = {};
5  }
6
7  // Rejestracja obserwatora (subskrypcja)
8  on(event, callback) {
9    if (!this.listeners[event]) {
10      this.listeners[event] = [];
11    }
12    this.listeners[event].push(callback);
13    return this; // Umożliwia chaining
14  }
15
16  // Emisja zdarzenia (powiadomienie obserwatorów)
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  // Usunięcie obserwatora
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  // Jednorazowa subskrypcja
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}

Obiekt listeners to rejestr: nazwa zdarzenia prowadzi do tablicy funkcji. on() dopisuje funkcję, a emit() wywołuje wszystkie synchronicznie, w kolejności rejestracji. off() porównuje funkcje po referencji, więc anonimowej strzałki podanej wprost do on() nie odepniesz. once() owija callback w funkcję, która po pierwszym wywołaniu sama się wyrejestrowuje.

Zastosowanie w Parku Jurajskim

DinosaurMonitor dziedziczy po EventEmitter, więc dostaje on(), emit() i off() za darmo. Sam zajmuje się tylko danymi dinozaurów i ogłaszaniem zmian:

1// System monitoringu dinozaurów oparty na 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    // Emitujemy specjalne zdarzenie dla ucieczki
23    if (newStatus === "escaped") {
24      this.emit("dinosaur:escaped", { id, name: dino.name, species: dino.species });
25    }
26  }
27}

Monitor nie wie nic o ogrodzeniach, kamerach ani ewakuacji - wie tylko, że ma ogłosić zdarzenie. Przy ucieczce emituje dwa: ogólne status:changed i specjalne dinosaur:escaped.

Teraz każdy system sam podłącza się do monitora, a my symulujemy ucieczkę T-Reksa:

1// Tworzenie systemu
2const monitor = new DinosaurMonitor();
3
4// Rejestracja obserwatorów - każdy system sam się podłącza
5monitor.on("dinosaur:escaped", (data) => {
6  console.log(`ALARM OGRODZENIA: ${data.name} uciekł! Aktywuję elektryczne ogrodzenie.`);
7});
8
9monitor.on("dinosaur:escaped", (data) => {
10  console.log(`ALARM KAMERY: Śledzę ${data.name} (${data.species}) na kamerach.`);
11});
12
13monitor.on("dinosaur:escaped", (data) => {
14  console.log(`ALARM EWAKUACJI: Rozpoczynam ewakuację z powodu ucieczki ${data.name}!`);
15});
16
17monitor.on("status:changed", (data) => {
18  console.log(`LOG: ${data.name} zmienił status z "${data.oldStatus}" na "${data.newStatus}"`);
19});
20
21// Rejestracja i symulacja
22monitor.registerDinosaur("TREX-001", { name: "Rexy", species: "T-Rex" });
23monitor.registerDinosaur("RAPTOR-01", { name: "Blue", species: "Velociraptor" });
24
25// Symulacja ucieczki
26monitor.updateStatus("TREX-001", "escaped");
27// LOG: Rexy zmienił status z "contained" na "escaped"
28// ALARM OGRODZENIA: Rexy uciekł! Aktywuję elektryczne ogrodzenie.
29// ALARM KAMERY: Śledzę Rexy (T-Rex) na kamerach.
30// ALARM EWAKUACJI: Rozpoczynam ewakuację z powodu ucieczki Rexy!

Kolejność wydruków odpowiada kolejności emit() w updateStatus(): najpierw log, potem trzy alarmy w kolejności rejestracji. Nowy system, na przykład drony, dopiszesz jednym monitor.on(...), bez dotykania klasy monitora. To właśnie główna zaleta wzorca: mniejsze powiązanie (coupling) między komponentami.

Wzorzec Publish/Subscribe (Pub/Sub)

Pub/Sub to rozszerzenie wzorca Observer, gdzie komunikacja odbywa się przez kanały/tematy (topics) zamiast bezpośrednich referencji. Nadawca publikuje na temat, a odbiorcy subskrybują tematy, nie wiedząc nic o sobie nawzajem:

1class PubSub {
2  constructor() {
3    this.topics = {};
4    this.subscriptionId = 0;
5  }
6
7  // Subskrypcja tematu
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    // Zwracamy funkcję do anulowania subskrypcji
17    return () => {
18      delete this.topics[topic][id];
19    };
20  }
21
22  // Publikacja wiadomości
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}

Każda subskrypcja dostaje własny numer, a subscribe() zwraca funkcję, która ją anuluje. Nie trzeba więc przechowywać referencji do callbacku, jak przy off().

Centralna szyna parku obsługuje dwa tematy - alerty bezpieczeństwa i harmonogram karmienia:

1// Centralny system komunikacji parku
2const parkBus = new PubSub();
3
4// Systemy subskrybują tematy, które ich interesują
5const unsubFence = parkBus.subscribe("security:alert", (data) => {
6  console.log(`Ogrodzenie: Otrzymano alert - ${data.message}`);
7});
8
9parkBus.subscribe("security:alert", (data) => {
10  console.log(`Centrum dowodzenia: Alert poziomu ${data.level} - ${data.message}`);
11});
12
13parkBus.subscribe("feeding:schedule", (data) => {
14  console.log(`Kuchnia: Przygotować ${data.food} dla ${data.dinosaur}`);
15});
16
17// Publikacja zdarzeń - nadawca nie wie kto nasłuchuje
18parkBus.publish("security:alert", {
19  level: "HIGH",
20  message: "Wykryto ruch przy ogrodzeniu sektora C"
21});
22
23parkBus.publish("feeding:schedule", {
24  dinosaur: "Rexy",
25  food: "200kg wołowiny",
26  time: "14:00"
27});
28
29// Anulowanie subskrypcji
30unsubFence(); // Ogrodzenie przestaje nasłuchiwać

Alert trafia do ogrodzenia i centrum dowodzenia, a wiadomość o karmieniu tylko do kuchni. Po unsubFence() ogrodzenie przestaje dostawać alerty, a centrum słucha dalej. Nazwy w stylu security:alert to tylko konwencja, która porządkuje tematy w grupy.

Praktyczny przykład: System powiadomień

Oba pomysły można połączyć. ParkNotificationSystem rozszerza EventEmitter o historię i kanał *, który słyszy wszystko:

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 - wszystkie powiadomienia
18  }
19
20  getHistory(channel) {
21    if (channel) {
22      return this.history.filter(n => n.channel === channel);
23    }
24    return [...this.history];
25  }
26}

Metoda notify() emituje każde powiadomienie dwa razy: na kanale docelowym i na kanale *. getHistory() bez argumentu zwraca kopię historii, więc nikt z zewnątrz nie wyczyści jej przypadkiem.

Panel administratora podpina się pod gwiazdkę, a systemy bezpieczeństwa i konserwacji pod własne kanały:

1const notifications = new ParkNotificationSystem();
2
3// Panel administratora - nasłuchuje wszystko
4notifications.on("*", (n) => {
5  console.log(`[ADMIN] [${n.priority.toUpperCase()}] ${n.channel}: ${n.message}`);
6});
7
8// System bezpieczeństwa - tylko alerty
9notifications.on("security", (n) => {
10  if (n.priority === "critical") {
11    console.log(`CZERWONY ALARM: ${n.message}`);
12  }
13});
14
15// Logowanie
16notifications.on("maintenance", (n) => {
17  console.log(`Konserwacja: ${n.message}`);
18});
19
20// Użycie
21notifications.notify("security", "Awaria kamery w sektorze B", "normal");
22notifications.notify("security", "T-Rex przerwał ogrodzenie!", "critical");
23notifications.notify("maintenance", "Wymiana żarówek w laboratorium", "low");
24
25console.log("Historia alertów:", notifications.getHistory("security"));

Panel admina wypisze wszystkie trzy powiadomienia, czerwony alarm pojawi się tylko przy priorytecie critical, a historia alertów zawiera dwa wpisy z kanału security.

Observer vs Pub/Sub - kiedy używać?

Kluczowa różnica to pośrednik: w Observerze subject sam trzyma listę obserwatorów, a w Pub/Sub strony spotykają się tylko na kanale. Żaden z wzorców nie jest z definicji synchroniczny ani asynchroniczny - nasze implementacje wołają callbacki synchronicznie, a w systemach rozproszonych wiadomości Pub/Sub zwykle płyną asynchronicznie przez brokera.

Podsumowanie

Muldoon podsumowuje: "Wzorzec Observer to jak system alarmowy parku - nie musisz ciągle sprawdzać każdego czujnika. Czujniki same Cię powiadomią:"

1. Observer - obiekt (Subject) powiadamia zarejestrowanych obserwatorów o zmianach
2. EventEmitter - implementacja Observera z metodami on(), emit(), off()
3. Pub/Sub - luźniejsze powiązanie przez kanały/tematy, nadawca nie zna odbiorców
4. Odsprzęganie - komponenty nie muszą się bezpośrednio znać, komunikują się przez zdarzenia

Te wzorce są fundamentem wielu bibliotek i frameworków - od Node.js (EventEmitter) przez React (state management) po architektury mikroserwisowe.

W prawdziwym projekcie nie pisz emitera od zera: Node.js ma gotowy EventEmitter w module node:events, a przeglądarka EventTarget z addEventListener(). Obserwatory wbudowane w przeglądarkę, takie jak IntersectionObserver, poznasz jeszcze w tej lokacji. W laboratorium poniżej rozszerzysz PubSub o metodę once().

Pamiętaj: w Observerze to czujniki wołają strażników, a nie strażnicy obchodzą wybiegi z latarką.

1// Wzorzec Observer i Pub/Sub
2// Park Jurajski - System powiadomien
3
4console.log("=== EVENT EMITTER ===");
5
6// Prosta implementacja EventEmitter
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=== SYSTEM MONITORINGU ===");
48
49// System monitoringu dinozaurow
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// Rejestracja obserwatorow
76monitor.on("dinosaur:escaped", (data) => {
77  console.log(`ALARM: ${data.name} uciekl! Aktywuje ogrodzenie.`);
78});
79
80monitor.on("dinosaur:escaped", (data) => {
81  console.log(`KAMERY: Sledze ${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--- Symulacja ucieczki ---");
92monitor.updateStatus("TREX-001", "escaped");
93
94console.log("\n=== PUB/SUB ===");
95
96// Wzorzec Publish/Subscribe
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(`Ogrodzenie: Alert - ${data.message}`);
120});
121
122parkBus.subscribe("security:alert", (data) => {
123  console.log(`Centrum: Poziom ${data.level} - ${data.message}`);
124});
125
126const unsubFeeding = parkBus.subscribe("feeding", (data) => {
127  console.log(`Kuchnia: ${data.food} dla ${data.dinosaur}`);
128});
129
130parkBus.publish("security:alert", {
131  level: "HIGH",
132  message: "Ruch przy ogrodzeniu sektora C"
133});
134
135parkBus.publish("feeding", {
136  dinosaur: "Rexy",
137  food: "200kg wolowiny"
138});
139
140// Anulowanie subskrypcji
141unsubFeeding();
142console.log("\nPo anulowaniu subskrypcji feeding:");
143parkBus.publish("feeding", { dinosaur: "Blue", food: "50kg miesa" });
144// Nic sie nie wyswietli
145
146// TODO: Dodaj metode once() do PubSub
147// TODO: Stworz system powiadomien z historia zdarzen
148console.log("\n=== Cwiczenie ===");
149console.log("Rozszerz PubSub o metode once() - patrz TODO");