ERPC

Geyser gRPC - Kod Örnekleri

Yaygın Geyser gRPC kullanım senaryoları için pratik Rust örnekleri. Kendi Geyser gRPC endpoint adresinizi girin. Solana Stream SDK, Geyser gRPC desteği sunar.

Temel abonelik: tüm oy dışı, başarısız olmayan işlemler

Bu, bilerek geniş tutulmuş bir işlem aboneliğidir. Başarısız işlemleri hariç tutmak trafiği azaltır, ancak kalan stream yine de oldukça büyük olabilir. Tüketicinizin gerçekten ihtiyaç duyduğu işe uyan hesap, imza veya işlem filtrelerini ekleyin ve işlemeyi alma döngüsünün dışında tutun.
rust
use anyhow::Result;
use futures::{sink::SinkExt, stream::StreamExt};
use solana_signature::Signature;
use std::collections::HashMap;
use solana_stream_sdk::{
    GeyserCommitmentLevel as CommitmentLevel, GeyserGrpcClient,
    GeyserSubscribeRequest as SubscribeRequest,
    GeyserSubscribeRequestFilterTransactions as SubscribeRequestFilterTransactions,
    GeyserUpdateOneof as UpdateOneof,
};

#[tokio::main]
async fn main() -> Result<()> {
    let endpoint = "http://your-erpc-grpc-endpoint"; // Set your ERPC gRPC URL

    let mut client = GeyserGrpcClient::build_from_shared(endpoint)?
        .connect()
        .await?;

    let (mut tx, mut stream) = client.subscribe().await?;

    tx.send(SubscribeRequest {
        transactions: HashMap::from([(
            "all_transactions".to_string(),
            SubscribeRequestFilterTransactions {
                vote: Some(false),   // Drop vote txs
                failed: Some(false), // Drop failed txs
                ..Default::default()
            },
        )]),
        commitment: Some(CommitmentLevel::Confirmed as i32),
        ..Default::default()
    })
    .await?;

    while let Some(Ok(msg)) = stream.next().await {
        if let Some(UpdateOneof::Transaction(tx)) = msg.update_oneof {
            let sig = tx
                .transaction
                .and_then(|t| Signature::try_from(t.signature.as_slice()).ok())
                .unwrap_or_default();
            println!("Slot: {} | Signature: {}", tx.slot, sig);
        }
    }
    Ok(())
}

Temiz kapanış sonrasında sınırlı yeniden bağlanma, boşluk kurtarma ve ping/pong içeren dayanıklı istemci

rust
use anyhow::Result;
use futures::stream::StreamExt;
use std::{collections::HashMap, time::Duration};
use solana_stream_sdk::{
    GeyserCommitmentLevel as CommitmentLevel, GeyserGrpcClient,
    GeyserSlotStatus as SlotStatus, GeyserSubscribeRequest as SubscribeRequest,
    GeyserSubscribeRequestFilterSlots as SubscribeRequestFilterSlots,
    GeyserUpdateOneof as UpdateOneof,
};

#[tokio::main]
async fn main() -> Result<()> {
    let endpoint = "http://your-erpc-grpc-endpoint";

    let mut subscribe_request = SubscribeRequest {
        slots: HashMap::from([(
            "slots".to_string(),
            SubscribeRequestFilterSlots::default(),
        )]),
        commitment: Some(CommitmentLevel::Confirmed as i32),
        ..Default::default()
    };

    let mut tracked_slot: u64 = 0;
    let mut retry_delay = Duration::from_secs(1);
    let max_retry_delay = Duration::from_secs(30);
    let mut first_diagnostic: Option<String> = None;

    loop {
        let mut client = match GeyserGrpcClient::build_from_shared(endpoint)?.connect().await {
            Ok(client) => client,
            Err(error) => {
                let current_diagnostic = error.to_string();
                first_diagnostic.get_or_insert_with(|| current_diagnostic.clone());
                eprintln!("Connection error: {current_diagnostic}");
                tokio::time::sleep(retry_delay).await;
                retry_delay = (retry_delay * 2).min(max_retry_delay);
                continue;
            }
        };

        subscribe_request.from_slot = if tracked_slot > 0 {
            Some(tracked_slot)
        } else {
            None
        };

        let (mut _sink, mut stream) = match client
            .subscribe_with_request(Some(subscribe_request.clone()))
            .await
        {
            Ok(subscription) => subscription,
            Err(error) => {
                let current_diagnostic = error.to_string();
                first_diagnostic.get_or_insert_with(|| current_diagnostic.clone());
                eprintln!("Subscribe error: {current_diagnostic}");
                eprintln!("First diagnostic retained: {}", first_diagnostic.as_deref().unwrap());
                return Err(error.into());
            }
        };

        loop {
            match stream.next().await {
                Some(Ok(msg)) => {
                    retry_delay = Duration::from_secs(1);
                    if let Some(UpdateOneof::Slot(slot)) = msg.update_oneof {
                        tracked_slot = slot.slot;
                        println!(
                            "Slot: {} | Status: {:?}",
                            slot.slot,
                            SlotStatus::try_from(slot.status).unwrap()
                        );
                    }
                }
                Some(Err(error)) => {
                    let current_diagnostic = error.to_string();
                    first_diagnostic.get_or_insert_with(|| current_diagnostic.clone());
                    eprintln!("Stream error: {current_diagnostic}");
                    eprintln!("First diagnostic retained: {}", first_diagnostic.as_deref().unwrap());
                    return Err(error.into());
                }
                None => break,
            }
        }

        tokio::time::sleep(retry_delay).await;
        retry_delay = (retry_delay * 2).min(max_retry_delay);
    }
}
Bu temkinli örnek yalnızca bağlantı kurma hatalarında ve temiz stream kapanışlarında yeniden dener. gRPC subscribe veya stream hatasında, o anki tanılamayı log'a yazıp ilk tanılamayı sakladıktan sonra durur. Elle yeniden denemeden önce gösterilen ERPC düzeltmesini ve varsa sunucunun yeniden deneme bilgisini inceleyin. Üretim kodu bilinen geçici hataları açıkça yeniden deneyebilir; ancak sunucunun duraklatmasına uymalı ve reddedilmiş, değiştirilmemiş bir isteği tekrar tekrar göndermemelidir.

