《互联网产品的客户端架构·内容、社交与实时互动》6.2 一条实时连接的生命周期

// file: src/realtime/connection.ts
// tier: L2
// verify: node --test --test-concurrency=1 test/realtime/connection.test.ts
export type ConnectionState =
  | "idle"
  | "connecting"
  | "open"
  | "reconnecting"
  | "closed"
  | "failed";

export type TransitionReason =
  | "start"
  | "opened"
  | "network-lost"
  | "retry"
  | "max-attempts"
  | "fatal"
  | "close"
  | "page-hidden"
  | "page-visible"
  | "dispose";

export interface TransitionEvent {
  from: ConnectionState;
  to: ConnectionState;
  reason: TransitionReason;
  at: number;
  attempt: number;
}

export interface ConnectionOptions {
  now?: () => number;
  maxAttempts?: number;
  baseDelayMs?: number;
  maxDelayMs?: number;
}

export type TransitionListener = (event: TransitionEvent) => void;

export function retryDelay(attempt: number, baseDelayMs = 500, maxDelayMs = 30_000): number {
  if (!Number.isInteger(attempt) || attempt < 1) throw new Error("attempt 必须是正整数");
  if (!Number.isFinite(baseDelayMs) || baseDelayMs <= 0) throw new Error("baseDelayMs 必须是正数");
  if (!Number.isFinite(maxDelayMs) || maxDelayMs < baseDelayMs) {
    throw new Error("maxDelayMs 不能小于 baseDelayMs");
  }
  return Math.min(maxDelayMs, baseDelayMs * 2 ** (attempt - 1));
}

export class ConnectionLifecycle {
  #state: ConnectionState = "idle";
  #attempt = 0;
  #nextRetryAt: number | null = null;
  #suspended = false;
  #disposed = false;
  #listeners = new Set<TransitionListener>();
  #now: () => number;
  #maxAttempts: number;
  #baseDelayMs: number;
  #maxDelayMs: number;

  constructor(options: ConnectionOptions = {}) {
    this.#now = options.now ?? (() => Date.now());
    this.#maxAttempts = options.maxAttempts ?? 5;
    this.#baseDelayMs = options.baseDelayMs ?? 500;
    this.#maxDelayMs = options.maxDelayMs ?? 30_000;
    if (!Number.isInteger(this.#maxAttempts) || this.#maxAttempts < 1) {
      throw new Error("maxAttempts 必须是正整数");
    }
    retryDelay(1, this.#baseDelayMs, this.#maxDelayMs);
  }

  get state(): ConnectionState {
    return this.#state;
  }

  get attempt(): number {
    return this.#attempt;
  }

  get nextRetryAt(): number | null {
    return this.#nextRetryAt;
  }

  get suspended(): boolean {
    return this.#suspended;
  }

  get disposed(): boolean {
    return this.#disposed;
  }

  onTransition(listener: TransitionListener): () => void {
    this.#listeners.add(listener);
    return () => {
      this.#listeners.delete(listener);
    };
  }

  start(): boolean {
    if (this.#disposed) return false;
    if (this.#state !== "idle" && this.#state !== "closed" && this.#state !== "failed") {
      return false;
    }
    this.#attempt = 0;
    this.#nextRetryAt = null;
    this.#suspended = false;
    this.#move("connecting", "start");
    return true;
  }

  opened(): boolean {
    if (this.#disposed) return false;
    if (this.#state !== "connecting") return false;
    this.#attempt = 0;
    this.#nextRetryAt = null;
    this.#move("open", "opened");
    return true;
  }

  lost(reason: TransitionReason = "network-lost"): boolean {
    if (this.#disposed) return false;
    if (this.#state !== "open" && this.#state !== "connecting") return false;
    this.#attempt += 1;
    if (this.#attempt > this.#maxAttempts) {
      this.#nextRetryAt = null;
      this.#move("failed", "max-attempts");
      return true;
    }
    this.#nextRetryAt = this.#now() + retryDelay(this.#attempt, this.#baseDelayMs, this.#maxDelayMs);
    this.#move("reconnecting", reason);
    return true;
  }

  retry(): boolean {
    if (this.#disposed) return false;
    if (this.#state !== "reconnecting") return false;
    if (this.#nextRetryAt !== null && this.#now() < this.#nextRetryAt) return false;
    this.#nextRetryAt = null;
    this.#move("connecting", "retry");
    return true;
  }

  fail(): boolean {
    if (this.#disposed) return false;
    if (this.#state === "idle" || this.#state === "closed" || this.#state === "failed") return false;
    this.#nextRetryAt = null;
    this.#move("failed", "fatal");
    return true;
  }

  close(): boolean {
    if (this.#disposed) return false;
    if (this.#state === "idle" || this.#state === "closed") return false;
    this.#nextRetryAt = null;
    this.#suspended = false;
    this.#move("closed", "close");
    return true;
  }

  hide(): boolean {
    if (this.#disposed || this.#suspended) return false;
    if (this.#state !== "open" && this.#state !== "connecting" && this.#state !== "reconnecting") {
      return false;
    }
    this.#suspended = true;
    this.#nextRetryAt = null;
    this.#move("closed", "page-hidden");
    return true;
  }

  show(): boolean {
    if (this.#disposed || !this.#suspended) return false;
    if (this.#state !== "closed") return false;
    this.#suspended = false;
    this.#attempt = 0;
    this.#move("connecting", "page-visible");
    return true;
  }

  dispose(): void {
    if (this.#disposed) return;
    if (this.#state !== "closed") {
      this.#nextRetryAt = null;
      this.#suspended = false;
      this.#move("closed", "dispose");
    }
    this.#disposed = true;
    this.#listeners.clear();
  }

  #move(to: ConnectionState, reason: TransitionReason): void {
    const from = this.#state;
    this.#state = to;
    const event: TransitionEvent = {
      from,
      to,
      reason,
      at: this.#now(),
      attempt: this.#attempt,
    };
    for (const listener of this.#listeners) listener(event);
  }
}

$ node --test --test-concurrency=1 test/realtime/connection.test.ts
✔ idle 只能先进入 connecting，再进入 open (0.946ms)
✔ 非法迁移返回 false，不产生事件也不改变状态 (0.466791ms)
✔ 连接丢失进入 reconnecting，并按尝试次数安排下一次重试时间 (0.127291ms)
✔ 未到重试时间不能 retry，假时钟推进后可以 (0.07825ms)
✔ 连续丢失时退避翻倍，超过上限进入 failed，之后可以从 failed 重启 (0.276167ms)
✔ 页面切后台关闭连接并标记 suspended，回到前台重新 connecting (0.099167ms)
✔ dispose 关闭连接、清理监听，并且是终态和幂等的 (0.095ms)
✔ 取消订阅后不再收到迁移事件，但状态机继续工作 (0.071541ms)
ℹ tests 8
ℹ suites 0
ℹ pass 8
ℹ fail 0
ℹ cancelled 0
ℹ skipped 0
ℹ todo 0
ℹ duration_ms 126.213708