ERPC

Geyser gRPC - Exemplos de código

Exemplos práticos em Rust para casos de uso comuns do Geyser gRPC. Informe o seu endpoint Geyser gRPC. O Solana Stream SDK oferece suporte ao Geyser gRPC.

Assinatura básica: todas as transações que não são de voto e não falharam

Esta assinatura de transações é intencionalmente ampla. Excluir as transações com falha reduz o tráfego, mas o stream restante ainda pode ser considerável. Adicione os filtros de conta, assinatura ou transação de que seu consumidor realmente precisa e mantenha o processamento fora do loop de recepção.
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(())
}

Cliente durável com reconexão limitada após encerramento limpo, recuperação de lacunas e ping/pong

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);
    }
}
Este exemplo conservador tenta novamente apenas falhas de estabelecimento de conexão e encerramentos limpos do stream. Ele para diante de um erro de assinatura ou stream gRPC após registrar o diagnóstico atual e guardar o primeiro. Examine a correção ERPC exibida e as informações de retry do servidor antes de tentar manualmente outra vez. O código de produção pode tentar explicitamente falhas transitórias conhecidas, mas deve respeitar uma pausa do servidor e não deve enviar repetidamente uma requisição rejeitada sem alterações.

Fazer stream das contas de um mint (contas de token)

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()
}

Fazer stream de transações (bem-sucedidas, que não são de voto)

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()
}

Destaques de um cliente de nível de produção

  • Cliente durável com reconexão e recuperação de lacunas (from_slot)
  • Tratamento de ping/pong para manter o stream ativo
  • Canais limitados/backpressure entre a entrada e o processamento
  • Atualizações dinâmicas de assinatura no mesmo stream bidi
  • Tratamento de erros, métricas e logging para consumidores lentos/atualizações descartadas

Padrão de cliente composto (recuperação de lacunas, ping/pong e assinaturas dinâmicas)

Esta composição mostra entrada limitada e assinaturas dinâmicas. A tarefa de entrada espera e reconecta apenas depois de uma falha de estabelecimento de conexão ou de um fim limpo do stream. Um erro de assinatura ou stream gRPC é informado e interrompe a tarefa para que a correção e as informações de retry possam ser examinadas; não é uma política genérica de retry.
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);
                }
            }
        }
    }
}

Contribuir com exemplos

Encontrou um padrão útil? Considere contribuir com ele para esta documentação.
Repositório do Solana Stream SDK: https://github.com/ValidatorsDAO/solana-stream
AnteriorFiltrosPróximoPerguntas frequentes