Bir mint için hesapları akışa al (token hesapları)

rust
SubscribeRequest {
    accounts: HashMap::from([(
        "token_accounts_by_mint".to_string(),
        SubscribeRequestFilterAccounts {
            owner: vec!["TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA".to_string()],
            filters: vec![
                SubscribeRequestFilterAccountsFilter {
                    filter: Some(Filter::TokenAccountState(true)),
                },
                SubscribeRequestFilterAccountsFilter {
                    filter: Some(Filter::Memcmp(SubscribeRequestFilterAccountsFilterMemcmp {
                        offset: 0,
                        data: Some(Data::Base58(
                            "pumpCmXqMfrsAkQ5r49WcJnRayYRqmXz6ae8H7H9Dfn".to_string(), // replace with mint
                        )),
                    })),
                },
            ],
            ..Default::default()
        },
    )]),
    commitment: Some(CommitmentLevel::Confirmed as i32),
    ..Default::default()
}

İşlemleri akışa al (başarılı, oy dışı)

rust
SubscribeRequest {
    transactions: HashMap::from([(
        "all_transactions".to_string(),
        SubscribeRequestFilterTransactions {
            vote: Some(false),
            failed: Some(false),
            ..Default::default()
        },
    )]),
    commitment: Some(CommitmentLevel::Confirmed as i32),
    ..Default::default()
}

Üretim seviyesinde istemcinin öne çıkan özellikleri

  • Yeniden bağlanma ve boşluk kurtarma içeren dayanıklı istemci (from_slot)
  • Akışı canlı tutmak için ping/pong yönetimi
  • Ingress ile işleme arasında sınırlı kanallar/geri basınç
  • Aynı bidi akışında dinamik abonelik güncellemeleri
  • Yavaş tüketiciler/atılan güncellemeler için hata yönetimi, metrikler ve loglama

Birleşik istemci kalıbı (boşluk kurtarma, ping/pong, dinamik abonelikler)

Bu birleşim, sınırlı veri alımını ve dinamik abonelikleri gösterir. Veri alım görevi yalnızca bir bağlantı kurma hatasından veya stream'in temiz biçimde sona ermesinden sonra bekler ve yeniden bağlanır. gRPC abonelik veya stream hatası raporlanır ve görev durur; böylece düzeltme ve yeniden deneme bilgisi incelenebilir. Bu genel bir yeniden deneme politikası değildir.
rust
use futures::{SinkExt, stream::StreamExt};
use solana_signature::Signature;
use std::{collections::HashMap, sync::Arc, time::Duration};
use tokio::sync::{RwLock, mpsc};
use solana_stream_sdk::{
    GeyserCommitmentLevel as CommitmentLevel, GeyserGrpcClient, GeyserSlotStatus,
    GeyserSubscribeRequest as SubscribeRequest,
    GeyserSubscribeRequestFilterSlots as SubscribeRequestFilterSlots,
    GeyserSubscribeRequestFilterTransactions as SubscribeRequestFilterTransactions,
    GeyserSubscribeUpdate as SubscribeUpdate, GeyserUpdateOneof as UpdateOneof,
};
use solana_stream_sdk::yellowstone_grpc_proto::geyser::SubscribeRequestPing;

const UPDATE_CHANNEL_CAPACITY: usize = 10_000;

