Subscribe

1const grpc = require('@grpc/grpc-js');
2const protoLoader = require('@grpc/proto-loader');
3
4const packageDef = protoLoader.loadSync('geyser.proto', {
5  keepCase: true, longs: String, enums: String, defaults: true, oneofs: true
6});
7const proto = grpc.loadPackageDefinition(packageDef).geyser;
8const client = new proto.Geyser('[IP_ADDRESS]:[PORT]', grpc.credentials.createInsecure());
9
10const stream = client.subscribe();
11
12// Handle incoming updates
13stream.on('data', (update) => {
14  if (update.account) {
15    console.log('[Account]', Buffer.from(update.account.account.pubkey).toString('hex'));
16  } else if (update.transaction) {
17    console.log('[Tx]', Buffer.from(update.transaction.transaction.signature).toString('hex'));
18  } else if (update.slot) {
19    console.log('[Slot]', update.slot.slot, update.slot.status);
20  } else if (update.ping) {
21    // Respond to ping with pong
22    stream.write({ ping: { id: update.ping.id } });
23  }
24});
25
26stream.on('error', (err) => {
27  console.error('Stream error:', err);
28  // Implement reconnection logic here
29});
30
31// Send subscription request -- monitor USDC account + all slots
32stream.write({
33  accounts: {
34    usdc: {
35      account: ['EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v'],
36      owner: [],
37      filters: []
38    }
39  },
40  slots: {
41    allSlots: { filter_by_commitment: false }
42  },
43  transactions: {},
44  transactionsStatus: {},
45  blocks: {},
46  blocksMeta: {},
47  commitment: 1,
48  entry: {},
49  accountsDataSlice: [],
50  ping: null
51});

1use yellowstone_grpc_client::GeyserGrpcClient;
2use yellowstone_grpc_proto::prelude::*;
3use std::collections::HashMap;
4use futures::StreamExt;
5
6#[tokio::main]
7async fn main() -> anyhow::Result<()> {
8    let mut client = GeyserGrpcClient::build_from_uri("http://[IP_ADDRESS]:[PORT]")
9        .connect()
10        .await?;
11
12    // Build subscription: accounts (USDC) + all slots
13    let mut accounts = HashMap::new();
14    accounts.insert("usdc".to_string(), SubscribeRequestFilterAccounts {
15        account: vec!["EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v".to_string()],
16        owner: vec![],
17        filters: vec![],
18        nonempty_txn_signature: None,
19    });
20
21    let mut slots = HashMap::new();
22    slots.insert("allSlots".to_string(), SubscribeRequestFilterSlots {
23        filter_by_commitment: Some(false),
24        interleave: None,
25    });
26
27    let request = SubscribeRequest {
28        accounts,
29        slots,
30        transactions: HashMap::new(),
31        transactions_status: HashMap::new(),
32        blocks: HashMap::new(),
33        blocks_meta: HashMap::new(),
34        commitment: Some(CommitmentLevel::Confirmed as i32),
35        entry: HashMap::new(),
36        accounts_data_slice: vec![],
37        ping: None,
38        from_slot: None,
39    };
40
41    let (_, mut stream) = client.subscribe_with_request(Some(request)).await?;
42
43    while let Some(msg) = stream.next().await {
44        match msg?.update_oneof {
45            Some(UpdateOneof::Account(acct)) => {
46                println!("Account update: {:?}", acct.account.map(|a| a.pubkey));
47            }
48            Some(UpdateOneof::Slot(slot)) => {
49                println!("Slot: {} status: {:?}", slot.slot, slot.status);
50            }
51            Some(UpdateOneof::Ping(_)) => {
52                println!("Received ping -- connection alive");
53            }
54            _ => {}
55        }
56    }
57
58    Ok(())
59}

1{
2  "filters": ["usdc"],
3  "account": {
4    "account": {
5      "pubkey": "EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v",
6      "lamports": "2039280",
7      "owner": "TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA",
8      "data": "...",
9      "write_version": "12345678",
10      "slot": "250000042"
11    },
12    "slot": "250000042",
13    "is_startup": false
14  }
15}

1{
2  "filters": ["allSlots"],
3  "slot": {
4    "slot": "250000043",
5    "parent": "250000042",
6    "status": "CONFIRMED"
7  }
8}