Universal Message

English | 中文

一个强大的通用 JavaScript/TypeScript 消息通信库，支持 Worker 线程、WebSocket 和其他通信渠道之间的无缝通信。

特性

• 🚀 通用性: 支持多种通信渠道（Worker 线程、WebSocket、MessagePort 等）
• 🔌 插件系统: 可扩展的插件架构，支持自定义消息编码/解码，内置复杂类型支持
• 🎯 类型安全: 完整的 TypeScript 支持，深度对象访问智能路径补全（obj.prop.array.0.field）
• 📦 零依赖: 轻量级，无外部依赖
• 🔄 双向通信: 支持客户端-服务器和点对点通信模式
• ⚡ 基于 Promise: 现代的 async/await API 和 Promise 支持，带超时处理
• 🏗️ 远程类: 完整的远程类实例化和方法调用支持
• 📡 共享对象: 跨环境对象共享和同步
• 🛡️ 错误处理: 全面的错误处理，保留堆栈跟踪信息
• 🎨 自定义序列化: 高级 JSON 序列化，支持 Date、Map、Set、RegExp、Buffer、BigInt 等

安装

npm install universal-message

yarn add universal-message

pnpm add universal-message

快速开始

基本用法

import { MessageServer, MessageClient } from 'universal-message';

// 服务器端
const server = new MessageServer({
  on: (callback) => {
    // 设置消息监听器（如 WebSocket onmessage, Worker onmessage）
    worker.onmessage = callback;
  },
  send: (data) => {
    // 设置消息发送器（如 WebSocket send, Worker postMessage）
    worker.postMessage(data);
  }
});

// 客户端
const client = new MessageClient({
  on: (callback) => {
    // 设置消息监听器
    self.onmessage = callback;
  },
  send: (data) => {
    // 设置消息发送器
    self.postMessage(data);
  }
});

Worker 线程示例

// main.ts
import { Worker } from 'worker_threads';
import { MessageServer } from 'universal-message';

const worker = new Worker('./worker.js');

const server = new MessageServer({
  on: (callback) => worker.on('message', callback),
  send: (data) => worker.postMessage(data)
});

// 发送消息并等待响应
const result = await server.send('calculate', { a: 10, b: 20 });
console.log(result); // 30

// 创建远程类实例
const calc = await server.newClass('Calculator');
const sum = await calc.call('add', 10, 20); // 30

// worker.js
import { parentPort } from 'worker_threads';
import { MessageClient } from 'universal-message';

const client = new MessageClient({
  on: (callback) => parentPort?.on('message', callback),
  send: (data) => parentPort?.postMessage(data)
});

// 注册类以供远程访问
class Calculator {
  add(a, b) { return a + b; }
  multiply(a, b) { return a * b; }
}
client.addClass(Calculator);

// 监听消息
client.on('calculate', ({ a, b }) => {
  return a + b;
});

WebSocket 示例

// 服务器端
import { MessageServer } from 'universal-message';

const ws = new WebSocket('ws://localhost:8080');

const server = new MessageServer({
  on: (callback) => {
    ws.onmessage = (event) => callback(JSON.parse(event.data));
  },
  send: (data) => {
    ws.send(JSON.stringify(data));
  }
});

// 使用类型安全
const user = await server.send<User>('getUser', { id: 123 });

// 客户端（类似的 WebSocket 设置）

MessagePort 示例

// 创建 MessageChannel
const channel = new MessageChannel();

// Port 1 - 服务器
const server = new MessageServer({
  on: (callback) => channel.port1.onmessage = callback,
  send: (data) => channel.port1.postMessage(data)
});

// Port 2 - 客户端
const client = new MessageClient({
  on: (callback) => channel.port2.onmessage = callback,
  send: (data) => channel.port2.postMessage(data)
});

API 参考

MessageServer

具有增强远程功能的服务器端消息处理器。

方法

• send<T>(key: string, data?: any, timeout?: number): Promise<T> - 发送消息并等待响应
• on(key: string, callback: Function) - 监听消息
• hasOn(key: string): boolean - 检查消息监听器是否存在
• remove(key: string) - 移除消息监听器
• removeAll(key: string) - 移除监听器并通知另一端
• emit<T>(key: string, data?: any, callback?: Function): Promise<T> - send 的别名
• newClass<T>(target: string | Class, constructorArgs?: any[]): Promise<RemoteClass<T>> - 创建远程类实例
• staticClass<T>(target: string | Class): StaticMethods<T> - 访问静态类方法
• addShared<T>(key: string, value?: T): SharedObject<T> - 创建或访问共享对象

MessageClient

具有类注册功能的客户端消息处理器。

方法

• send<T>(key: string, data?: any, timeout?: number): Promise<T> - 发送消息并等待响应
• on(key: string, callback: Function) - 监听消息
• hasOn(key: string): boolean - 检查消息监听器是否存在
• remove(key: string) - 移除消息监听器
• addClass<T>(className: string, classConstructor: T) - 注册类以供远程访问
• addShared<T>(key: string, value?: T): SharedObject<T> - 创建或访问共享对象

MessageShared

为服务器和客户端提供共享对象功能的基类。

共享对象

// 跨环境创建或访问共享对象
const sharedCounter = client.shared('counter', 0);

// 获取值
const count = await sharedCounter.value(); // 0

// 设置值
await sharedCounter.set('', 5);

// 在共享对象上调用方法
const sharedArray = client.shared('myArray', [1, 2, 3]);
await sharedArray.call('push', 4); // [1, 2, 3, 4]

// 删除属性
await sharedArray.del('0'); // 移除第一个元素

插件系统

Universal Message 支持强大的插件系统来扩展功能：

