Перейти к основному содержанию

​
Установка

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) — управление backoff при переподключении. Когда соединение падает, SDK ждёт initial_delay, затем удваивает задержку на каждой попытке (ограничение max_delay). max_attempts: 0 — бесконечные повторы. По умолчанию: 100 мс начальная, 30 с максимум, множитель 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 отправляет на уровне proto сообщения 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. 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