@ atcute/ firehose (法语)

aT协议的轻量级XRPC订阅客户端.

npm install @atcute/firehose

此软件包为 XRPC 订阅提供通用客户端 - 基于 WebSocket 的流线 aT议定书使用的协议。 它处理框架解码、自动重接和可选计划 审定.

用于消费 Bluesky 网络起火,请考虑使用 @atcute/jetstream 相反, 它提供了更简单的基于JSON的界面.

使用量

连接到接力火花

import { FirehoseSubscription } from '@atcute/firehose';
import { ComAtprotoSyncSubscribeRepos } from '@atcute/atproto';

const subscription = new FirehoseSubscription({
    service: 'wss://bsky.network',
    nsid: ComAtprotoSyncSubscribeRepos.mainSchema,
});

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

当您从循环中跳出时,连接会打开,然后关闭。 页:1 基础的 WebSocket 在断开时自动重接.

取消订阅

通过一个 AbortSignal 关闭连接并拒绝重复:

const subscription = new FirehoseSubscription({
    service: 'wss://bsky.network',
    nsid: ComAtprotoSyncSubscribeRepos.mainSchema,
    signal: AbortSignal.timeout(10_000),
});

for await (const message of subscription) {
    console.log(message.$type);
}

处理信件类型

信件包括: $type 字段表示其类型 :

for await (const message of subscription) {
    switch (message.$type) {
        case 'com.atproto.sync.subscribeRepos#commit': {
            // repository commit (record creates, updates, deletes)
            console.log('commit:', message.repo, message.rev);
            break;
        }

        case 'com.atproto.sync.subscribeRepos#handle': {
            // handle change
            console.log('handle:', message.did, message.handle);
            break;
        }

        case 'com.atproto.sync.subscribeRepos#migrate': {
            // account migration
            console.log('migrate:', message.did, message.migrateTo);
            break;
        }

        case 'com.atproto.sync.subscribeRepos#tombstone': {
            // account deletion
            console.log('tombstone:', message.did);
            break;
        }

        case 'com.atproto.sync.subscribeRepos#identity': {
            // identity update
            console.log('identity:', message.did);
            break;
        }

        case 'com.atproto.sync.subscribeRepos#account': {
            // account status change
            console.log('account:', message.did, message.active);
            break;
        }
    }
}

跟踪光标以恢复

使用函数用于 params 为每次连接尝试提供当前光标:

let cursor: number | undefined;

// load saved cursor if resuming
const saved = localStorage.getItem('firehose-cursor');
if (saved) {
    cursor = Number(saved);
}

const subscription = new FirehoseSubscription({
    service: 'wss://bsky.network',
    nsid: ComAtprotoSyncSubscribeRepos.mainSchema,
    // function is called on each connection/reconnection
    params: () => ({ cursor }),
});

for await (const message of subscription) {
    if ('seq' in message) {
        cursor = message.seq;

        // periodically save cursor for recovery
        if (cursor % 1000 === 0) {
            localStorage.setItem('firehose-cursor', String(cursor));
        }
    }
}

每次连接尝试时都会调用参数函数,所以当 WebSocket 在一个 断开,它自动使用最新的光标值.

使用多个服务器

通过一系列 URL 自动失效。 客户端在每次连接上随机选择一个:

const subscription = new FirehoseSubscription({
    service: ['wss://bsky.network', 'wss://bsky-relay.example.com'],
    nsid: ComAtprotoSyncSubscribeRepos.mainSchema,
});

处理出错

XRPC订阅可以发送出出错帧. 处理他们与 onError 回调 :

const subscription = new FirehoseSubscription({
    service: 'wss://bsky.network',
    nsid: ComAtprotoSyncSubscribeRepos.mainSchema,
    onError(err) {
        console.error('firehose error:', err);
        // common errors:
        // - "FutureCursor": cursor is ahead of the server
        // - "ConsumerTooSlow": client is not consuming messages fast enough
    },
});

连接周期召回

处理记录或 UI 更新的连接事件 :

const subscription = new FirehoseSubscription({
    service: 'wss://bsky.network',
    nsid: ComAtprotoSyncSubscribeRepos.mainSchema,
    onConnectionOpen(event) {
        console.log('connected to firehose');
    },
    onConnectionClose(event) {
        console.log('disconnected:', event.code, event.reason);
    },
    onConnectionError(event) {
        console.error('connection error:', event.error);
    },
});

在运行时更新选项

使用 updateOptions()。这引发了重联:

const subscription = new FirehoseSubscription({
    service: 'wss://bsky.network',
    nsid: ComAtprotoSyncSubscribeRepos.mainSchema,
});

// later, switch to a different service
subscription.updateOptions({
    service: 'wss://different-relay.example.com',
});

禁用信件验证

默认情况下,电文将对照计划进行验证。 如果您为更好的性能, 请禁用此选项 信任服务器 :

const subscription = new FirehoseSubscription({
    service: 'wss://bsky.network',
    nsid: ComAtprotoSyncSubscribeRepos.mainSchema,
    validateEvents: false,
});

WebSocket 选项

将选项传递给基础 派对 网络安全套 自定义再连接行为 :

const subscription = new FirehoseSubscription({
    service: 'wss://bsky.network',
    nsid: ComAtprotoSyncSubscribeRepos.mainSchema,
    ws: {
        maxRetries: 10,
        minReconnectionDelay: 1000,
        maxReconnectionDelay: 30000,
    },
});