@cf-platform/mq

消息队列库，提供统一的MQ接口，支持多种技术方案。

特性

• 🎯 统一接口：提供统一的MQ接口，支持多种技术方案
• 🔌 多种方案：支持CIM、WebSocket，预留RabbitMQ、Kafka、Redis等方案
• 🏭 工厂模式：采用工厂模式，便于扩展新的MQ技术方案
• 🔌 适配器模式：采用适配器模式，提供统一的API接口
• 💓 心跳检测：支持心跳检测机制，确保连接稳定性
• 🔄 自动重连：支持自动重连机制，提高连接可靠性
• 📡 事件驱动：全链路基于 EventManager，统一的事件订阅/发布能力
• 📦 TypeScript：完整的TypeScript类型定义

技术方案

当前支持的技术方案：

• CIM：基于CIM的消息队列方案
• WebSocket：原生WebSocket实现

预留的技术方案：

• RabbitMQ：RabbitMQ消息队列
• Kafka：Kafka消息队列
• Redis：Redis Pub/Sub方案

安装

npm install @cf-platform/mq

或

yarn add @cf-platform/mq

快速开始

使用CIM方案

import { createCimMq } from '@cf-platform/mq';

const mq = createCimMq({
  url: 'ws://localhost:8080',
  loginParams: {
    uid: 'user123',
    appVersion: '1.0.0',
    deviceId: 'device-001'
  }
});

// 通过事件订阅鉴权结果
mq.on('connect', (reply) => {
  console.log('CIM 认证成功:', reply);
});

// 通过事件订阅消息
mq.on('message', (msg) => {
  console.log('收到消息:', msg);
});

mq.connect();

使用WebSocket方案

import { createWebSocketMq } from '@cf-platform/mq';

const mq = createWebSocketMq({
  url: 'ws://localhost:8080',
  heartbeatInterval: 30000,
  reconnectInterval: 3000,
  maxReconnectAttempts: 5
});

mq.on('message', (event) => {
  console.log('收到消息:', event.data);
});

mq.connect();

使用通用工厂方法

import { createMq, MqTechnology } from '@cf-platform/mq';

const mq = createMq({
  technology: MqTechnology.WEBSOCKET,
  url: 'ws://localhost:8080'
});

mq.connect();

API 文档

Mq 类

Mq 将事件方法全部代理到适配器的 EventManager（来自 @mm-custom/method），所有事件均通过 on()/off() 订阅，不再通过构造选项传入回调。

构造函数

constructor(options: MqOptions)

连接管理方法

• connect(): void — 建立连接，复用适配器重新建立底层连接
• reconnect(): void — 重新连接，重置重连计数后立即尝试连接
• close(): void — 关闭连接并清理定时器（心跳/重连/连接超时），事件订阅保留以便再次连接复用

消息方法

• sendMessage(data: any): boolean — 发送消息（接受任意类型，适配器负责序列化）

状态查询方法

• getConnectionState(): ConnectionState — 获取连接状态
• isConnected(): boolean — 检查是否已连接
• isConnecting(): boolean — 检查是否正在连接中
• isClosing(): boolean — 检查是否正在关闭中
• isClosed(): boolean — 检查是否已关闭
• getAdapterType(): string — 获取当前使用的适配器类型

事件方法（代理到适配器的 EventManager）

• on(eventName: string, listener: Listener, options?: ListenerOptions): () => void — 订阅事件，返回取消订阅函数
• once(eventName: string, listener: Listener): () => void — 订阅一次性事件
• off(eventName: string, listenerOrId?: string | Listener): boolean — 取消订阅
• emit(eventName: string, data?: any, options?: EmitterOptions): Promise<boolean> — 异步触发事件
• emitSync(eventName: string, data?: any): boolean — 同步触发事件
• clear(eventName?: string): this — 清除事件监听器
• listenerCount(eventName?: string): number — 获取监听器数量
• eventNames(): string[] — 获取所有事件名称
• setMaxListeners(n: number): this — 设置最大监听器数
• hasListeners(eventName: string): boolean — 是否有监听器

