atproto-jetstream#

WebSocket event stream consumer library for AT Protocol Jetstream.

Overview#

atproto-jetstream provides a comprehensive async stream consumer for AT Protocol Jetstream events. This library enables real-time consumption of AT Protocol repository events, identity changes, and account updates through WebSocket connections with support for filtering, compression, and graceful shutdown patterns.

Binaries#

• atproto-jetstream-consumer: Real-time event stream consumer with filtering and compression support

Features#

• High-performance WebSocket streaming with automatic reconnection
• Flexible event handler system with multiple custom handlers
• Zstandard compression support with custom dictionaries
• Event filtering by collections and DIDs
• Graceful shutdown with cancellation token support
• Structured error handling with detailed error codes

Usage#

Basic Event Consumer#

use atproto_jetstream::{Consumer, ConsumerTaskConfig, EventHandler, JetstreamEvent, CancellationToken};
use async_trait::async_trait;

struct MyEventHandler;

#[async_trait]
impl EventHandler for MyEventHandler {
    async fn handle_event(&self, event: JetstreamEvent) -> anyhow::Result<()> {
        println!("Received event: {:?}", event);
        Ok(())
    }

    fn handler_id(&self) -> String {
        "my-handler".to_string()
    }
}

let config = ConsumerTaskConfig {
    user_agent: "my-app/1.0".to_string(),
    compression: false,
    zstd_dictionary_location: String::new(),
    jetstream_hostname: "jetstream1.us-east.bsky.network".to_string(),
    collections: vec!["app.bsky.feed.post".to_string()],
};

let consumer = Consumer::new(config);
consumer.register_handler(std::sync::Arc::new(MyEventHandler)).await?;

let cancellation_token = CancellationToken::new();
consumer.run_background(cancellation_token).await?;

With Compression#

let config = ConsumerTaskConfig {
    compression: true,
    zstd_dictionary_location: "./data/zstd_dictionary".to_string(),
    // ... other config
};

// Download dictionary first:
// curl -o data/zstd_dictionary https://github.com/bluesky-social/jetstream/raw/refs/heads/main/pkg/models/zstd_dictionary

Command Line Examples#

# Basic streaming
cargo run --bin atproto-jetstream-consumer \
    --hostname jetstream1.us-east.bsky.network \
    --collections app.bsky.feed.post \
    --user-agent "my-consumer/1.0"

# With compression
cargo run --bin atproto-jetstream-consumer \
    --hostname jetstream1.us-east.bsky.network \
    --collections app.bsky.feed.post \
    --compression \
    --zstd-dictionary ./data/zstd_dictionary

# With filtering
cargo run --bin atproto-jetstream-consumer \
    --hostname jetstream1.us-east.bsky.network \
    --collections "app.bsky.feed.post,app.bsky.actor.profile" \
    --dids "did:plc:user123,did:plc:user456"

License#

MIT License