gRPC 客户端（Yellowstone）

安装

cargo add orbitflare-sdk --features grpc

构建客户端

以下示例启用全部选项：

use orbitflare_sdk::{GeyserClientBuilder, RetryPolicy, Result};
use std::time::Duration;

let client = GeyserClientBuilder::new()
    .url("http://ny.rpc.orbitflare.com:10000")
    .fallback_url("http://fra.rpc.orbitflare.com:10000")
    .retry(RetryPolicy {
        initial_delay: Duration::from_millis(100),
        max_delay: Duration::from_secs(30),
        multiplier: 2.0,
        max_attempts: 0,
    })
    .timeout_secs(30)
    .keepalive_secs(60)
    .ping_interval_secs(10)
    .max_missed_pongs(3)
    .channel_capacity(4096)
    .build()?;

最简配置：

let client = GeyserClientBuilder::new()
    .url("http://ny.rpc.orbitflare.com:10000")
    .build()?;

构建器方法

.url(url) — 主 gRPC 端点。回退读取环境变量 ORBITFLARE_GRPC_URL。

.url("http://ny.rpc.orbitflare.com:10000")

.urls(&[...]) — 一次设置主端点与备用端点。第一个元素为主端点。

.urls(&["http://ny.rpc.orbitflare.com:10000", "http://fra.rpc.orbitflare.com:10000"])

.fallback_url(url) / .fallback_urls(&[...]) — 添加备用端点。连接失败时 SDK 会轮换尝试。

.fallback_url("http://fra.rpc.orbitflare.com:10000")

.retry(policy) — 控制重连退避。连接断开时，SDK 先等待 initial_delay，随后每次尝试加倍（上限 max_delay）。将 max_attempts 设为 0 表示无限重试。默认：初始 100ms、最大 30s、2 倍乘数、无限次。

.retry(RetryPolicy {
    initial_delay: Duration::from_millis(200),
    max_delay: Duration::from_secs(15),
    multiplier: 2.0,
    max_attempts: 0,
})

.timeout_secs(n) — 单次 gRPC 请求超时。默认：30。

.timeout_secs(15)

.keepalive_secs(n) — TCP keepalive 间隔。操作系统按此间隔发送探测以在 TCP 层检测死亡连接。默认：60。

.keepalive_secs(30)

.ping_interval_secs(n) — SDK 发送协议层 Ping 的频率。服务端应以 Pong 回应。默认：10。

.ping_interval_secs(15)

.max_missed_pongs(n) — 连续多少次 Ping 无应答后，SDK 认为连接已死并重连。默认：3。在默认参数下，约 30 秒内可检测到死亡连接。

.max_missed_pongs(5)

.channel_capacity(n) — 后台任务与你的代码之间的有界通道容量。若消费事件较慢，后台任务在填满时会阻塞，而非无限占用内存。默认：4096。

.channel_capacity(8192)

编写 YAML 配置

创建包含所需过滤器的 YAML 文件：

# grpc.yml
transactions:
  pumpfun:
    vote: false
    failed: false
    account_include:
      - "6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P"
  jupiter:
    vote: false
    failed: false
    account_include:
      - "JUP6LkbZbjS1jKKwapdHNy74zcZ3tLUZoi5QNyVTaV4"

accounts:
  usdc_mint:
    account:
      - "EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v"
  token_program:
    owner:
      - "TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA"

slots:
  all:
    filter_by_commitment: true

commitment: confirmed

YAML 过滤器参考

transactions — 每个条目是一个命名过滤器。若交易涉及任一 account_include 地址则匹配。account_exclude 排除匹配。account_required 表示所列地址必须全部出现。vote 与 failed 按交易类型过滤。 accounts — 按地址监视特定账户，或通过 owner 监视某程序拥有的全部账户。 slots — 订阅槽位更新。filter_by_commitment 仅在你设定的承诺级别发送更新。 commitment — "processed"、"confirmed" 或 "finalized"。默认为 "confirmed"。 YAML 支持 ${ENV_VAR} 展开：

transactions:
  target:
    account_include:
      - "${TARGET_PROGRAM}"

订阅与读取事件

自 YAML

let mut stream = client.subscribe_yaml("grpc.yml")?;

编程方式

运行时动态构建过滤器：

use std::collections::HashMap;
use orbitflare_sdk::proto::geyser::*;

