键入- pubsub

用于Google Cloud和Firebase的型式安全 PubSub抽象.

特征

• 类型安全通讯: 对出版商和处理者的完整类型脚本支持
• 运行时间验证: 使用 Zod 的 Schema 校验可以防止无效信件
• 重复预防: 防止重复的可选事件标记 处理
• 存储信件处理: 选择基于时间发送信件
• 自定义默认: 配置您自己的全库默认值 每个手控器都超过 箱外的你得到:
  • 以7天的指数回放启用复写
  • 缩放到每个话题最多250个实例( 比 Firebase 默认安全) 三千亿
  • 每例512MB 内存分配

安装

npm install @codecompose/typed-pubsub

同伴依赖关系

此软件包具有以下对等依赖性:

• @google-cloud/pubsub
• firebase-functions
• zod

快速启动

1. 与 Zod 定义您的信件计划
2. 创建 PubSub 客户端
3. 用您的计划初始化已键入的 PubSub
4. 创建类型安全出版社和处理器

import { PubSub } from "@google-cloud/pubsub";
import { createTypedPubsub } from "@codecompose/typed-pubsub";
import { z } from "zod";

// 1. Define your schemas
const schemas = {
  user_created: z.object({
    userId: z.string(),
    email: z.string().email(),
  }),
};

// 2. Create PubSub client
const client = new PubSub();

// 3. Initialize typed PubSub
const pubsub = createTypedPubsub({
  client,
  schemas,
  region: "us-central1",
});

// 4a. Create and use a publisher
await pubsub.createPublisher("user_created")({
  userId: "123",
  email: "[email protected]",
});

// 4b. Create a handler
export const handle_user_created = pubsub.createHandler({
  topic: "user_created",
  handler: async (data) => {
    // data is fully typed based on the schema
    console.log(`New user created: ${data.email}`);
  },
});

用法实例

基本设置

import { PubSub } from "@google-cloud/pubsub";
import { createTypedPubsub } from "@codecompose/typed-pubsub";
import { z } from "zod";

// Define your schemas
const schemas = {
  user_created: z.object({
    userId: z.string(),
    email: z.string().email(),
    createdAt: z.string().datetime(),
  }),
  order_placed: z.object({
    orderId: z.string(),
    userId: z.string(),
    items: z.array(
      z.object({
        productId: z.string(),
        quantity: z.number().int().positive(),
      }),
    ),
    total: z.number().positive(),
  }),
};

// Create typed Pubsub client
const pubsub = createTypedPubsub({
  client: new PubSub(),
  schemas,
  region: "us-central1",
});

// Publish a message
await pubsub.createPublisher("user_created")({
  userId: "123",
  email: "[email protected]",
  createdAt: new Date().toISOString(),
});

// Create a handler
export const handle_order_placed = pubsub.createHandler({
  topic: "order_placed",
  handler: async (data) => {
    // data is fully typed based on the schema
    console.log(`Processing order ${data.orderId} for user ${data.userId}`);
    // Process the order...
  },
  options: {
    memory: "1GiB",
    timeoutSeconds: 60,
    markEvent: true, // Never process this event more than once
  },
});

说明: 例子使用 srave case来表示主题名称( 如“ user created ” ) 和导出云函数(像 handle order placed)。 这是因为 外壳目前在GCP中的 Cloud Run 函数名中被忽略, 因此使用 用于一致性的蛇 case 优先.

防止与事件标记重复处理

PubSub有至少一次发送的语义,意思是信息有时可能 不止一次交付。 对于必须执行一次的行动 (如支付处理或用户信用奖),此库提供事件 标记.

这个 markEvent 选项使用您提供的存储器( 如 Redis) 来跟踪 已处理事件标识,确保每个事件得到准确处理 一次,尽管 潜在的再加工.

import { PubSub } from "@google-cloud/pubsub";
import { createTypedPubsub } from "@codecompose/typed-pubsub";
import { redis } from "./redis-client";

// Create event marking functions
const eventMarkingFunctions = {
  isEventProcessed: async (eventId) => {
    return Boolean(await redis.get(`event:${eventId}`));
  },
  markEventAsProcessed: async (eventId) => {
    await redis.set(`event:${eventId}`, "1", "EX", 86400); // 24 hours TTL
  },
};

// Create typed Pubsub client with event marking
const pubsub = createTypedPubsub({
  client: new PubSub(),
  schemas,
  region: "us-central1",
  options: {
    eventMarkingFunctions,
    defaultHandlerOptions: {
      markEvent: true, // Enable for all handlers by default
      vpcConnector: "redis-connector", // If Redis is in a VPC
    },
  },
});

处理时间敏感事件

对于时间敏感的操作,您可能想要丢弃太过的消息 老样子 这个 retryMaxAgeMinutes 选项自动减少超过 指定年龄,防止处理陈旧数据.

// Configure time-based message dropping in the handler
export const handle_time_sensitive_events = pubsub.createHandler({
  topic: "time_sensitive_events",
  handler: async (data) => {
    // Process time-sensitive event
  },
  options: {
    retryMaxAgeMinutes: 60, // Only process events less than 1 hour old
  },
});

API 参考

createTypedPubsub(options)

创建一个类型安全的 Pubsub 客户端, 用于使用 schema 验证处理信件 .

参数

• client: Google Cloud Pubsub 客户端实例
• schemas: 每一专题的Zod计划记录
• region: 用于 Pubsub 函数的 GCP 区域
• options: (可选) 配置选项
  • eventMarkingFunctions: (可选)跟踪已处理事件的职能
  • defaultHandlerOptions: (可选) 所有处理器的默认选项
• onMessagePublished: (可选) Firebase 消息已发布处理器, 默认值 用于实施火基地

密钥处理器选项

关于重试行为的重要注释: 当您可以启用/ 无法重试时 并设定重试事件的最大年龄,实际重试时间由 pubSub 订阅的 ACK 截止日期, 而不是 timeoutSeconds 设置。 这个 超时会关闭服务器进程, 但事件将只进行重试 aCK 最后期限到期后, 无法通过此库配置 .

限制和替代办法

虽然这个库提供了极佳的种类安全和计划验证 PubSub消息, PubSub 有一些固有的限制:

• 对重试时间的控制有限(由订阅ACK截止日期控制)
• 没有内置支持手动触发重试
• 没有针对不同故障情形的精细错误处理

如果您使用时需要更高级的功能, 如控制重试 间隔、 手动重试触发或长期背景进程, 考虑使用专门的任务队列系统。 这个 键入任务 库提供类似的 类型安全保障,对任务执行有更严格的控制.