内置事件

| 事件名 | 触发时机 | 数据 | |--------|---------|------| | connect | 连接建立成功（CIM: CLIENT_BIND 认证通过后） | ReplyMessage (CIM) / Event (WebSocket) | | disconnect | 连接断开 | { code: number, reason: string } | | stateChange | 连接状态变化 | ConnectionState | | message | 收到服务器消息 | ReceivedMessage (CIM) / MessageEvent (WebSocket) | | error | 连接发生错误 | Event | | reconnectAttempt | 自动重连尝试 | { attempt: number, max: number } | | reconnectFailed | 重连次数耗尽 | { max: number } |

MqFactory 类

静态方法

• createAdapter(options: MqOptions): BaseAdapter — 根据配置创建适配器
• registerAdapter(technology: string, factory: (options: any) => BaseAdapter): void — 注册自定义适配器
• hasAdapter(technology: string): boolean — 检查适配器是否已注册
• getRegisteredTypes(): string[] — 获取所有已注册的适配器类型

配置选项

CommonMqOptions（通用配置）

interface CommonMqOptions {
  url: string;                     // WebSocket 服务器地址
  protocols?: string | string[];   // 协议列表
  reconnectInterval?: number;      // 自动重连间隔（毫秒），0 表示不自动重连，默认 3000ms
  maxReconnectAttempts?: number;   // 最大重连次数，默认 5
  enableHeartbeat?: boolean;       // 是否启用心跳检测，默认 true
  heartbeatTimeout?: number;       // 心跳检测超时时间（毫秒），默认 35000ms
  connectionTimeout?: number;      // 连接超时时间（毫秒），默认 10000ms
}

WebSocketOptions（WebSocket 特定配置）

interface WebSocketOptions extends CommonMqOptions {
  heartbeatInterval?: number;  // 心跳间隔（毫秒），默认 30000ms
}

CimMqOptions（CIM 特定配置）

interface CimMqOptions extends CimAdapterOptions {
  technology: MqTechnology.CIM;
  // 继承自 CimAdapterOptions：
  //   url: string                           // WebSocket 服务器地址
  //   protocols?: string | string[]          // 协议列表
  //   reconnectInterval?: number             // 自动重连间隔（毫秒），默认 3000
  //   maxReconnectAttempts?: number          // 最大重连次数，默认 5
  //   enableHeartbeat?: boolean              // 是否启用心跳检测，默认 true
  //   heartbeatTimeout?: number              // 心跳超时（毫秒），默认 35000
  //   connectionTimeout?: number             // 连接超时（毫秒），默认 10000
  //   loginParams?: CimLoginParams           // CIM 登录认证参数
}

CimLoginParams

| 字段 | 类型 | 默认值 | 必填 | 描述 | |------|------|--------|------|------| | uid | string | - | ✓ | 用户唯一标识 | | appVersion | string | - | ✓ | 应用版本号 | | deviceId | string | - | ✓ | 设备唯一标识 | | channel | string | 'web' | | 客户端通道类型 | | osVersion | string | 自动检测 | | 操作系统版本 | | packageName | string | 'com.farsunset.cim' | | 应用包名 | | language | string | 自动检测 | | 系统语言 | | deviceName | string | 自动检测 | | 设备名称 |

连接状态

enum ConnectionState {
  CONNECTING = 0,  // 连接中
  OPEN = 1,        // 已连接
  CLOSING = 2,     // 关闭中
  CLOSED = 3       // 已关闭
}

技术方案类型

enum MqTechnology {
  CIM = 'cim',           // CIM 方案
  WEBSOCKET = 'websocket', // WebSocket 方案
  RABBITMQ = 'rabbitmq', // RabbitMQ 方案（预留）
  KAFKA = 'kafka',       // Kafka 方案（预留）
  REDIS = 'redis'        // Redis 方案（预留）
}

使用示例

