如何为游戏网络模块封装返回RxJS Observable的sendPacket方法

要实现sendPacket返回Observable的效果，核心是解决请求与响应的匹配问题：

• 给每个发出去的请求包分配唯一ID，服务端返回响应时必须携带该ID
• 用映射表存储所有待响应请求对应的观察者对象，收到响应后根据ID找到对应观察者推送数据

import { Observable } from 'rxjs';
// 如果你用RxJS 6及以上版本，需要用操作符的话从rxjs/operators导入
import { timeout } from 'rxjs/operators';

const network = {
  // 存储未收到响应的请求观察者：key为请求ID，value为Observable的观察者对象
  _pendingRequests: new Map(),
  // 自增ID生成器，保证每个请求ID唯一
  _reqIdSeq: 1,
  // 你原来的TCP客户端实例
  _tcpClient: /* 原有TCP客户端实例 */,

  sendPacket(packet) {
    // 生成唯一请求ID，绑定到请求包上
    const reqId = this._reqIdSeq++;
    packet.reqId = reqId;

    // 返回Observable对象
    return new Observable((observer) => {
      // 把当前观察者存入待处理映射
      this._pendingRequests.set(reqId, observer);

      // 发送原始请求
      this._tcpClient.sendToServer(packet);

      // 退订清理逻辑：用户取消订阅、请求完成/出错时自动执行
      return () => {
        this._pendingRequests.delete(reqId);
        // 可选扩展：如果需要主动取消请求，这里可以加发取消指令的逻辑
      };
    });
  },

  onReceived(pkg) {
    const reqId = pkg.reqId;
    // 没有匹配的待处理请求，判定为服务端主动推送消息，单独处理即可
    if (!reqId || !this._pendingRequests.has(reqId)) {
      // 这里加你原有全局推送消息的处理逻辑
      return;
    }

    // 拿到对应请求的观察者，推送响应
    const observer = this._pendingRequests.get(reqId);
    observer.next(pkg);
    // 单次请求对应单次响应的场景直接结束流，如果是多响应场景（如订阅流）可以去掉这行
    observer.complete();
    // 清理待处理映射
    this._pendingRequests.delete(reqId);
  }
}

和你期望的用法完全一致，还可以直接用RxJS的操作符扩展能力：

// 基础用法
network.sendPacket(myPacket).subscribe(
  response => {
    // 处理正常响应
  },
  error => {
    // 处理错误
  }
);

// 扩展10秒超时逻辑
network.sendPacket(myPacket)
  .pipe(timeout(10000))
  .subscribe(response => /* 处理响应 */);

• 前后端需要提前约定协议规则：所有响应包必须携带对应请求的reqId字段才能正确匹配
• 对于服务端主动推送的无对应请求的消息（如全服通知、其他玩家交互消息），不需要携带reqId，在onReceived中单独处理即可
• 长时间未响应的请求建议加超时逻辑，避免_pendingRequests里堆积无效条目导致内存泄漏

内容的提问来源于stack exchange，提问作者TienThanh