@crossdelta/cloudevents

中文文档

0.10.1发布于 2026/10/25MITTypeScriptESM + CJS近 30 天下载 4041

用于类型Script的 CloudEvents 工具包 - Zod验证,处理器发现,NATS JetStream & Core

文档目录23 节

快速启动
为什么用这个?
核心概念
事件类型对事件数据
处理器
出版
串流设置
消费
配置
环境变量
可选内容:Google Cloud Pub/Sub
流出选项
消费者选择
高级特性
与频道的基本合同
频道配置
多个事件, 相同流
基础设施
发展与生产
后向兼容性
API 密码
分布式追踪
许可证

@ crossdelta/ clodevents (英语)

类型安全事件驱动微服务海军财务报告和审定财务报表佐德语Name 验证,使用云节规格.

                         NATS JetStream
                        ┌──────────────┐
  ┌──────────────┐      │              │      ┌──────────────┐
  │    Service   │      │   Stream:    │      │   Service    │
  │   (publish)  │─────▶│   ORDERS     │─────▶│  (consume)   │
  └──────────────┘      │              │      └──────────────┘
                        └──────────────┘
        │                                            │
        │  publish(...)                              │  handleEvent(...)
        ▼                                            ▼
  ┌──────────────┐                            ┌──────────────┐
  │ { orderId,   │                            │ Zod schema   │
  │   total }    │                            │ validates    │
  └──────────────┘                            └──────────────┘

bun add @crossdelta/cloudevents zod@4

先决条件: 运行 NATS 服务器已启用 JetStream .

说明: 需要 Zod v4 支持全部类型脚本 .

快速启动

说明: 服务 从不创建流在服务消费前,必须先有流:

发展: 使用内兹克州或者说平台 SDK (单位:千美元)pf dev 自动创建合同)

生产: 使用基础设施即用代码(Pulumi、Terraform)或串流设置下面的函数

1. 创建事件处理器 (单位:千美元)src/events/order-created.handler.ts:

import { handleEvent } from '@crossdelta/cloudevents'
import { z } from 'zod'

// Export schema for mock generation
export const OrderCreatedSchema = z.object({
  orderId: z.string(),
  total: z.number(),
})

// Export type for use in use-cases
export type OrderCreatedEvent = z.infer<typeof OrderCreatedSchema>

export default handleEvent(
  {
    schema: OrderCreatedSchema,
    type: 'order.created',
  },
  async (data) => {
    console.log(`New order: ${data.orderId}, total: ${data.total}`)
  },
)

2. 开始消费:

import { consumeJetStreams } from '@crossdelta/cloudevents'

consumeJetStreams({
  streams: ['ORDERS'],
  consumer: 'my-service',
  discover: './src/events/**/*.handler.ts',
})

3. 从另一处出版:

import { publish } from '@crossdelta/cloudevents'

await publish('order.created', { orderId: 'ord_123', total: 99.99 })
// String API auto-pluralizes the domain token: subject → 'orders.created'
// Use createContract({ type, channel, schema }) for explicit subject control (no pluralization)

就是这样。 Handlers是自动发现的,与Zod验证的,并且消息在JetStream中持续存在.

流创建 由您的发展环境或基础设施处理,而不是由服务处理:

发展: 内兹克州, (中文(简体) ), @ crossdelta/ platform-sdk (英语),或手动设置

生产:见串流设置下图 ensureJetStreams() 基础设施代码的使用

为什么用这个?

问题	解决方案
重新启动时丢失的消息	JetStream 持续消息
零散处理器登记	通过光彩图案自动发现
运行时类型出错	用 TypeScript 推论验证 Zod
有毒信息崩溃服务	DLQ 隔离无效消息

核心概念

事件类型对事件数据

重要的区别:

事件类型 (单位:千美元)order.created:生活在云端信封 (单位:千美元)ce.type) (中文(简体) ). 用于路由和处理器匹配.
事件数据 (单位:千美元){ orderId, total }实际有效载荷。对没有包括类型.

