Subscribe

use yellowstone_grpc_client::GeyserGrpcClient;
use yellowstone_grpc_proto::prelude::{
    SubscribeRequest, SubscribeRequestFilterSlots,
    SubscribeRequestFilterTransactions, CommitmentLevel,
};
use futures::stream::StreamExt;
use std::collections::HashMap;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Connect to the gRPC endpoint
    let mut client = GeyserGrpcClient::build_from_shared("http://127.0.0.1:10000")?
        .x_token(Some("your-token-here"))?
        .connect()
        .await?;

    // Create subscription request
    let mut slots = HashMap::new();
    slots.insert(
        "client".to_string(),
        SubscribeRequestFilterSlots {
            filter_by_commitment: Some(true),
            ..Default::default()
        },
    );

    let mut transactions = HashMap::new();
    transactions.insert(
        "client".to_string(),
        SubscribeRequestFilterTransactions {
            vote: Some(false),
            failed: Some(false),
            account_include: vec!["TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA".to_string()],
            ..Default::default()
        },
    );

    let request = SubscribeRequest {
        slots,
        transactions,
        commitment: Some(CommitmentLevel::Confirmed as i32),
        ..Default::default()
    };

    // Subscribe and receive updates
    let (mut subscribe_tx, mut stream) = client
        .subscribe_with_request(Some(request))
        .await?;

    println!("Subscription started, receiving updates...");
    
    while let Some(message) = stream.next().await {
        match message {
            Ok(msg) => {
                println!("Received update: filters={:?}", msg.filters);
                // Handle different update types
                if let Some(update) = msg.update_oneof {
                    // Process the update
                }
            }
            Err(error) => {
                eprintln!("Error: {:?}", error);
                break;
            }
        }
    }

    Ok(())
}