完整的 WebSocket 使用示例

import { createWebSocketMq } from '@cf-platform/mq';

const mq = createWebSocketMq({
  url: 'wss://example.com/ws',
  protocols: ['chat', 'notification'],
  heartbeatInterval: 30000,
  reconnectInterval: 5000,
  maxReconnectAttempts: 10,
  enableHeartbeat: true,
  heartbeatTimeout: 35000
});

// 连接生命周期
mq.on('connect', () => console.log('连接已建立'));
mq.on('disconnect', () => console.log('连接已断开'));

// 连接状态追踪
mq.on('stateChange', (state) => console.log('连接状态变化:', state));

// 接收消息
mq.on('message', (event) => {
  const data = JSON.parse(event.data);
  console.log('收到消息:', data);
});

// 错误处理
mq.on('error', (event) => console.error('连接错误:', event));

// 重连追踪
mq.on('reconnectAttempt', ({ attempt, max }) => {
  console.log(`重连尝试 ${attempt}/${max}`);
});
mq.on('reconnectFailed', ({ max }) => {
  console.error(`重连失败，已达到最大尝试次数 ${max}`);
});

// 建立连接
mq.connect();

// 发送消息
mq.sendMessage({ type: 'chat', content: 'Hello World' });

// 检查连接状态
if (mq.isConnected()) {
  console.log('已连接');
}

// 关闭连接
mq.close();

// 可以再次连接（会自动重建适配器）
mq.connect();

扩展自定义适配器

BaseAdapter 继承自 EventManager，自定义适配器也需要继承 EventManager 以获得事件能力。

import { MqFactory, BaseAdapter, ConnectionState } from '@cf-platform/mq';
import { EventManager } from '@mm-custom/method';

// 自定义适配器：继承 EventManager 并实现 BaseAdapter
class CustomAdapter extends EventManager implements BaseAdapter {
  connect(): void { /* ... */ }
  reconnect(): void { /* ... */ }
  close(): void { /* ... */ }
  sendMessage(data: any): boolean { /* ... */ return true; }
  getConnectionState(): ConnectionState { return ConnectionState.OPEN; }
  isConnected(): boolean { return true; }
  isConnecting(): boolean { return false; }
  isClosing(): boolean { return false; }
  isClosed(): boolean { return false; }
  getType(): string { return 'Custom'; }
}

// 注册自定义适配器
MqFactory.registerAdapter('custom', (options) => new CustomAdapter());

// 检查是否已注册
console.log(MqFactory.hasAdapter('custom')); // true
console.log(MqFactory.getRegisteredTypes()); // ['cim', 'rabbitmq', 'kafka', 'redis', 'websocket', 'custom']

架构设计

本库采用以下设计模式：

1. 工厂模式：MqFactory 负责创建不同技术方案的适配器，支持动态注册自定义适配器
2. 适配器模式：BaseAdapter 继承 EventManager，为所有适配器统一事件和操作接口
3. 策略模式：通过配置选择不同的技术方案
4. 事件驱动：适配器继承 EventManager 发出事件，Mq 通过组合模式代理所有事件方法到适配器

分层架构

┌──────────────────────────────────────────┐
│  Mq (门面)                                │
│  • 组合模式，所有事件代理到适配器            │
│  • 统一的 connect/sendMessage/close API    │
├──────────────────────────────────────────┤
│  CimAdapter / WebSocketAdapter (业务层)    │
│  • 心跳超时检测 & 定时管理                  │
│  • 自动重连策略 (attempts / backoff)        │
│  • 连接超时检测                             │
│  • 认证流程 (CimAdapter: CLIENT_BIND 等)    │
│  • 状态变化通知 (stateChange)               │
│  • 继承 EventManager，发出内置事件           │
├──────────────────────────────────────────┤
│  CIM / 原生 WebSocket (传输层)             │
│  • WebSocket 连接生命周期                   │
│  • Protobuf 编解码 (CIM) / JSON (原生)      │
│  • PING→PONG 自动响应 (CIM)                │
└──────────────────────────────────────────┘

