Rust SDK

The Rust SDK is a typed client for every /v1 REST endpoint and for the /v1/stream WebSocket. It consists of two crates, published from the metastreams-sdk repository:

Crate	Contents
`metastreams-types`	Every request and response of the API as serde types, without an HTTP or WebSocket dependency
`metastreams-sdk`	`MetastreamsProvider`, with one method per REST endpoint, and `StreamConnection`, with typed stream subscriptions. It re-exports the types crate as `metastreams_sdk::types`.

The API uses the same types, so they match its responses field for field.

Installation

The crates are not on crates.io yet. Add them as a git dependency pinned to a tag:

Cargo.toml

[dependencies]
metastreams-sdk = { git = "https://github.com/metastreams/metastreams-sdk", tag = "v0.1.0" }
tokio = { version = "1", features = ["rt-multi-thread", "macros"] }
futures = "0.3" # StreamExt, for subscriptions

To track a commit, pin a rev instead of a tag. The minimum supported Rust version is 1.93. TLS uses rustls with the webpki root certificates, so you don’t install a crypto provider yourself.

Provider

MetastreamsProvider::new(key) creates a client for the production API, and MetastreamsProvider::new_with_base_url(key, url) creates one for another origin. The provider implements Clone, and its Debug output hides the key. Each method sends one request, with a 30-second request timeout and a 10-second connect timeout. A method never retries.

use metastreams_sdk::{MetastreamsProvider, MultiChainAddress, SpotChain};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let provider = MetastreamsProvider::new(std::env::var("API_KEY")?)?;
    let bonk: MultiChainAddress = "DezXAZ8z7PnrnRJjz3wXBoRgixCa6xjnB7YaB1pPB263".parse()?;

    let token = provider.token_details(SpotChain::Solana, &bonk).await?;
    println!("{} {:?}", token.symbol, token.price.usd);
    Ok(())
}

The examples in the SDK repository read the key from METASTREAMS_API_KEY. The samples on this page read it from API_KEY, as the rest of these docs do.

Methods

Method	Endpoint	Returns
`token_details(chain, &address)`	`GET /v1/tokens/{chain}/{address}`	`UiSpotToken`
`token_ohlcv(chain, &address, &SpotCandlesQuery)`	`GET /v1/tokens/{chain}/{address}/ohlcv`	`UiCandlesResponse`
`token_trades(chain, &address, &TokenTradesQuery)`	`GET /v1/tokens/{chain}/{address}/trades`	`UiSpotTrades`
`token_holders(chain, &address, &TokenHoldersQuery)`	`GET /v1/tokens/{chain}/{address}/holders`	`UiTokenHolderList`
`identity_trades(&IdentityTradesQuery)`	`GET /v1/trades`	`UiSpotTrades`
`wallet_balances(&WalletBalancesQuery)`	`GET /v1/wallets/balances`	`UiWalletBalances`
`positions(&PositionsQuery)`	`GET /v1/positions`	`UiSpotPositions`
`native_prices(&NativePricesQuery)`	`GET /v1/native-prices`	`UiNativePrices`
`discovery_board(&SpotPulseSubscriptionConfig)`	`POST /v1/discovery`	`UiSpotPulseResponse`
`trending_board(&SpotTrendingFilter)`	`POST /v1/trending`	`UiSpotTrendingUpdate`
`session_token_create(&CreateSessionTokenRequest)`	`POST /v1/session-tokens`	`UiSessionToken`
`connect()`	`GET /v1/stream`	`StreamConnection`

When identity_trades, wallet_balances or positions leaves chains at None, the SDK sends every chain it knows.

Pagination

A paginated response carries has_more and next_cursor. Copy next_cursor into the query to read the next page:

use metastreams_sdk::types::trade::TokenTradesQuery;
use metastreams_sdk::{MetastreamsProvider, MultiChainAddress, SpotChain};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let provider = MetastreamsProvider::new(std::env::var("API_KEY")?)?;
    let bonk: MultiChainAddress = "DezXAZ8z7PnrnRJjz3wXBoRgixCa6xjnB7YaB1pPB263".parse()?;

    let mut query = TokenTradesQuery { limit: Some(100), ..Default::default() };
    loop {
        let page = provider.token_trades(SpotChain::Solana, &bonk, &query).await?;
        for trade in &page.trades {
            println!("{} {:?} {:?}", trade.id, trade.side, trade.volume.usd);
        }
        if !page.has_more {
            break;
        }
        query.cursor = page.next_cursor;
    }
    Ok(())
}

Session tokens

A backend creates a session token for a browser user with session_token_create:

use metastreams_sdk::MetastreamsProvider;
use metastreams_sdk::types::session::CreateSessionTokenRequest;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let provider = MetastreamsProvider::new(std::env::var("API_KEY")?)?;

    let request = CreateSessionTokenRequest {
        origins: vec!["https://app.example.com".into()],
        subject: Some("user-42".into()),
        ttl_seconds: Some(900),
    };
    let session = provider.session_token_create(&request).await?;
    println!("expires at {}", session.expires_at.0);
    Ok(())
}

Streams

provider.connect() opens one WebSocket, which carries many subscriptions. The filter type selects the channel and the payload types:

Filter	Channel	`Exit` payload
`TradeFilter`	`spot.trades`	None
`TransferFilter`	`spot.transfers`	None
`CandleFilter`	`spot.candles`	None
`TokenFilter`	`spot.tokens`	`UiTokenExit`
`WalletBalancesFilter`	`wallet.balances`	None
`SpotPositionsFilter`	`spot.positions`	None
`SpotHoldersFilter`	`spot.holders`	None
`NativePricesFilter`	`chain.nativePrices`	None
`SpotTrendingFilter`	`spot.trending`	None

