Subscribe

const grpc = require('@grpc/grpc-js');
const protoLoader = require('@grpc/proto-loader');

const packageDef = protoLoader.loadSync('geyser.proto', {
  keepCase: true, longs: String, enums: String, defaults: true, oneofs: true
});
const proto = grpc.loadPackageDefinition(packageDef).geyser;
const client = new proto.Geyser('88.216.36.108:10001', grpc.credentials.createInsecure());

const stream = client.subscribe();

// Handle incoming updates
stream.on('data', (update) => {
  if (update.account) {
    console.log('[Account]', Buffer.from(update.account.account.pubkey).toString('hex'));
  } else if (update.transaction) {
    console.log('[Tx]', Buffer.from(update.transaction.transaction.signature).toString('hex'));
  } else if (update.slot) {
    console.log('[Slot]', update.slot.slot, update.slot.status);
  } else if (update.ping) {
    // Respond to ping with pong
    stream.write({ ping: { id: update.ping.id } });
  }
});

stream.on('error', (err) => {
  console.error('Stream error:', err);
  // Implement reconnection logic here
});

// Send subscription request -- monitor USDC account + all slots
stream.write({
  accounts: {
    usdc: {
      account: ['EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v'],
      owner: [],
      filters: []
    }
  },
  slots: {
    allSlots: { filter_by_commitment: false }
  },
  transactions: {},
  transactionsStatus: {},
  blocks: {},
  blocksMeta: {},
  commitment: 1,
  entry: {},
  accountsDataSlice: [],
  ping: null
});

use yellowstone_grpc_client::GeyserGrpcClient;
use yellowstone_grpc_proto::prelude::*;
use std::collections::HashMap;
use futures::StreamExt;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let mut client = GeyserGrpcClient::build_from_uri("http://88.216.36.108:10001")
        .connect()
        .await?;

    // Build subscription: accounts (USDC) + all slots
    let mut accounts = HashMap::new();
    accounts.insert("usdc".to_string(), SubscribeRequestFilterAccounts {
        account: vec!["EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v".to_string()],
        owner: vec![],
        filters: vec![],
        nonempty_txn_signature: None,
    });

    let mut slots = HashMap::new();
    slots.insert("allSlots".to_string(), SubscribeRequestFilterSlots {
        filter_by_commitment: Some(false),
        interleave: None,
    });

    let request = SubscribeRequest {
        accounts,
        slots,
        transactions: HashMap::new(),
        transactions_status: HashMap::new(),
        blocks: HashMap::new(),
        blocks_meta: HashMap::new(),
        commitment: Some(CommitmentLevel::Confirmed as i32),
        entry: HashMap::new(),
        accounts_data_slice: vec![],
        ping: None,
        from_slot: None,
    };

    let (_, mut stream) = client.subscribe_with_request(Some(request)).await?;

    while let Some(msg) = stream.next().await {
        match msg?.update_oneof {
            Some(UpdateOneof::Account(acct)) => {
                println!("Account update: {:?}", acct.account.map(|a| a.pubkey));
            }
            Some(UpdateOneof::Slot(slot)) => {
                println!("Slot: {} status: {:?}", slot.slot, slot.status);
            }
            Some(UpdateOneof::Ping(_)) => {
                println!("Received ping -- connection alive");
            }
            _ => {}
        }
    }

    Ok(())
}

{
  "filters": ["usdc"],
  "account": {
    "account": {
      "pubkey": "EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v",
      "lamports": "2039280",
      "owner": "TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA",
      "data": "...",
      "write_version": "12345678",
      "slot": "250000042"
    },
    "slot": "250000042",
    "is_startup": false
  }
}

{
  "filters": ["allSlots"],
  "slot": {
    "slot": "250000043",
    "parent": "250000042",
    "status": "CONFIRMED"
  }
}