事件流转

CIM.onPing / WebSocket.onmessage
  │
  ▼
CimAdapter / WebSocketAdapter (EventManager)
  │  emitSync('stateChange'/'message'/'error'/'reconnectAttempt'/'reconnectFailed'/'connect'/'disconnect')
  │
  ▼  Mq 通过组合代理所有 on/off/emit 到适配器
Mq
  │  用户通过 mq.on() 订阅
  ▼
用户

依赖项

• @cf-platform/cim: CIM 协议传输层（Protobuf 编解码 + WebSocket）
• @mm-custom/method: 工具方法库（提供 EventManager、browserVersion 等）

开发

# 安装依赖
npm install

# 构建
npm run build

版本

当前版本：2.0.1

作者

maomao

许可证

ISC

更新日志

v2.0.1 (修复)

• 🐛 修复 CimAdapter 残留连接定时器打断重连 — 连接在认证阶段断开时，旧的 _connectionTimer 未被清理，会在重连过程中误触发 close() 中断新连接。现 _doConnect 开头与 onClose 均清理连接定时器
• 🐛 修复 WebSocketAdapter 旧连接泄漏 — _doConnect 原先直接覆盖 this._ws，旧连接未关闭、回调未解绑，会产生孤儿连接与重复事件。现先解绑回调并关闭旧连接
• 🐛 修复 WebSocketAdapter.reconnect() 不重置重连计数 — 重连次数耗尽后手动 reconnect() 即使连上，下次掉线也会立即判失败。现已重置 _reconnectAttempts
• 🔧 统一 WebSocketAdapter 重连逻辑 — _handleConnectionError 不再自行累加计数与排程，统一交由 _handleReconnect 处理，两条路径重试次数一致（均为 max）
• 🧹 去除 CimAdapter _doConnect 末尾多余的心跳启动（onOpen 已启动）
• 🧹 清理 protocols 的 '' 哨兵来回转换、合并重复 @mm-custom/method import、MqFactory 死代码回退

v2.0.0

• 🔥 CimAdapter 重构：接管完整的心跳检测、自动重连、认证流程（CLIENT_BIND / HANDSHAKE）
• 🔥 CimMqOptions 改为 extends CimAdapterOptions（含 loginParams）
• 🆕 新增 CimLoginParams 类型，替代旧的 LoginParams（已从 @cf-platform/cim 移除）
• 🆕 CIM 方案支持通过 loginParams 配置认证参数
• 📝 更新架构文档，反映新的分层设计（传输层 / 业务层分离）
• ⬆️ 依赖 @cf-platform/cim v2.0.0（传输层重构，移除 gprotobuf 依赖）

v1.0.5

• 支持CIM和WebSocket技术方案
• 实现工厂模式和适配器模式
• 支持心跳检测和自动重连
• 完整的TypeScript类型定义
• Mq 类继承 EventManager，全链路事件驱动
• BaseAdapter 继承 EventManager，CimAdapter 同步继承
• MqFactory 实现真正的适配器注册机制（registerAdapter/hasAdapter/getRegisteredTypes）
• BaseAdapter 新增 getType() 方法
• Mq 新增 isClosing() 方法
• close() 后可再次 connect()/reconnect()
• close() 自动清理适配器事件订阅，防止监听器泄漏
• sendMessage 接受任意类型数据
• 配置选项全面事件化：移除 onOpen/onMessage/onClose/onError 回调及 customEvents
• 新增事件：message、error（Mq 级别）
• 修复 WebSocket 重连计数器无限重置问题（拆分 connect/doConnect）
• 修复手动 reconnect() 不必要的延迟（直接调用 doConnect）
• 修复 CimAdapter getConnectionState 不安全类型强转（增加范围校验）
• 目录重命名 adpter → adapter（修正拼写）
• 移除浏览器绑定（window.setInterval → setInterval）
• 补全 CONNECTING 状态通知，完善状态流转