Yellowstone gRPC Streams

import Client, {
  CommitmentLevel,
  SubscribeRequest,
} from "@triton-one/yellowstone-grpc";

const GRPC_URL = "your-geyser-grpc-endpoint";
const X_TOKEN = "your-api-token";
const PING_INTERVAL_MS = 30_000; // 30 seconds

async function main() {
  // 1. Open Connection
  const client = new Client(GRPC_URL, X_TOKEN, {
    "grpc.max_receive_message_length": 64 * 1024 * 1024, // 64MiB
  });

  // 2. Subscribe to Events
  const stream = await client.subscribe();

  // 3. Handle Stream Closure and Errors
  const streamClosed = new Promise<void>((resolve, reject) => {
    stream.on("error", (error) => {
      console.error("Stream error:", error);
      reject(error);
      stream.end(); // Ensure stream is closed on error
    });
    stream.on("end", () => {
      console.log("Stream ended.");
      resolve();
    });
    stream.on("close", () => {
      console.log("Stream closed.");
      resolve();
    });
  });

  // 4. Handle Incoming Data
  stream.on("data", (data) => {
    const ts = new Date().toUTCString();
    if (data.slot) { // Check if it's a slot update
      console.log(
        `${ts}: Received slot update: ${data.slot.slot}, Commitment: ${data.slot.status}`
      );
    } else if (data.pong) {
      console.log(`${ts}: Received pong (ping response id: ${data.pong.id})`);
    } else {
      // console.log(`${ts}: Received other data:`, data); // For debugging other message types
    }
  });

  // 5. Define Slot Subscription Request
  const slotRequest: SubscribeRequest = {
    slots: {
      // No specific slot filter here, will receive all based on commitment.
      // To filter by commitment on slot messages themselves:
      // slot: { filterByCommitment: true },
    },
    commitment: CommitmentLevel.CONFIRMED, // Requesting CONFIRMED slots

    // Other subscription types (set to empty objects if not used)
    accounts: {},
    accountsDataSlice: [],
    transactions: {},
    // transactionsStatus: {}, // Deprecated or handled by transaction filters
    blocks: {},
    blocksMeta: {},
    entry: {},
  };

  // 6. Send Subscribe Request
  try {
    await new Promise<void>((resolve, reject) => {
      stream.write(slotRequest, (err) => {
        if (err) {
          console.error("Failed to send slot subscription request:", err);
          reject(err);
        } else {
          console.log("Slot subscription request sent successfully.");
          resolve();
        }
      });
    });
  } catch (error) {
    console.error("Error in sending slot subscription:", error);
    client.close(); // Close client if initial subscription fails
    return;
  }


  // 7. Implement Ping to Keep Connection Alive
  const pingRequest: SubscribeRequest = {
    ping: { id: Math.floor(Math.random() * 1000000) }, // Use a unique ID for pings
    // All other filter fields must be present but can be empty
    accounts: {},
    accountsDataSlice: [],
    transactions: {},
    blocks: {},
    blocksMeta: {},
    entry: {},
    slots: {},
  };

  const pingInterval = setInterval(async () => {
    try {
      await new Promise<void>((resolve, reject) => {
        console.log(`${new Date().toUTCString()}: Sending ping (id: ${pingRequest.ping?.id})`);
        stream.write(pingRequest, (err) => {
          if (err) {
            console.error("Failed to send ping:", err);
            reject(err);
          } else {
            resolve();
          }
        });
      });
       // Update ping ID for next ping
      pingRequest.ping = { id: Math.floor(Math.random() * 1000000) };
    } catch (error) {
      console.error("Error sending ping:", error);
      // Consider logic to handle persistent ping failures (e.g., close stream, reconnect)
    }
  }, PING_INTERVAL_MS);

  // 8. Wait for Stream to Close
  try {
    await streamClosed;
  } catch (error) {
    console.error("Stream closed due to an error:", error);
  } finally {
    clearInterval(pingInterval); // Stop sending pings
    client.close(); // Close the gRPC client
    console.log("Client closed, ping interval cleared.");
  }
}

main().catch(console.error);