import { MessagePlugin } from 'universal-message';

// 自定义插件示例
class LoggingPlugin extends MessagePlugin {
  name = 'logging-plugin';
  
  async onSend(data) {
    console.log('发送:', data);
    return data;
  }
  
  async onMessage(data) {
    console.log('接收:', data);
    return data;
  }
}

// 添加插件
server.addPlugin(new LoggingPlugin());

// 移除插件
server.removePlugin('logging-plugin');

内置插件

• MessageEcoderPlugin: 处理复杂数据类型的编码/解码，包括：
  • ECoderAbortController: AbortController 序列化，带信号传播
  • ECoderClassTransformer: 使用 class-transformer 自动类实例序列化

自定义编码器

为特定数据类型创建自定义编码器：

import { MessageEcoder } from 'universal-message';

const CustomEncoder: MessageEcoder<MyClass> = {
  target: (data) => data instanceof MyClass,
  key: 'MyClass',
  encode: (data, tool) => {
    // 自定义序列化逻辑
    return { customData: data.serialize() };
  },
  decode: (data, tool) => {
    // 自定义反序列化逻辑
    return MyClass.deserialize(data.customData);
  },
  update: (newValue, target, tool) => {
    // 如果需要，处理更新
    target.update(newValue);
  }
};

远程类使用

Universal Message 提供强大的远程类实例化和方法调用功能：

// 定义具有复杂功能的类
class DatabaseManager {
  private connections = new Map();
  
  constructor(config: DbConfig) {
    this.config = config;
  }
  
  async connect(database: string) {
    // 连接逻辑
    return `Connected to ${database}`;
  }
  
  async query(sql: string, params: any[]) {
    // 查询逻辑
    return { rows: [], count: 0 };
  }
  
  static getDrivers() {
    return ['mysql', 'postgres', 'sqlite'];
  }
}

// 客户端 - 注册类
client.addClass('DatabaseManager', DatabaseManager);

// 服务器端 - 使用具有完整类型安全的远程类
const db = await server.newClass<DatabaseManager>('DatabaseManager', [{ host: 'localhost' }]);

// 调用实例方法
const result = await db.call('connect', 'myapp');
const rows = await db.call('query', 'SELECT * FROM users', []);

// 访问静态方法
const drivers = await db.static.call('getDrivers');

// 类型安全的深度属性访问
const config = await db.get('config');
const host = await db.get('config.host'); // 类型安全的深度访问
await db.set('config.timeout', 5000);

高级类型安全

使用 TypeScript，你可以获得深度对象路径的智能自动补全：

class User {
  profile = {
    personal: {
      name: 'John',
      age: 30,
      addresses: [
        { street: '123 Main St', city: 'NYC' }
      ]
    },
    settings: {
      theme: 'dark',
      notifications: { email: true, push: false }
    }
  };
  
  updateProfile(data: any) { /* ... */ }
}

const user = await server.newClass<User>('User');

// 所有这些路径都有完整的类型检查和自动补全：
const name = await user.get('profile.personal.name');           // string
const age = await user.get('profile.personal.age');             // number  
const street = await user.get('profile.personal.addresses.0.street'); // string
const theme = await user.get('profile.settings.theme');         // string
const emailNotif = await user.get('profile.settings.notifications.email'); // boolean

// 带参数类型检查的方法调用
await user.call('updateProfile', { name: 'Jane' });

高级特性

复杂数据类型支持

Universal Message 包含高级 JSON 序列化，支持：

// 日期、Map、Set、正则表达式、错误、URL
const complexData = {
  timestamp: new Date(),
  userMap: new Map([['user1', { id: 1 }], ['user2', { id: 2 }]]),
  tags: new Set(['admin', 'user']),
  pattern: /^[a-z]+$/i,
  config: new URL('https://api.example.com'),
  lastError: new Error('连接失败')
};

await server.send('processData', complexData);
// 所有类型都会自动正确序列化和反序列化

// BigInt 支持
const bigNumber = BigInt('12345678901234567890');
await server.send('calculateLarge', bigNumber);

// Buffer/ArrayBuffer 支持（Node.js 和 Web）
const buffer = Buffer.from('Hello World');
const arrayBuffer = new ArrayBuffer(16);
await server.send('processBytes', { buffer, arrayBuffer });

AbortController 支持

内置可取消操作支持：

const controller = new AbortController();

// 发送带取消控制器的操作
const promise = server.send('longRunningTask', { 
  signal: controller.signal 
});

// 5 秒后取消
setTimeout(() => controller.abort(), 5000);

try {
  const result = await promise;
} catch (error) {
  if (error.name === 'AbortError') {
    console.log('操作已取消');
  }
}

Class-Transformer 集成

自动类实例序列化：

import { Transform, Type } from 'class-transformer';

class Address {
  street: string;
  city: string;
}

class Person {
  name: string;
  
  @Type(() => Date)
  birthDate: Date;
  
  @Type(() => Address)
  address: Address;
}

// 实例会自动序列化，保留装饰器
const person = new Person();
person.name = 'John';
person.birthDate = new Date();
person.address = new Address();

await server.send('processPerson', person);
// Person 实例在另一端正确重构

错误处理与堆栈跟踪

try {
  const result = await server.send('riskyOperation', data, 10000); // 10秒超时
} catch (error) {
  if (error.code === 'TIMEOUT') {
    console.log('操作超时');
  } else {
    console.log('错误:', error.message);
    console.log('堆栈:', error.stack); // 跨环境保留
  }
}

贡献

欢迎贡献！请随时提交 Pull Request。

许可证

MIT 许可证。详情请查看 LICENSE。