# @cf-platform/mq

> 消息队列库

Latest version **2.0.2** (published 2026-08-14) · ISC license · 0 weekly downloads

## Install

```sh
npm install @cf-platform/mq
pnpm add @cf-platform/mq
yarn add @cf-platform/mq
bun add @cf-platform/mq
```

## Health

**Score 65/100 (B)** — status: active.

Positive: has types; esm support; no vulnerabilities; recently updated; high maintenance score.

Warnings: low downloads.

## Facts

| | |
|---|---|
| Version | 2.0.2 |
| Published | 2026-08-14 |
| First published | 2024-04-08 |
| Weekly downloads | 0 |
| License | ISC |
| TypeScript types | bundled |
| Module format | ESM + CommonJS |
| Dependencies | 2 |
| Unpacked size | 44.2 KB |
| Known vulnerabilities | 0 |
| Install scripts | no |
| Author | maomao |
| Maintainers | object_fu |
| Keywords | mq |

## Links

- npm: https://www.npmjs.com/package/@cf-platform/mq
- npm.io page: https://npm.io/package/@cf-platform/mq

## Dependencies (2)

- [@cf-platform/cim](https://npm.io/package/@cf-platform/cim.md) ^2.0.0
- [@mm-custom/method](https://npm.io/package/@mm-custom/method.md) ^2.2.11

## Recent versions

- 2.0.2 (latest) — 2026-08-14
- 2.0.1 — 2026-07-03
- 1.0.4 — 2024-04-28
- 1.0.3 — 2024-04-22
- 1.0.2 — 2024-04-09
- 1.0.1 — 2024-04-08
- 1.0.0 — 2024-04-08

## README

# @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方案

## 安装

```bash
npm install @cf-platform/mq
```

或

```bash
yarn add @cf-platform/mq
```

## 快速开始

### 使用CIM方案

```typescript
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方案

```typescript
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();
```

### 使用通用工厂方法

```typescript
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()` 订阅，不再通过构造选项传入回调。

#### 构造函数

```typescript
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（通用配置）

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

#### WebSocketOptions（WebSocket 特定配置）

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

#### CimMqOptions（CIM 特定配置）

```typescript
interface CimMqOptions extends CimAdapterOptions {
  technology: MqTechnology.CIM;
  // 继承自 CimAdapterOptions：
  //   url: string                           // WebSocket 服务器地址
  //   protocols?: string | string[]          // 协议列表
  //   reconnectInterval?: number             // 自动重连间隔（毫秒），默认 3000
  //   maxReconnectAttempts?: number          // 最大重连次数，默认 5
  //   enableHeartbeat?: boolean              // 是否启用心跳检测，默认 false
  //   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` | 自动检测 | | 设备名称 |

### 连接状态

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

### 技术方案类型

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

## 使用示例

### 完整的 WebSocket 使用示例

```typescript
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` 以获得事件能力。

```typescript
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 等）

## 开发

```bash
# 安装依赖
npm install

# 构建
npm run build
```

## 版本

当前版本：2.0.2

## 作者

maomao

## 许可证

ISC

## 更新日志

### v2.0.2 (修复)

- 🐛 **CimAdapter 心跳检测默认关闭** — `enableHeartbeat` 默认值从 `true` 改为 `false`。CIM 心跳检测依赖于服务端定时发送 PING，若服务端未配置 PING 或间隔过长（>35s），客户端会被动断开。默认关闭可避免未显式配置时出现非预期断连

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

---
_Source: https://npm.io/package/@cf-platform/mq · Machine-readable twin of the npm.io package page. Health data is recomputed on every publish._