npm.io
2.0.1 • Published 1 month ago

@cf-platform/mq

Licence
ISC
Version
2.0.1
Deps
2
Size
43 kB
Vulns
0
Weekly
0
SummaryDependencyVersions

@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 状态通知，完善状态流转

Keywords

mq