let mut filters = HashMap::new();
filters.insert("target".into(), SubscribeRequestFilterTransactions {
    vote: Some(false),
    failed: Some(false),
    signature: None,
    account_include: vec![some_address.to_string()],
    account_exclude: vec![],
    account_required: vec![],
});

let request = SubscribeRequest {
    transactions: filters,
    commitment: Some(1),
    ping: Some(SubscribeRequestPing { id: 1 }),
    ..Default::default()
};

let mut stream = client.subscribe(request);

读取流

subscribe_yaml 与 subscribe 均返回 GeyserStream。调用 .next() 获取下一条事件：

use orbitflare_sdk::proto::geyser::subscribe_update::UpdateOneof;

while let Some(update) = stream.next().await {
    let update = update?;
    match update.update_oneof {
        Some(UpdateOneof::Transaction(tx)) => {
            // tx.slot - the slot this transaction was in
            // tx.transaction - the transaction info (signature, accounts, instructions, meta)
        }
        Some(UpdateOneof::Account(acct)) => {
            // acct.slot - the slot
            // acct.account - account info (pubkey, lamports, owner, data)
            // acct.is_startup - true during initial account snapshot
        }
        Some(UpdateOneof::Slot(slot)) => {
            // slot.slot - the slot number
            // slot.status - processed, confirmed, finalized, etc.
        }
        Some(UpdateOneof::BlockMeta(meta)) => {
            // meta.slot, meta.blockhash, meta.parent_slot
            // meta.executed_transaction_count
        }
        _ => {}
    }
}

stream.next() 会阻塞直到事件到达。流关闭时返回 None。每条事件为 Result — Err 表示已用尽全部重试且连接永久不可用。 Pong 消息在内部消费，不会出现在你的流中。

关闭

stream.close();

立即停止后台任务。或直接丢弃流。

多路流

一个客户端可同时运行多条流。每条流使用独立的后台连接：

let mut pumpfun = client.subscribe_yaml("config/pumpfun.yml")?;
let mut raydium = client.subscribe_yaml("config/raydium.yml")?;
let mut slots = client.subscribe_yaml("config/slots.yml")?;

它们共享端点健康状态 — 若某条流将失败端点隔离，其他流在下次重连时会跳过该端点。但各流的连接与生命周期完全独立。可随时新建流，包括根据已有流的数据动态创建。

完整示例

监听 pump.fun 交易、解码签名并打印每条摘要的流：

use orbitflare_sdk::{GeyserClientBuilder, Result};
use orbitflare_sdk::proto::geyser::subscribe_update::UpdateOneof;

#[tokio::main]
async fn main() -> Result<()> {
    let client = GeyserClientBuilder::new()
        .url("http://ny.rpc.orbitflare.com:10000")
        .fallback_url("http://fra.rpc.orbitflare.com:10000")
        .build()?;

    let mut stream = client.subscribe_yaml("grpc.yml")?;
    let mut tx_count: u64 = 0;

    println!("streaming...");

    while let Some(update) = stream.next().await {
        let update = update?;

        match update.update_oneof {
            Some(UpdateOneof::Transaction(tx)) => {
                tx_count += 1;
                if let Some(info) = &tx.transaction {
                    let sig = bs58::encode(&info.signature).into_string();
                    let fee = info.meta.as_ref().map(|m| m.fee).unwrap_or(0);
                    let ix_count = info.transaction
                        .as_ref()
                        .and_then(|t| t.message.as_ref())
                        .map(|m| m.instructions.len())
                        .unwrap_or(0);

                    println!(
                        "#{tx_count} slot={} sig={}... fee={fee} instructions={ix_count}",
                        tx.slot,
                        &sig[..16],
                    );
                }
            }
            Some(UpdateOneof::Slot(slot)) => {
                println!("slot {} ({:?})", slot.slot, slot.status);
            }
            _ => {}
        }
    }

    Ok(())
}

配合以下 grpc.yml：

transactions:
  pumpfun:
    vote: false
    failed: false
    account_include:
      - "6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P"

slots:
  all:
    filter_by_commitment: true

commitment: confirmed

​安装

​构建客户端

​构建器方法

​编写 YAML 配置

​YAML 过滤器参考

​订阅与读取事件

​自 YAML

​编程方式

​读取流

​关闭

​多路流

​完整示例

安装