#[tokio::main]
async fn main() {
    let endpoint = "http://your-erpc-grpc-endpoint".to_string();

    let subscribe_request = Arc::new(RwLock::new(SubscribeRequest {
        slots: HashMap::from([(
            "slots".to_string(),
            SubscribeRequestFilterSlots {
                ..Default::default()
            },
        )]),
        transactions: HashMap::from([(
            "all_transactions".to_string(),
            SubscribeRequestFilterTransactions {
                vote: Some(false),
                failed: Some(false),
                ..Default::default()
            },
        )]),
        commitment: Some(CommitmentLevel::Confirmed as i32),
        ..Default::default()
    }));

    let (updates_tx, updates_rx) = mpsc::channel::<SubscribeUpdate>(UPDATE_CHANNEL_CAPACITY);
    let (subscribe_tx, subscribe_rx) = mpsc::unbounded_channel::<SubscribeRequest>();

    let dispatcher_handle = tokio::spawn(async move {
        dispatcher_loop(updates_rx).await;
    });

    let ingress_handle = tokio::spawn(async move {
        ingress_loop(
            updates_tx,
            subscribe_rx,
            endpoint,
            subscribe_request,
        )
        .await;
    });

    if let Err(e) = tokio::join!(dispatcher_handle, ingress_handle).0 {
        println!("Error: {:?}", e);
    }
}

async fn ingress_loop(
    updates_tx: mpsc::Sender<SubscribeUpdate>,
    mut subscribe_rx: mpsc::UnboundedReceiver<SubscribeRequest>,
    endpoint: String,
    subscribe_request: Arc<RwLock<SubscribeRequest>>,
) {
    let mut tracked_slot: u64 = 0;
    let mut retry_delay = Duration::from_secs(1);
    let max_retry_delay = Duration::from_secs(30);

    loop {
        let builder = match GeyserGrpcClient::build_from_shared(endpoint.clone()) {
            Ok(builder) => builder,
            Err(error) => {
                eprintln!("Client configuration error: {error}");
                return;
            }
        };

        let mut client = match builder.connect().await {
            Ok(client) => client,
            Err(error) => {
                eprintln!("Connection error: {error}");
                tokio::time::sleep(retry_delay).await;
                retry_delay = (retry_delay * 2).min(max_retry_delay);
                continue;
            }
        };

        subscribe_request.write().await.from_slot = if tracked_slot > 0 {
            Some(tracked_slot)
        } else {
            None
        };

        match client
            .subscribe_with_request(Some(subscribe_request.read().await.clone()))
            .await
        {
            Ok((mut subscribe_tx, mut stream)) => {
                loop {
                    tokio::select! {
                        subscription = subscribe_rx.recv() => {
                            match subscription {
                                Some(subscribe_request) => {
                                    if let Err(error) = subscribe_tx.send(subscribe_request).await {
                                        eprintln!("Subscription update error: {error}");
                                        return;
                                    }
                                }
                                None => return,
                            }
                        }
                        result = stream.next() => {
                            match result {
                                Some(Ok(update)) => {
                                    retry_delay = Duration::from_secs(1);
                                    if matches!(update.update_oneof, Some(UpdateOneof::Ping(_))) {
                                        if let Err(error) = subscribe_tx
                                            .send(SubscribeRequest {
                                                ping: Some(SubscribeRequestPing { id: 1 }),
                                                ..Default::default()
                                            })
                                            .await
                                        {
                                            eprintln!("Ping response error: {error}");
                                            return;
                                        }
                                    }
                                    if matches!(update.update_oneof, Some(UpdateOneof::Ping(_)) | Some(UpdateOneof::Pong(_))) {
                                        continue;
                                    }
                    if let Some(UpdateOneof::Slot(ref slot_update)) = update.update_oneof {
                        tracked_slot = slot_update.slot;
                        continue;
                    }
                                    if let Err(error) = updates_tx.send(update.clone()).await {
                                        eprintln!("Update consumer stopped: {error}");
                                        return;
                                    }
                                }
                                Some(Err(error)) => {
                                    eprintln!("Stream error: {error}");
                                    return;
                                }
                                None => {
                                    println!("Stream ended cleanly");
                                    break;
                                }
                            }
                        }
                    }
                }
            }
            Err(error) => {
                eprintln!("Subscribe error: {error}");
                return;
            }
        };

        println!("Disconnected; retrying after bounded backoff...");
        tokio::time::sleep(retry_delay).await;
        retry_delay = (retry_delay * 2).min(max_retry_delay);
    }
}

async fn dispatcher_loop(mut updates_rx: mpsc::Receiver<SubscribeUpdate>) {
    loop {
        if let Some(update) = updates_rx.recv().await {
            match update.update_oneof {
                Some(UpdateOneof::Transaction(tx)) => {
                    let sig = tx
                        .transaction
                        .and_then(|t| Signature::try_from(t.signature.as_slice()).ok())
                        .unwrap_or_default();
                    println!("Slot: {} | Signature: {}", tx.slot, sig);
                }
                _ => {
                    println!("Received update: {:?}", update);
                }
            }
        }
    }
}

Örneklere katkıda bulunma

Kullanışlı bir desen mi buldunuz? Bunu bu dokümanlara katkı olarak eklemeyi düşünün.
Solana Stream SDK deposu: https://github.com/ValidatorsDAO/solana-stream
ÖncekiFiltrelerSonrakiSSS