@ tcute/ 喷气流

aT协议的轻量级喷气流用户.

npm install @atcute/jetstream

喷气流 是提供过滤器的流线服务 aT协议事件作为JSON在WebSocket上的火花. 此软件包提供了订阅客户端 对这些事件.

使用量

签署事件

import { subscribeEvents } from '@atcute/jetstream';

const events = subscribeEvents({
    service: 'wss://jetstream.us-east.bsky.network',
});

for await (const message of events) {
    console.log(message.$type, message.seq, message.did);
}

连接在迭代开始时打开, 自动重接, 在迭代开始时关闭 结束. service 接受一个主机,因为 v2 序列光标是 Jetstream 实例所特有的 .

过滤

过滤器在省略或空时合并并匹配一切 :

const postEvents = subscribeEvents({
    service: 'wss://jetstream.us-east.bsky.network',
    collections: ['app.bsky.feed.post', 'app.bsky.graph.*'],
    dids: ['did:plc:z72i7hdynmk6r22z27h6tvur'],
    kinds: ['commit'],
});

collections 只有过滤器执行事件。 把它和 kinds: ['commit'] 排除身份, 账户和同步事件.

处理事件类型

使用情况 $type 在访问其字段前缩小一个事件:

for await (const message of events) {
    switch (message.$type) {
        case 'network.bsky.jetstream.subscribeEvents#commit': {
            console.log(message.operation, message.collection, message.rkey, message.record);
            break;
        }

        case 'network.bsky.jetstream.subscribeEvents#identity': {
            console.log(message.identity.handle);
            break;
        }

        case 'network.bsky.jetstream.subscribeEvents#account': {
            console.log(message.account.active, message.account.status);
            break;
        }

        case 'network.bsky.jetstream.subscribeEvents#sync': {
            console.log(message.sync.rev);
            break;
        }
    }
}

承诺字段直接放在信息上。 身份、账户和同步消息 相应的上游中继事件。 此版本软件包未知的事件类型被跳过 不推进光标.

正在验证记录

默认验证事件信封,但执行 record 数值仍未确认。 缩窄 他们与 is() 从 @atcute/lexicons数字 :

import { AppBskyFeedPost } from '@atcute/bluesky';
import { is } from '@atcute/lexicons';

for await (const message of events) {
    if (message.$type !== 'network.bsky.jetstream.subscribeEvents#commit') {
        continue;
    }

    if (is(AppBskyFeedPost.mainSchema, message.record)) {
        console.log(`@${message.did}: ${message.record.text}`);
    }
}

设定 validateEvents: false 以跳过信封验证。 记录仍必须单独验证.

从光标恢复

通过一个起始序列号,或一个 CursorStore 以装入并保存位置 :

import { readFile, writeFile } from 'node:fs/promises';

const events = subscribeEvents({
    service: 'wss://jetstream.us-east.bsky.network',
    cursor: {
        async load() {
            const saved = await readFile('cursor', 'utf8').catch(() => undefined);
            return saved !== undefined ? Number(saved) : undefined;
        },
        async save(seq) {
            await writeFile('cursor', String(seq));
        },
    },
});

load 决定订阅的开始地点 并且每一次重新连接都要再次咨询 而不是仅仅在第一次连接上,所以在其他地方写的商店对下一次连接生效 尝试。返回 undefined 从直播提示开始。 序列号被视为存储 只有订阅才能持续.

当消费者要求下一次活动时,存储的光标会前进。 写入受以下限制: cursorSaveInterval,默认为 5 秒;待决写入在重新连接前被冲出 当迭接结束。 将间隔设置为 0 以保存每个事件.

连接时显示存储失败 onConnectionError 并且连接尝试是 再审. (单位:千美元) save 迭代时的失败被抛出, 除了最后的冲出, 该冲出 onError 因此,它没有掩盖为什么迭代结束.

交货至少一次。 重接的复制件在同一个转发器运行时被删除, 但是 一个新的浏览器可以重放它保存的光标事件。 因此,事件处理者应是一能的.

取消订阅

a 一个 AbortSignal 拦住溪口 重复拒绝方式 signal.reason 冲出任何待决事项后 光标 :

const events = subscribeEvents({
    service: 'wss://jetstream.us-east.bsky.network',
    signal: AbortSignal.timeout(10_000),
});

处理出错

onError 接收错误的帧、验证失败和服务器错误帧。 使用服务器出错 FirehoseError数字 :

import { FirehoseError, subscribeEvents } from '@atcute/jetstream';

const events = subscribeEvents({
    service: 'wss://jetstream.us-east.bsky.network',
    onError: (err) => {
        if (err instanceof FirehoseError) {
            console.error('jetstream sent', err.error);
        } else {
            console.error('jetstream error:', err);
        }
    },
});

在 WebSocket 升级前返回出错, 包括 CursorTooOld 财务报告和审定财务报表 InvalidRequest,没有 可见于 WebSocket 客户端,并看起来像连接失败。 默认重审; 使用 ws.maxRetries 来设定一个限制。 达到极限时停止重接,但延展器打开.

连接选项

onConnectionOpen, (中文(简体) ), onConnectionClose,以及 onConnectionError 报告套接字的生命周期,以及 ws 将选项传递到下面 派对 网络锁定 :

const events = subscribeEvents({
    service: 'wss://jetstream.us-east.bsky.network',
    onConnectionClose: (event) => console.log('disconnected', event.code, event.reason),
    ws: {
        minReconnectionDelay: 1_000,
        maxReconnectionDelay: 30_000,
    },
});