TradeFilter, TransferFilter, CandleFilter and TokenFilter are in metastreams_sdk::types::filter. SpotTrendingFilter is in metastreams_sdk::types::trending, and the other filter types are in metastreams_sdk::types::stream. Each subscription is a futures::Stream of Events. An Event::Update carries one payload, and an Event::Exit means a token the subscription matched no longer matches. A channel without exits uses an empty exit type, so its Exit arm is match exit {}.

use futures::StreamExt;
use metastreams_sdk::types::filter::TradeFilter;
use metastreams_sdk::{Event, MetastreamsProvider, MultiChainAddress, SpotChain};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let provider = MetastreamsProvider::new(std::env::var("API_KEY")?)?;
    let bonk: MultiChainAddress = "DezXAZ8z7PnrnRJjz3wXBoRgixCa6xjnB7YaB1pPB263".parse()?;

    let connection = provider.connect().await?;
    let filter = TradeFilter {
        chains: vec![SpotChain::Solana],
        tokens: Some(vec![bonk]),
        ..Default::default()
    };
    let mut trades = connection.subscribe(&filter).await?;
    while let Some(event) = trades.next().await {
        match event? {
            Event::Update(trade) => println!("{:?} {} {}", trade.side, trade.amount, trade.symbol),
            Event::Exit(exit) => match exit {},
        }
    }
    Ok(())
}

Subscription options

subscribe_throttled(&filter, Duration) sets throttleMs on a TokenFilter or CandleFilter, rounded up to whole milliseconds. Exits are never throttled. subscribe_candles(&CandleFilter, CandleOptions) also sets the candle metric, denomination and closed_only.

Subscription lifecycle

The stream sends no snapshot, except on spot.trending, whose first update is the full board. On other channels, read the REST endpoint after subscribe returns, as Initial snapshot describes. id() returns the subscription ID the server assigned, and dropping a subscription unsubscribes it. A subscription ends with one Err(MetastreamsError::Closed(reason)), followed by None. A payload that fails to decode yields Err(MetastreamsError::FrameDecode { .. }), and the subscription continues. connection.closed() resolves to the CloseReason when the socket closes. connection.close() closes the socket normally, and each subscription then ends with CloseReason::Client. The SDK sends a ping every 15 seconds and treats the socket as lost after 45 seconds without a frame. Each subscription buffers up to 1,024 unread events. When the buffer is full, the SDK closes that subscription with CloseReason::Lagged and leaves the others open.

Close reasons

`CloseReason`	Cause	Transient
`ServerRestarting`	Close code `1001`	Yes
`SlowConsumer`	Close code `4008`	Yes
`KeyRevoked`	Close code `1008`	No
`CreditsExhausted`	Close code `4402`	No
`Lagged`	The subscription’s buffer filled	Yes
`Lost(LostCause)`	The socket ended without a close code, or went silent	Yes
`Other { code, reason }`	A close code this SDK version does not name	Yes
`Client`	You closed the connection	Yes

Reconnection

The SDK does not reconnect. Subscriptions end with their connection, and missed updates are not replayed. After a transient close, connect, subscribe again and re-read the snapshot from REST. Reconnect at once after ServerRestarting, and back off after other transient reasons. After KeyRevoked or CreditsExhausted, stop. See Reconnection.

Errors

Every method returns metastreams_sdk::Result<T>, whose error is MetastreamsError:

Variant	Cause
`Api { status, error, request_id }`	The API refused the request with an error envelope. Match on `error.code`, never on `error.message`.
`RateLimited { retry_after, request_id }`	A `429`, including a stream handshake past the connection limit, with the `Retry-After` wait
`Rejected { channel, error }`	The server rejected a subscribe request with an `error` frame
`Closed(CloseReason)`	A subscription ended
`FrameDecode { channel, source }`	A stream payload did not decode
`Http`, `WebSocket`	A transport failure
`Decode`, `Unexpected`	A response body the SDK could not read
`Url`, `UnsupportedScheme`, `InvalidKey`	The base URL or the key cannot be used

error.code is an ErrorCode from metastreams_sdk::types::error. A code that this SDK version does not know becomes ErrorCode::Unknown. is_transient() reports whether the same call can succeed later: a transport failure, a 429, a 5xx, or a transient close. The example below retries transient errors and waits for Retry-After on a 429:

use std::time::Duration;

use metastreams_sdk::types::token::UiSpotToken;
use metastreams_sdk::{MetastreamsError, MetastreamsProvider, MultiChainAddress, SpotChain};

async fn token_with_retries(
    provider: &MetastreamsProvider,
    address: &MultiChainAddress,
    attempts: u32,
) -> metastreams_sdk::Result<UiSpotToken> {
    let mut attempt = 0;
    loop {
        match provider.token_details(SpotChain::Solana, address).await {
            Err(error) if error.is_transient() && attempt + 1 < attempts => {
                let delay = match &error {
                    MetastreamsError::RateLimited { retry_after: Some(wait), .. } => *wait,
                    _ => Duration::from_millis((1000 * 2u64.pow(attempt.min(5))).min(30_000)),
                };
                tokio::time::sleep(delay).await;
                attempt += 1;
            }
            result => return result,
        }
    }
}

The sample also needs the time feature of tokio.

Get started

Core concepts

Streams

Recipes

SDKs

Migration guides

Resources

Installation

Provider

Methods

Session tokens

Streams

Subscription options

Subscription lifecycle

Close reasons

Reconnection

Errors

​Installation

​Provider

​Methods

​Pagination

​Session tokens

​Streams

​Subscription options

​Subscription lifecycle

​Close reasons

​Reconnection

​Errors

Installation

Provider

Methods

Pagination

Session tokens

Streams

Subscription options

Subscription lifecycle

Close reasons

Reconnection

Errors