# How to stream with RabbitStream
RabbitStream is designed to be a seamless, drop-in replacement for your existing Yellowstone gRPC clients, requiring only a change of endpoint URL.
{% hint style="info" %}
#### Beta-access already available for everyone with a Shyft gRPC token.
{% endhint %}
### Access and Authentication
RabbitStream requires a valid Shyft gRPC token. gRPC access is included with our *Build, Grow*, and *Accelerate* plans. Once you have your token, you can access the stream using the following regional endpoints:
| **Ams** (Amsterdam) | `https://rabbitstream.ams.shyft.to/` |
| -------------------------- | ------------------------------------ |
| **VA** (Virginia, Ashburn) | `https://rabbitstream.va.shyft.to/` |
| NY (New York) | `https://rabbitstream.ny.shyft.to/` |
| **Fra** (Frankfurt) | `https://rabbitstream.fra.shyft.to/` |
### Connecting to RabbitStream: Client Examples
Once you have your Shyft gRPC Access Token, you can connect to Rabbitstream in the following manner:
{% tabs %}
{% tab title="TypeScript" %}
```typescript
import "dotenv/config";
import Client, {
CommitmentLevel,
SubscribeRequest,
SubscribeRequestAccountsDataSlice,
} from "@triton-one/yellowstone-grpc";
const client = new Client(
"https://rabbitstream.ams.shyft.to/",
process.env.X_TOKEN,
undefined
);
const req: SubscribeRequest = {
accounts: {},
slots: {},
transactions: {
pumpFun: {
vote: false,
failed: false,
signature: undefined,
accountInclude: ["6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P"],
accountExclude: [],
accountRequired: [],
},
},
transactionsStatus: {},
entry: {},
blocks: {},
blocksMeta: {},
accountsDataSlice: [] as SubscribeRequestAccountsDataSlice[],
ping: undefined,
commitment: CommitmentLevel.PROCESSED,
};
async function handleStream(client: Client, args: SubscribeRequest) {
// Subscribe for events
console.log(`Subscribing and starting stream...`);
const stream = await client.subscribe();
// Create `error` / `end` handler
const streamClosed = new Promise((resolve, reject) => {
stream.on("error", (error) => {
console.log("ERROR", error);
reject(error);
stream.end();
});
stream.on("end", () => {
resolve();
});
stream.on("close", () => {
resolve();
});
});
// Handle updates
stream.on("data", (data) => {
console.log("Received data....");
console.dir(data, { depth: null });
});
// Send subscribe request
await new Promise((resolve, reject) => {
stream.write(args, (err: any) => {
if (err === null || err === undefined) {
resolve();
} else {
reject(err);
}
});
}).catch((reason) => {
console.error(reason);
throw reason;
});
await streamClosed;
}
async function subscribeCommand(client: Client, args: SubscribeRequest) {
while (true) {
try {
await handleStream(client, args);
} catch (error) {
console.error("Stream error, restarting in 1 second...", error);
await new Promise((resolve) => setTimeout(resolve, 1000));
}
}
}
subscribeCommand(client, req);
```
{% endtab %}
{% tab title="Rust" %}
```rust
use {
backoff::{future::retry, ExponentialBackoff},
clap::Parser as ClapParser,
futures::{
future::TryFutureExt,
stream::StreamExt,
},
log::{error, info},
std::{collections::HashMap, env, sync::Arc, time::Duration},
tokio::sync::Mutex,
tonic::transport::channel::ClientTlsConfig,
yellowstone_grpc_client::{GeyserGrpcClient, Interceptor},
yellowstone_grpc_proto::{
geyser::SubscribeRequestFilterTransactions,
prelude::{ CommitmentLevel, SubscribeRequest},
},
};
type TxnFilterMap = HashMap;
const PUMPFUN_AMM_PROGRAM_ID: &str = "pAMMBay6oceH9fJKBRHGP5D4bD4sWpmSwMn52FMfXEA";
#[derive(Debug, Clone, ClapParser)]
#[clap(author, version, about)]
struct Args {
#[clap(short, long, help = "gRPC endpoint")]
endpoint: String,
#[clap(long, help = "X-Token")]
x_token: String,
}
impl Args {
async fn connect(&self) -> anyhow::Result> {
GeyserGrpcClient::build_from_shared(self.endpoint.clone())?
.x_token(Some(self.x_token.clone()))?
.connect_timeout(Duration::from_secs(10))
.timeout(Duration::from_secs(10))
.tls_config(ClientTlsConfig::new().with_native_roots())?
.max_decoding_message_size(1024 * 1024 * 1024)
.connect()
.await
.map_err(Into::into)
}
pub fn get_txn_updates(&self) -> anyhow::Result {
let mut transactions: TxnFilterMap = HashMap::new();
transactions.insert(
"client".to_owned(),
SubscribeRequestFilterTransactions {
vote: Some(false),
failed: Some(false),
account_include: vec![PUMPFUN_AMM_PROGRAM_ID.to_string()],
account_exclude: vec![],
account_required: vec![],
signature: None,
},
);
Ok(SubscribeRequest {
accounts: HashMap::default(),
slots: HashMap::default(),
transactions,
transactions_status: HashMap::default(),
blocks: HashMap::default(),
blocks_meta: HashMap::default(),
entry: HashMap::default(),
commitment: Some(CommitmentLevel::Processed as i32),
accounts_data_slice: Vec::default(),
ping: None,
from_slot: None,
})
}
}
#[tokio::main]
async fn main() -> anyhow::Result<()> {
env::set_var(
env_logger::DEFAULT_FILTER_ENV,
env::var_os(env_logger::DEFAULT_FILTER_ENV).unwrap_or_else(|| "info".into()),
);
env_logger::init();
let args = Args::parse();
let zero_attempts = Arc::new(Mutex::new(true));
retry(ExponentialBackoff::default(), move || {
let args = args.clone();
let zero_attempts = Arc::clone(&zero_attempts);
async move {
let mut zero_attempts = zero_attempts.lock().await;
if *zero_attempts {
*zero_attempts = false;
} else {
info!("Retry to connect to the server");
}
drop(zero_attempts);
let client = args.connect().await.map_err(backoff::Error::transient)?;
info!("Connected");
let request = args.get_txn_updates().map_err(backoff::Error::Permanent)?;
geyser_subscribe(client, request)
.await
.map_err(backoff::Error::transient)?;
Ok::<(), backoff::Error>(())
}
.inspect_err(|error| error!("failed to connect: {error}"))
})
.await
.map_err(Into::into)
}
async fn geyser_subscribe(
mut client: GeyserGrpcClient,
request: SubscribeRequest,
) -> anyhow::Result<()> {
let ( subscribe_tx, mut stream) = client.subscribe_with_request(Some(request)).await?;
info!("stream opened");
while let Some(message) = stream.next().await {
match message {
Ok(msg) => println!("Received Message: {:#?}", msg),
Err(e) => error!("Failed to receive message: {e}"),
}
}
info!("stream closed");
Ok(())
}
```
{% endtab %}
{% endtabs %}
### Performance Benchmarks: The Speed Advantage
RabbitStream is ideal for applications that need low-latency transaction monitoring, such as *DeFi dashboards, arbitrage bots, or token analytics tools*. We compared the detection latency between RabbitStream and Yellowstone gRPC. All the code for this comparison tool is available on GitHub for you to try out.
