命令

具有插件支持的轻量级、 类型安全的命令模式执行 :

• 零依赖关系
• 宽插件系统
• 🔄 Async/Stream/发电机支持
• QQ 完整类型脚本类型推论
• * 了解背景的处决

安装

npm install @collidor/command

特征

• 类型安全的命令执行 - 返回类型自动匹配处理器
• 插件架构 - 将输出转换为承诺、流或自定义类型
• 灵活上下文 - 通过上下文对象执行状态
• 简单的 API - 只有两种方法: 注册和执行
• 运算器支持 -- -- 发电机和合成流的内置处理

基本用途

使用类继承

import { Command, CommandBus } from "@collidor/command";

// 1. Define command
class CreateUser extends Command<{name: string}, { id: string }> {}

// 2. Create bus
const bus = new CommandBus();

// 3. Register handler
bus.register(CreateUser, (command, context) => ({
  id: Math.random().toString(36).substr(2, 9),
}));

// 4. Execute (type inferred as { id: string })
const user = bus.execute(new CreateUser({ name: "Alice" }));

使用 createCommand (邦德勒/无害安全)

当捆绑用于生产时,JavaScript分解器常常会操纵类名(例如. class CreateUser 变成 class a) (中文(简体) ). 由于总线、 DI 提供者和网络插件(httpPlugin, (中文(简体) ), portChannelPlugin, (中文(简体) ), windowCustomEventPlugin使用其构建器名称的路由命令, createCommand 提供一种类型安全的方法,以固定、不可忽略的名称定义命令:

import { createCommand, CommandBus } from "@collidor/command";

// Define command with explicit types and name: createCommand<PayloadType, ReturnType>("CommandName")
export const CreateUser = createCommand<{ name: string }, { id: string }>("CreateUser");

// Void payload command
export const PingCommand = createCommand<void, string>("PingCommand");

const bus = new CommandBus();

bus.register(CreateUser, (command) => ({
  id: "user_123",
}));

bus.register(PingCommand, () => "pong");

const user = bus.execute(new CreateUser({ name: "Alice" }));
const pong = bus.execute(new PingCommand());

串流方法

对于命令处理器产生多个同步事件的情景,您现在可以注册返回 AsyncIterable 的 Async 流处理器. 这样您就可以使用一个自然的等待程序来消耗命令流.

正在注册 Async Stream 处理器

使用寄存器StreamAsync来注册返回Async 转发器的处理器. 这个处理器可以进行同步的工作,并随着时间的推移产生出事件.

import { CommandBus, Command } from "@collidor/command";

// Define a command
class UpdateUser extends Command<number, { status: string }> {}

// Create a CommandBus instance
const commandBus = new CommandBus();

// Register an async stream handler
commandBus.registerStreamAsync(UpdateUser, async function* (command, context) {
  for (let i = 0; i < command.data; i++) {
    // Simulate asynchronous delay
    await new Promise((resolve) => setTimeout(resolve, 100));
    yield { status: `update-${i}` };
  }
});

正在消耗同步流

StreamAsync方法返回一个 Async 指示器. 您可以轻松地消耗流 等待... 循环:

(async () => {
  // Execute the command and consume the async stream
  for await (const result of commandBus.streamAsync(new UpdateUser(5))) {
    console.log(result);
  }
  // Expected output:
  // { status: 'update-0' }
  // { status: 'update-1' }
  // { status: 'update-2' }
  // { status: 'update-3' }
  // { status: 'update-4' }
})();

插件系统

同步操作

const asyncBus = new CommandBus({
  plugin: (command, ctx, handler) => {
    return Promise.resolve(handler?.(command, ctx));
  }
});

// Returns Promise<{ id: string }>
const futureUser = asyncBus.execute(new CreateUser());

流程处理

const streamBus = new CommandBus({
  plugin: async function* (command, ctx, handler) {
    yield await handler?.(command, ctx);
    yield await handler?.(command, ctx);
  }
});

// Returns AsyncIterable<{ id: string }>
for await (const result of streamBus.execute(new CreateUser())) {
  console.log(result);
}

API 文档

CommandBus<TContext, TPlugin>

构造器

方法

高级使用

自定义上下文

interface AppContext {
  requestId: string;
  user: { id: string };
}

const bus = new CommandBus<AppContext>({
  context: {
    requestId: "123",
    user: { id: "system" }
  }
});

bus.register(CreateUser, (cmd, ctx) => {
  console.log(ctx.user.id); // "system"
  return { id: ctx.requestId };
});

处理插件出错

const errorHandlingPlugin: CommandBusPlugin<
    Command,
    typeof context,
    Command[COMMAND_RETURN]
  > = {
    handler: (command, ctx, handler) => {
        try {
            return {
                success: true,
                value: handler?.(command, ctx)
            };
        } catch (error) {
            console.error("Command failed:", command);
            return {
                success: false,
                error,
            }
        }
    },
  };

const safeCommandBus = new CommandBus({ plugin: errorHandlingPlugin });

插件

HTTP 插件

HTTP 插件将您的指令Bus 与 HTTP 端点—— 无论是在客户端还是服务器端—— 整合, 同时保留全类型安全并充分利用插件系统 .

HTTP 客户端插件

HttpClientPlugin拦截命令执行,并将 HTTP POST 请求发送给您的 API 端点 。 它支持:

• 动态路由:提供根据命令和上下文生成URL的功能.
• 自定义信头:供应静态信头或返回信头的函数.
• 自定义序列器:覆盖不同内容类型的默认JSON序列器.

示例使用法:

import { CommandBus, Command, httpClientPlugin } from "@collidor/command";

// Define a command
class CreateUser extends Command<{ name: string }, { id: string }> {}

// Create the HTTP client plugin with custom headers
const clientPlugin = httpClientPlugin("http://api.example.com/createUser", {
  headers: () => ({
    "X-API-Key": "my-secret-key",
  }),
});

// Create the CommandBus with the HTTP client plugin installed.
// If a local handler is registered, it will take precedence.
const bus = new CommandBus({ plugin: clientPlugin });

// Execute the command.
// makes an HTTP POST.
const user = await bus.execute(new CreateUser({ name: "Alice" }));
console.log(user); // e.g. { id: "generated-id-xxx" }

注释:

如果提供了本地处理器,将会被使用;否则,HTTP客户端插件会发送POST请求. 默认序列器会将数据转换为JSON,并期望JSON响应,但你可以自行提供.

HTTP 服务器插件

HttpServerPlugin 允许您以 HTTP 端点曝光您的指令Bus 。 它提供帮助方法:

• 从请求中提取一个命令: 获取命令从请求中获取
• 处理请求: 处理请求

此插件将收到的请求解为命令, 在总线上执行, 并返回结果 .

示例使用法:

import { CommandBus, Command, httpServerPlugin } from "@collidor/command";

// Define a command
class CreateUser extends Command<{ name: string }, { id: string }> {}

// Create an HTTP server plugin instance
const serverPlugin = httpServerPlugin();

// Create the CommandBus with the server plugin installed
const bus = new CommandBus({ plugin: serverPlugin });

// Register a command handler
bus.register(CreateUser, (command) => ({
  id: "generated-id-" + Math.random().toString(36).substring(2, 9),
}));

// Simulate an HTTP POST request (e.g., within your web framework)
const request = new Request("http://api.example.com", {
  method: "POST",
  headers: { "Content-Type": "application/json" },
  body: JSON.stringify({
    data: { name: "Alice" },
    name: CreateUser.name,
  }),
});

// Handle the request using the plugin
serverPlugin.handleRequest(request).then((result) => {
  console.log(result);
  // Expected output: { id: "generated-id-xxx" }
});

注释:

确保插件通过命令Bus安装,以便获得命令From Request可以访问命令构造器. 服务器插件默认使用与客户端插件相同的序列器逻辑. 如果需要, 请自定义 .

自定义序列化

HTTP 两种插件都允许您提供自定义的序列器,如果您需要使用不同的数据格式或内容类型工作的话.

示例

import { Serializer } from "@collidor/command/plugins/httpPlugin";

const customSerializer: Serializer<Blob> = {
  serialize: (data, headers) => {
    headers.set("Content-Type", "application/octet-stream");
    return new Blob([JSON.stringify(data)], { type: "application/octet-stream" });
  },
  deserializeResponse: async (response) => {
    if (!response.ok) {
      throw new Error(`HTTP error: ${response.status}`);
    }
    const text = await response.text();
    return JSON.parse(text);
  },
  deserializeRequest: async (request) => {
    const text = await request.text();
    return JSON.parse(text);
  },
};

const clientPlugin = httpClientPlugin("http://api.example.com", {
  serializer: customSerializer,
});

通过集成HTTP客户端和服务器插件,您可以轻松地将您的命令处理器与远程HTTP服务连接起来,同时仍然享受到由命令库提供的类型安全和插件扩展性的所有好处.

端口频道插件

新的PortChannel插件从@collidor/event 调用基本的PortChannel,在不同的环境中(例如服务器和客户端之间)连接命令执行. 它规定:

• 缓冲事件 :

  如果某一事件在任何订阅者连接之前发布,插件会缓冲该事件,并在订阅者出现时立即冲出.

• 最大缓冲超时 :

  缓冲事件在可配置的超时后被自动丢弃(默认为5000ms)来防止内存出漏.

• 程序间沟通:

  允许通过 MessagePort 发送命令并接收回复,使其对网络工人,iframes 或其他多文本环境等情景最为理想.

示例使用法:

import { CommandBus, Command, PortChannelPlugin } from "@collidor/command";

// Define a command
class CreateUser extends Command<{ name: string }, { id: string }> {}

// Create an instance of the PortChannel Plugin.
// Optionally, configure the buffer timeout (in milliseconds)
const portPlugin = new PortChannelPlugin({ bufferTimeout: 5000 });

// Create a CommandBus with the PortChannel Plugin installed.
const bus = new CommandBus({ plugin: portPlugin });

// Register a command handler on the bus.
bus.register(CreateUser, (command, context) => ({
  id: "user-" + Math.random().toString(36).slice(2),
}));

// Dispatch a command via the PortChannel Plugin.
// The plugin buffers events until a subscriber (e.g. a connected MessagePort) attaches.
const user = await bus.execute(new CreateUser({ name: "Alice" }));
console.log(user); // e.g. { id: "user-..." }

注释:

• 缓行:

  在订阅者连接之前发送的命令或事件将存储在缓冲器中,然后在订阅者(通过添加Port订阅)附加后立即冲出.

• 超时控制 :

  缓冲事件的最大时间由缓冲Timeout选项控制.

• 互通性:

  这个插件在服务器和客户端分别居住在执行上下文(如网络工人或iframes)的情景中特别有用,确保事件不会在初始连接延迟时丢失.

类型转换

基于插件的自动返回类型包装 :

// Given this plugin:
  const plugin: CommandBusPlugin<
    Command,
    typeof context,
    Command[COMMAND_RETURN][]
  > = {
    handler: (command, c, h) => [(h?.(command, c))],
  };

// Return type becomes Array<{ id: string }>
const result = bus.execute(new CreateUser());

捐款

1. 分叉仓库
2. 创建特性分支( git checkout - b 特性/ fooBar)
3. 执行更改( git 承诺 - a 添加一些 fooBar)
4. 推到分支( git 推出源功能/ fooBar)
5. 创建新的拉请求

许可证

麻省理工学院