Design Patterns in TypeScript · Part 5 — Observer & Pub/Sub

Phần 5/10 trong series Design Patterns in TypeScript. Trước: Tiếp:

Đây là Phần 5 của series 10 bài về các design pattern mà mọi senior web nên nắm — giải thích bằng TypeScript chạy được, use case web thực tế, và bài tập ở cuối mỗi phần.

Trong Phần 4 bạn đổi thuật toán mà không sửa caller. Mùi liên quan là polling hoặc nối cứng mọi consumer vào producer: setInterval(() => checkCart(), 500), hoặc onOrderPlaced gọi thẳng sendEmail(), updateAnalytics(), refreshUI() trong một hàm. Điều đó ghép ai đổi với ai phải phản ứng. Observer và Pub/Sub cho phép bên quan tâm đăng ký và được báo khi có sự kiện — producer không cần biết tên họ.

Ý đồ

Observer định nghĩa quan hệ một-nhiều: khi subject đổi trạng thái, mọi observer gắn vào nó được báo tự động. Subject giữ danh sách observer (hoặc callback) và đẩy cập nhật. Bạn dùng khi nhiều phản ứng độc lập phải theo cùng một event — analytics, refresh UI, invalidate cache, audit log — và bạn không muốn sửa emitter mỗi lần product thêm listener.

Pub/Sub đẩy tách coupling thêm một bước: publisher và subscriber không tham chiếu nhau; broker hoặc event bus định tuyến message theo topic hoặc channel. Observer thường là “trong process, subject biết listener”; Pub/Sub là “ai biết bus đều publish hoặc nghe được”. Trong app TypeScript cả hai hiện dưới dạng event map có kiểu, EventTarget DOM, subscription store, và dispatch WebSocket.

Polling và nối cứng vs đăng ký

Polling hỏi “đã đổi chưa?” theo timer — tốn tài nguyên và trễ:

// ❌ couples timing to business logic; burns CPU when idle
setInterval(() => {
  if (cart.version !== lastSeen) refreshBadge(cart);
}, 500);

Nối cứng liệt kê mọi phản ứng trong producer:

// ❌ every new side effect edits this function
function placeOrder(order: Order) {
  persist(order);
  sendConfirmationEmail(order);
  trackPurchase(order);
  invalidateProductCache(order.productId);
  bus.publish('order:placed', order); // still better than inline imports everywhere
}

Cách sửa là báo, đừng gọi: emit một lần; subscriber đăng ký ở chỗ khác.

Event emitter có kiểu

Trọng tâm senior là Emitter<Events> nhỏ với Events là record tên event → kiểu payload. on, off, on, off, emit giữ type-safe: sai tên event hoặc payload là lỗi biên dịch.

/** Map of event names to payload types — the contract for this emitter. */
type EventMap = Record<string, unknown>;

type Handler<Events extends EventMap, K extends keyof Events> = (
  payload: Events[K],
) => void;

export class Emitter<Events extends EventMap> {
  private readonly listeners: {
    [K in keyof Events]?: Set<Handler<Events, K>>;
  } = {};

  on<K extends keyof Events>(event: K, handler: Handler<Events, K>): () => void {
    let set = this.listeners[event];
    if (!set) {
      set = new Set();
      this.listeners[event] = set;
    }
    set.add(handler);
    return () => this.off(event, handler);
  }

  off<K extends keyof Events>(event: K, handler: Handler<Events, K>): void {
    this.listeners[event]?.delete(handler);
  }

  emit<K extends keyof Events>(event: K, payload: Events[K]): void {
    const set = this.listeners[event];
    if (!set) return;
    // Copy so a handler that unsubscribes mid-loop does not skip others.
    for (const handler of [...set]) {
      try {
        handler(payload);
      } catch (error) {
        // Isolate failures — one bad observer must not break the rest.
        console.error(`[Emitter] handler failed for "${String(event)}"`, error);
      }
    }
  }
}

Định nghĩa contract một lần, rồi dùng:

type AppEvents = {
  'cart:updated': { itemCount: number };
  'user:signed-in': { userId: string };
};

const appBus = new Emitter<AppEvents>();

appBus.on('cart:updated', ({ itemCount }) => {
  document.querySelector('#badge')!.textContent = String(itemCount);
});