const Schema = z.object({
  orderId: z.string(), 
})

export default handleEvent(
  {
    schema: Schema,
    type: 'order.created',
  },
  async (data) => { ... }
)

处理器

放下一个 *.handler.ts 任何地方的文件——是自动注册的:

// src/events/user-signup.handler.ts
import { z } from 'zod'

const UserSignupSchema = z.object({ 
  email: z.email(),
  name: z.string(),
})

// Export type for use in use-cases
export type UserSignupEvent = z.infer<typeof UserSignupSchema>

export default handleEvent(
  {
    schema: UserSignupSchema,
    type: 'user.signup',
  },
  async (data) => {
    await sendWelcomeEmail(data.email)
  },
)

出版

await publish('order.created', orderData)

串流设置

仅供基础设施使用 - 服务不应该叫这个功能.

在基础设施设置中创建流( Pulumi, 部署脚本) :

import { ensureJetStreams } from '@crossdelta/cloudevents'

await ensureJetStreams({
  streams: [
    {
      stream: 'ORDERS',
      subjects: ['order.>'],  // singular — matches contract subjects (order.created, order.completed, …)
      config: {
        maxAge: 7 * 24 * 60 * 60 * 1000, // 7 days retention
        replicas: 3,                       // For HA
      },
    },
  ],
})

消费

import { consumeJetStreams } from '@crossdelta/cloudevents'

// JetStream (recommended) — persistent, retries, exactly-once
consumeJetStreams({
  streams: ['ORDERS'],
  consumer: 'billing',
  discover: './src/events/**/*.handler.ts',
})

// Core NATS — fire-and-forget, simpler
await consumeNatsEvents({
  subject: 'notifications.*',
  consumerName: 'my-service',
  discover: './src/events/**/*.handler.ts',
})

// NATS Core Request-Reply — synchronous validation feedback
// Consumer returns result, publisher awaits it
import { request } from '@crossdelta/cloudevents'
const result = await request('call.received', payload)
// result: { accepted: true, data: { ... } } | { accepted: false, error: '...' }

配置

环境变量

NATS_URL=nats://localhost:4222
NATS_USER=myuser        # optional
NATS_PASSWORD=mypass    # optional

可选内容:Google Cloud Pub/Sub

要使用 Google Cloud Pub/ Sub 而不是 NATS, 请安装可选的依赖 :

bun add @google-cloud/pubsub
# or
npm install @google-cloud/pubsub

那就用那个 publishToPubSub 函数 :

import { publishToPubSub } from '@crossdelta/cloudevents'

await publishToPubSub('order.created', { orderId: 'ord_123', total: 99.99 })

说明: @google-cloud/pubsub 不被捆绑到此库中以保持软件包大小。只有当您需要 Google Cloud Pub/ Sub 支持时才安装它 .

流出选项

await ensureJetStreams({
  streams: [
    {
      // Required
      stream: 'ORDERS',
      subjects: ['order.>'],  // singular — matches contract subjects

      // Optional
      config: {
        maxAge: 7 * 24 * 60 * 60 * 1000,  // Message retention (ms)
        maxBytes: 1024 * 1024 * 1024,       // Reserved at creation, not on write
        replicas: 1,                        // Replication factor
        storage: 'file',                    // 'file' or 'memory'
        retention: 'limits',                // 'limits', 'interest', or 'workqueue'
      },
    },
  ],
  // Optional
  servers: 'nats://localhost:4222',
})

消费者选择

consumeJetStreams({
  // Required
  streams: ['ORDERS'],
  consumer: 'my-service',
  discover: './src/events/**/*.handler.ts',

  // Optional
  servers: 'nats://localhost:4222',
  filterSubjects: ['order.created'],  // Filter specific subjects
  maxDeliver: 5,                       // Retry attempts
  ackWait: 30_000,                     // Timeout per attempt (ms)
  quarantineTopic: 'dlq',              // For poison messages
})

高级特性