appBus.emit('cart:updated', { itemCount: 3 });
// appBus.emit('cart:updated', { count: 3 });     // ❌ compile error
// appBus.emit('user:signed-in', { itemCount: 1 }); // ❌ wrong payload shape

once và handle hủy đăng ký rõ ràng là ergonomics chuẩn:

export class Emitter<Events extends EventMap> {
  // ... on, off, emit as above

  once<K extends keyof Events>(event: K, handler: Handler<Events, K>): () => void {
    const wrapped: Handler<Events, K> = (payload) => {
      unsubscribe();
      handler(payload);
    };
    const unsubscribe = this.on(event, wrapped);
    return unsubscribe;
  }
}

Trả () => void từ on để caller dọn mà không cần giữ reference function gốc. Trong lifecycle React/Vue, gọi unsubscribe trong cleanup useEffect hoặc onUnmounted.

Nền tảng đã có sẵn

Browser và Node (bản mới) có sẵn EventTarget + CustomEvent. Dùng khi cần tương thích API DOM, semantics addEventListener, hoặc bubbling trên element. Dùng Emitter<Events> riêng khi muốn record có kiểu nhiều event tên mà không xoay detail kiểu.

const target = new EventTarget();

function isRecord(value: unknown): value is Record<string, unknown> {
  return typeof value === 'object' && value !== null;
}

function onTyped<T extends Record<string, unknown>>(
  target: EventTarget,
  type: string,
  handler: (detail: T) => void,
  isDetail: (value: unknown) => value is T,
  signal?: AbortSignal,
): void {
  const listener = (event: Event) => {
    if (!(event instanceof CustomEvent)) return;
    if (!isDetail(event.detail)) return;
    handler(event.detail);
  };
  target.addEventListener(type, listener, { signal });
}

function isItemDetail(value: unknown): value is { id: string } {
  return isRecord(value) && typeof value.id === 'string';
}

onTyped(
  target,
  'item:selected',
  (detail) => console.log(detail.id),
  isItemDetail,
  AbortSignal.timeout(60_000), // auto-unsubscribe after 60s
);

target.dispatchEvent(
  new CustomEvent('item:selected', { detail: { id: 'sku-42' } }),
);

AbortController / signal là pattern nền tảng tự hủy đăng ký: truyền signal vào addEventListener; gọi abort() (hoặc AbortSignal.timeout) và mọi listener gắn signal đó bị gỡ. Ưu tiên cho listener UI một lần gắn với vòng đời component.

Pub/Sub qua event bus

Khi module không được import lẫn nhau, expose module bus có kiểu: publisher chỉ publish, subscriber chỉ subscribe.

// events.ts — shared contract
export type DomainEvents = {
  'order:placed': { orderId: string; totalCents: number };
  'inventory:low': { sku: string; remaining: number };
};

// bus.ts — single broker instance
import { Emitter } from './emitter';
import type { DomainEvents } from './events';

const emitter = new Emitter<DomainEvents>();

export function publish<K extends keyof DomainEvents>(
  topic: K,
  payload: DomainEvents[K],
): void {
  emitter.emit(topic, payload);
}

export function subscribe<K extends keyof DomainEvents>(
  topic: K,
  handler: (payload: DomainEvents[K]) => void,
): () => void {
  return emitter.on(topic, handler);
}

// analytics.ts — never imports checkout.ts
import { subscribe } from './bus';

export function registerAnalytics(): () => void {
  return subscribe('order:placed', ({ orderId, totalCents }) => {
    track('purchase', { orderId, totalCents });
  });
}

// checkout.ts — never imports analytics.ts
import { publish } from './bus';

export function placeOrder(orderId: string, totalCents: number): void {
  saveOrder(orderId, totalCents);
  publish('order:placed', { orderId, totalCents });
}

Topic là key của DomainEvents; thêm topic là thay đổi cấp kiểu mọi subscriber thấy. Cho messaging cross-tab hoặc cross-service có thể đổi bus in-memory bằng Redis, MQTT, hoặc BroadcastChannel — pattern giữ nguyên.

Liên hệ với signal & reactivity

Reactivity framework (Vue ref, signal Solid/Preact, signal Angular, RxJS observable) là Observer có theo dõi dependency tự động. Bạn đọc signal trong effect; runtime ghi cạnh đó và chạy lại effect khi signal đổi. Bạn không on('change', …) thủ công cho mọi dependency — đồ thị được dựng từ lần đọc. Về ý niệm: subject = nguồn reactive, observer = effect / computed / subscriber. RxJS thêm operator (map, filter, debounce) trên cùng luồng báo — vẫn Observer, protocol phong phú hơn. Hiếm khi tự implement signal; bạn có implement emitter và bus có kiểu trong app.

Use case web thực tế

• DOM event — click, input, resize, visibilitychange; capture/bubble là đồ thị Observer của browser.
• Store state — subscribe Zustand, listener Redux, hoặc store tự gọi emit('state', slice) sau set.
• WebSocket / SSE — một connection dispatch kiểu message tới handler đã đăng ký.
• Giao tiếp cross-component — micro-frontend hoặc package không được import UI lẫn nhau.
• Domain event — order:placed, user:deleted cho audit, email, projection (frontend kiểu CQRS).

Cạm bẫy

1. Rò bộ nhớ (#1) — subscribe mà không unsubscribe khi component unmount hoặc request kết thúc giữ closure sống. Luôn trả và gọi hàm unsubscribe, hoặc dùng AbortSignal.
2. Bão emit / cascade — handler A emit B, B emit A; debounce hoặc queue nếu cần.
3. Giả định thứ tự — thứ tự duyệt Set là insertion order trong JS hiện đại, nhưng đừng dựa vào thứ tự giữa handler cho đúng nghiệp vụ.
4. Một observer ném lỗi — không try/catch từng handler thì một bug giết cả vòng báo. Cô lập (như emit ở trên).
5. Mất this — truyền obj.handleUpdate mất this; dùng arrow function hoặc handler.bind(obj).

// ❌ leak: subscription outlives the component
function mountWidget() {
  appBus.on('cart:updated', updateWidget);
  // never unsubscribes on teardown
}

// ✅ cleanup
function mountWidget(): () => void {
  return appBus.on('cart:updated', updateWidget);
}
// on teardown: const off = mountWidget(); ... later: off();

Bảng tra nhanh

// Typed in-process emitter
type Events = { 'x': { n: number } };
const bus = new Emitter<Events>();
const off = bus.on('x', (p) => console.log(p.n));
bus.emit('x', { n: 1 });
off();

// Platform EventTarget + auto-unsubscribe
const ac = new AbortController();
target.addEventListener('x', handler, { signal: ac.signal });
ac.abort();

// Pub/Sub: typed bus module, publishers ≠ subscribers
publish('order:placed', { orderId: '1', totalCents: 100 });
subscribe('order:placed', handler);

Quyết định: cùng module, ít listener → Emitter trên class; cross-module → bus có kiểu; tích hợp DOM → EventTarget; luồng async có operator → RxJS.

Bài tập / Exercises

1. Cài Emitter<Events> với on, off, emit sao cho emit('cart:updated', wrongShape) lỗi biên dịch.

type EventMap = Record<string, unknown>;
type Handler<Events extends EventMap, K extends keyof Events> = (
  payload: Events[K],
) => void;

class Emitter<Events extends EventMap> {
  private readonly listeners: {
    [K in keyof Events]?: Set<Handler<Events, K>>;
  } = {};

  on<K extends keyof Events>(event: K, handler: Handler<Events, K>): () => void {
    let set = this.listeners[event];
    if (!set) {
      set = new Set();
      this.listeners[event] = set;
    }
    set.add(handler);
    return () => this.off(event, handler);
  }

  off<K extends keyof Events>(event: K, handler: Handler<Events, K>): void {
    this.listeners[event]?.delete(handler);
  }

  emit<K extends keyof Events>(event: K, payload: Events[K]): void {
    const set = this.listeners[event];
    if (!set) return;
    for (const h of [...set]) h(payload);
  }
}

type ShopEvents = { 'cart:updated': { itemCount: number } };
const shop = new Emitter<ShopEvents>();
shop.on('cart:updated', ({ itemCount }) => console.log(itemCount));
shop.emit('cart:updated', { itemCount: 2 });

2. Thêm once(event, handler) và xác minh handler chạy đúng một lần dù emit hai lần.

// Extend Emitter with:
once<K extends keyof Events>(event: K, handler: Handler<Events, K>): () => void {
  const wrapped: Handler<Events, K> = (payload) => {
    off();
    handler(payload);
  };
  const off = this.on(event, wrapped);
  return off;
}

let calls = 0;
const off = shop.once('cart:updated', () => { calls += 1; });
shop.emit('cart:updated', { itemCount: 1 });
shop.emit('cart:updated', { itemCount: 2 });
console.log(calls); // 1
off(); // idempotent cleanup

3. Dùng EventTarget + CustomEvent + AbortController: đăng ký listener, dispatch một lần, gọi abort(), dispatch lần hai — chứng minh lần hai không chạy handler.

const target = new EventTarget();
const ac = new AbortController();
let hits = 0;

target.addEventListener(
  'ping',
  () => { hits += 1; },
  { signal: ac.signal },
);

target.dispatchEvent(new CustomEvent('ping'));
ac.abort();
target.dispatchEvent(new CustomEvent('ping'));
console.log(hits); // 1

4. Minh họa rò bộ nhớ: hàm đăng ký lên bus global mỗi lần gọi mà không unsubscribe. Sửa bằng cách trả và gọi unsubscribe.

const globalBus = new Emitter<{ tick: { n: number } }>();

// ❌ leak
function brokenMount() {
  globalBus.on('tick', () => { /* ... */ });
}
brokenMount();
brokenMount();
// two handlers — double work, retained closures

// ✅ idempotent mount
let teardown: (() => void) | undefined;

function safeMount(): void {
  teardown?.();
  teardown = globalBus.on('tick', () => { /* ... */ });
}

function safeUnmount(): void {
  teardown?.();
  teardown = undefined;
}

safeMount();
safeMount();
// still one handler after second mount
safeUnmount();

5. Phác subscribe(topic, handler) / publish(topic, payload) có kiểu cho DomainEvents với ít nhất hai topic và hai module subscriber không import lẫn nhau.

type DomainEvents = {
  'order:placed': { orderId: string };
  'inventory:low': { sku: string };
};

const core = new Emitter<DomainEvents>();

export function publish<K extends keyof DomainEvents>(
  topic: K,
  payload: DomainEvents[K],
): void {
  core.emit(topic, payload);
}

export function subscribe<K extends keyof DomainEvents>(
  topic: K,
  handler: (payload: DomainEvents[K]) => void,
): () => void {
  return core.on(topic, handler);
}

// module A
subscribe('order:placed', ({ orderId }) => console.log('audit', orderId));

// module B
subscribe('inventory:low', ({ sku }) => console.log('restock', sku));

publish('order:placed', { orderId: 'o-1' });

Nâng cao:Bọc handler Emitter để mỗi cái chạy trong queueMicrotask và log tên event + thời gian; emit mười event liên tiếp và xác nhận handler này không chặn handler kia (backpressure khái niệm).

class Emitter<Events extends EventMap> {
  protected readonly listeners: {
    [K in keyof Events]?: Set<Handler<Events, K>>;
  } = {};
  // on, off, emit ...
}

class AsyncEmitter<Events extends EventMap> extends Emitter<Events> {
  override emit<K extends keyof Events>(event: K, payload: Events[K]): void {
    const set = this.listeners[event];
    if (!set) return;
    for (const handler of [...set]) {
      queueMicrotask(() => {
        const t0 = performance.now();
        try {
          handler(payload);
        } finally {
          console.log(String(event), performance.now() - t0, 'ms');
        }
      });
    }
  }
}

Điểm chính

• Observer = báo nhiều listener khi state đổi; bỏ polling và bỏ nối cứng mọi phản ứng vào producer.
• Emitter<Events> với record payload cho an toàn compile-time cho on / emit.
• EventTarget + AbortSignal là câu chuyện hủy đăng ký nền tảng; dùng cho vòng đời gắn DOM.
• Pub/Sub thêm bus để publisher và subscriber không biết nhau.
• Signal/RxJS là Observer có tracking hoặc operator; việc của bạn trong app vẫn là kỷ luật subscribe/unsubscribe.
• Bug #1: leak — luôn ghép subscribe với teardown.

Tiếp theo

Phần 6 — Decorator & Middleware: bọc hành vi quanh hàm hoặc object lõi không cần bùng nổ subclass — logging, auth, cache, pipeline kiểu Express.