通道元数据( 基础设施集成)

合同可包括: 频道元数据 来定义它们属于哪个NATS JetStream流. 这将使基础设施全码工作流程能够实现合同中的流程.

与频道的基本合同

import { createContract } from '@crossdelta/cloudevents'
import { z } from 'zod'

export const OrdersCreatedContract = createContract({
  type: 'order.created',
  channel: {
    stream: 'ORDERS',
    // subject defaults to 'order.created' if omitted
  },
  schema: z.object({
    orderId: z.string(),
    total: z.number(),
  }),
})

频道配置

interface ChannelConfig {
  stream: string    // JetStream stream name (e.g., 'ORDERS')
  subject?: string  // NATS subject (optional override)
}

主题路线 :

无 subject: 默认为 type —— 说 order.created 留下来 order.created
与 subject: 使用精确的字符串

说明: publish('order.x', data) (字符串 API) 自动拓扑 → orders.x合同没有. 配置 JetStream 有单个主题的流流( )order.>与合同活动相匹配.

// subject = 'order.created' (from type)
createContract({ type: 'order.created', channel: { stream: 'ORDERS' } })

// subject = 'orders.v1.created' (explicit override)
createContract({ type: 'order.created', channel: { stream: 'ORDERS', subject: 'orders.v1.created' } })

多个事件, 相同流

// order.created → ORDERS stream
export const OrdersCreatedContract = createContract({
  type: 'order.created',
  channel: { stream: 'ORDERS' },
  schema: OrderCreatedSchema,
})

// order.updated → ORDERS stream (same stream!)
export const OrdersUpdatedContract = createContract({
  type: 'order.updated',
  channel: { stream: 'ORDERS' },
  schema: OrderUpdatedSchema,
})

结果 : 两个活动都分享 ORDERS 包含主题的流 order.created 财务报告和审定财务报表 order.updated。。。 .

基础设施

频道元数据是概念 - 它定义了路线,而不是基础设施政策.

合同定义:

事件类型( E)order.created(中文(简体) )
流出路由 (ORDERS(中文(简体) )
主题映射( O)order.created(中文(简体) )

基础设施定义:

留用(7天,14天等)
存储( 磁盘、内存)
复制品(1,3等)
限制( 最大消息大小等)

实例基础设施( Pulumi) :

import { collectStreamDefinitions } from './helpers/streams'
import { ensureJetStreams } from '@crossdelta/cloudevents'

// Collect from contracts
const streams = collectStreamDefinitions()
// [{ stream: 'ORDERS', subjects: ['order.created', 'order.updated'] }]

// Materialize with policies
await ensureJetStreams({
  streams: streams.map(({ stream, subjects }) => ({
    stream,
    subjects,
    config: {
      maxAge: 14 * 24 * 60 * 60 * 1000, // 14 days
      replicas: 3,                       // High availability
      storage: 'file',                   // Persistent
    },
  })),
})

发展与生产

环境	流线	怎么样	保留
发展	以弗时	由服务自动创建	无(纪念)
生产	持久性	通过基础设施实现	明确(7d,14d等)

关键原则: 合同定义什麽财务报告和审定财务报表地点,基础设施定义 怎么样 财务报告和审定财务报表 多长时间。。。 .

后向兼容性

频道元数据是选项 - 无合同 channel 工作如前:

// Legacy contract (still works)
export const LegacyContract = createContract({
  type: 'legacy.event',
  schema: LegacySchema,
  // No channel - handler still works
})

共享合同

当多个服务消耗同一事件时, 使用 createContract 创建共享活动合同:

// packages/contracts/src/events/orders/created.ts
import { createContract } from '@crossdelta/cloudevents'
import { z } from 'zod'

export const OrderCreatedSchema = z.object({
  orderId: z.string(),
  total: z.number(),
})

export type OrderCreatedData = z.infer<typeof OrderCreatedSchema>

export const OrdersCreatedContract = createContract({
  type: 'order.created',
  schema: OrderCreatedSchema,
})

然后导入和用于处理器 :

import { handleEvent } from '@crossdelta/cloudevents'
import { OrdersCreatedContract } from '@my-org/contracts'

export default handleEvent(
  OrdersCreatedContract,
  async (data) => {
    // data is typed as OrderCreatedData
    console.log(data.orderId)
  },
)

福利:

事件计划的唯一真相来源
没有重复计划
跨服务类别安全

多用途

按租户过滤事件 :

export default handleEvent({
  type: 'order.created',
  schema: OrderSchema,
  tenantId: 'tenant-a',  // Only process tenant-a events
}, async (data) => { ... })

通关卡处理器( Catch- all)

使用 type: '*' 创建处理器全部事件,无论类型。可用于审计记录、衡量标准收集或事件重放.

import { handleEvent } from '@crossdelta/cloudevents'
import { z } from 'zod'

export default handleEvent(
  {
    type: '*',
    schema: z.record(z.string(), z.unknown()),
  },
  async (data, context) => {
    await persistAuditEvent({
      eventType: context.eventType,
      source: context.source,
      data,
    })
  },
)

运行行为 :

具体操作人员(type: 'order.created'总是优先
通配符处理器仅在没有特定处理器匹配时运行
每个消费者只推荐一位通配卡管理员
与JetStream公司和NATS公司合作

自定义匹配

添加自定义过滤逻辑 :

export default handleEvent({
  type: 'order.created',
  schema: OrderSchema,
  match: (event) => event.data.total > 100,  // Only high-value orders
}, async (data) => { ... })

目录

分相是内相. 对于分布式系统,提供Redis商店:

const redisStore = {
  has: (id) => redis.exists(`idem:${id}`),
  add: (id, ttl) => redis.set(`idem:${id}`, '1', 'PX', ttl),
}

await consumeJetStreams({
  streams: ['ORDERS'],
  consumer: 'my-service',
  discover: './src/events/**/*.handler.ts',
  idempotencyStore: redisStore,
})

死信队列

无效消息被隔离, 不丢失 :

await consumeJetStreams({
  streams: ['ORDERS'],
  consumer: 'my-service',
  discover: './src/events/**/*.handler.ts',
  quarantineTopic: 'events.quarantine',  // For malformed messages
  errorTopic: 'events.errors',            // For handler errors
})

无法验证的信件到 quarantineTopic, 崩溃处理器前往的信息 errorTopic。。。 .

HTTP 摄入量( Hono)

import { Hono } from 'hono'
import { cloudEvents } from '@crossdelta/cloudevents'

const app = new Hono()
app.use('/events', cloudEvents({ discover: 'src/events/**/*.handler.ts' }))

API 密码

函数	目的
`handleEvent(options, handler)`	创建处理器( 支持) `type: '*'` 通配符)
`createContract(options)`	创建共享事件合同
`ensureJetStreams(options)`	创建/更新 JetStream 流
`consumeJetStreams(options)`	从多条溪流相接
`consumeNatsEvents(options)`	装火和忘记
`publish(type, data)`	向 JetStream 发布事件( 火灾和遗忘)
`request(type, data, opts?)`	通过 NATS Core 发送命令并等待同步回复
`isNatsConnected()`	检查NATS 发布器连接是否为实时(用于健康检查)
`isConsumerConnected()`	检查是否有任何NATS的消费者连接是实时的(准备状态探测器)
`configureNatsPublisher(config)`	为出版商单通设置连接选项
`closeConnection()`	排水并关闭NATS连接(用于CLI工具)

分布式追踪

何时 @crossdelta/telemetry 已安装(可选的对等依赖),微量上下文自动在 NATS 中传播:

发布: 当前 W3C traceparent 作为 CloudEvent 扩展字段注入.
旋转编号: traceparent 在处理器运行之前,在 Async LocalStorage 中提取并恢复扩展.

这使得Grafana的端到端分布痕迹得以实现:

Storefront → API Gateway → NATS → Orders Service → NATS → Audit Service
                └── all spans share the same traceId ──┘

不需要配置。若为 @crossdelta/telemetry 没有安装, CloudEvents 正常工作没有痕量传播.

许可证

麻省理工学院横跨德尔塔

版本与发布记录

@crossdelta/cloudevents 共发布 71 个版本，首次发布于 2025/12/03；过去 12 个月发布了 71 个版本。

版本	发布日期	距今
`0.10.1`latest	2026/10/25	16 天前
`0.10.0`	2026/09/08	2 个月前
`0.9.0`	2026/09/07	2 个月前
`0.8.4`	2026/07/07	4 个月前
`0.8.3`	2026/07/02	4 个月前
`0.8.2`	2026/04/25	6 个月前

关于 @crossdelta/cloudevents 的常见问题

@crossdelta/cloudevents 是什么？

用于类型Script的 CloudEvents 工具包 - Zod验证,处理器发现,NATS JetStream & Core

如何安装 @crossdelta/cloudevents？

在项目目录运行 npm install @crossdelta/cloudevents，或使用 pnpm add / yarn add / bun add。它还需要宿主项目提供 peer 依赖：zod（^4.0.0）。npm 7 及以上与 pnpm 8 及以上默认会自动安装必需的 peer 依赖，Yarn 需要手动安装。它还提供命令行工具 cloudevents，可以用 npx cloudevents 直接运行。

@crossdelta/cloudevents 支持 TypeScript 吗？

支持。包内自带 TypeScript 类型声明（package.json 声明了 types 入口），安装后即可获得类型提示，无需再安装 @types 包。

@crossdelta/cloudevents 支持 ESM 还是 CommonJS？

同时提供 ES Module 与 CommonJS 入口，import 和 require() 都可以使用，打包工具会自动选择合适的版本。

@crossdelta/cloudevents 可以商用吗？

@crossdelta/cloudevents 使用 MIT 许可。宽松许可：允许商业使用、修改和再分发，通常只需在分发时保留原版权与许可声明。以上为概括，不构成法律意见，请以许可原文为准。

@crossdelta/cloudevents 的下载量有多少？

根据 npm 官方下载统计，@crossdelta/cloudevents 在 2026/10/11 – 2026/11/09 这 30 天内被下载 4,041 次，日均约 135 次。下载量会把 CI 构建、镜像同步和作为其他包依赖的安装一并计入，适合横向比较热度，不能直接等同于使用人数。

@crossdelta/cloudevents 还在维护吗？

最新版本 0.10.1 发布于 2026/10/25（16 天前）；至今共发布 71 个版本，其中过去 12 个月发布了 71 个；首次发布于 2025/12/03。发布频率只是参考信号：功能稳定的工具可能很少需要更新，建议结合源码仓库的 issue 和提交记录判断。

npm install @crossdelta/cloudevents

pnpmpnpm add @crossdelta/cloudevents
yarnyarn add @crossdelta/cloudevents
bunbun add @crossdelta/cloudevents

peer 依赖zod ^4.0.0hono ^4.12.12可选@faker-js/faker >=9.0.0可选@google-cloud/pubsub >=4.0.0可选@crossdelta/telemetry ^0.13.2可选

npm 7+ 与 pnpm 8+ 会自动安装必需的 peer 依赖，Yarn 需手动安装。

命令行npx cloudevents

在你的项目目录中运行此命令。

近 30 天下载

4,041

约 4041 次日均 135 次

统计周期 2026/10/11 – 2026/11/09
数据来自 npm 官方下载统计，今天更新

最新版本0.10.1发布于 16 天前

开源许可MIT宽松许可

包体积408 kB7 个文件 · 解压后

运行依赖6 个另有 5 个 peer 依赖

模块格式ESM + CJS

TypeScript自带类型

发布版本71 个近 12 个月 71 个

首次发布2025/12/0311 个月前

元数据更新2026/10/25

每一个软件包背后，
都有分享与创造的人。