Struct BulkWsClient 

pub struct BulkWsClient { /* private fields */ }

Cloneable client handle.

• Hot reads (ticker, price, margin): watch::Receiver::borrow() — zero async overhead, just a ref-counted pointer swap.
• Writes (subscribe, place orders): go through the mpsc command channel to the actor, which serializes all mutations.
• Cold reads (open orders list): round-trip through the actor via oneshot — still fast, but async.

Implementations

impl BulkWsClient

pub async fn connect(config: WSConfig) -> Result<Self>

Connect to the exchange and spawn the actor task. Returns immediately once the WebSocket handshake succeeds.

Arguments

• config: web socket config

Examples found in repository

examples/md_listener.rs (lines 44-49)

async fn main() -> eyre::Result<()> {
    tracing_subscriber::fmt()
        .with_env_filter(
            EnvFilter::from_default_env().add_directive(tracing::Level::INFO.into())
        )
        .init();

    let args = Args::parse();

    info!("Connecting to {} for symbols: {:?}", args.url, args.symbols);
    let client = BulkWsClient::connect(WSConfig {
        url: args.url,
        symbols: args.symbols.clone(),
        signer: None,
        ..Default::default()
    })
        .await?;

    // ── Subscribe to L2 snapshots ───────────────────────────────────────
    if args.l2_levels > 0 {
        for sym in &args.symbols {
            client
                .subscribe_l2_snapshot(sym, Some(args.l2_levels))
                .await?;
        }
    }

    // ── Register event handlers ─────────────────────────────────────────
    client
        .on(Topic::Ticker, |event| {
            if let Event::Ticker(t) = event {
                info!(
                    "[TICKER] {:<10} mark={:1.2} last={:1.2} funding={:.6} vol={:.2}",
                    t.symbol, t.mark_price, t.last_price, t.funding_rate, t.volume,
                );
            }
        })
        .await;

    client
        .on(Topic::L2Snapshot, |event| {
            if let Event::L2Snapshot(book) = event {
                let (bids, asks) = &book.levels;
                let best_bid = bids.first().map(|l| l.price).unwrap_or(0.0);
                let best_ask = asks.first().map(|l| l.price).unwrap_or(0.0);
                let spread = best_ask - best_bid;
                info!(
                    "[L2]     {:<10} bid={:1.2} ask={:1.2} spread={:.2} levels={}x{}",
                    book.symbol,
                    best_bid,
                    best_ask,
                    spread,
                    bids.len(),
                    asks.len(),
                );
            }
        })
        .await;

    client
        .on(Topic::Error, |event| {
            if let Event::Error(e) = event {
                error!("[ERROR] {e}");
            }
        })
        .await;

    // ── Wait for Ctrl-C ─────────────────────────────────────────────────
    info!("Listening... press Ctrl-C to stop.");

    tokio::signal::ctrl_c().await?;
    info!("\nShutting down...");

    client.shutdown().await;
    process::exit(0);
}

pub async fn shutdown(&self)

Shut down the actor and close the WebSocket.

Examples found in repository

examples/md_listener.rs (line 107)

async fn main() -> eyre::Result<()> {
    tracing_subscriber::fmt()
        .with_env_filter(
            EnvFilter::from_default_env().add_directive(tracing::Level::INFO.into())
        )
        .init();

    let args = Args::parse();

    info!("Connecting to {} for symbols: {:?}", args.url, args.symbols);
    let client = BulkWsClient::connect(WSConfig {
        url: args.url,
        symbols: args.symbols.clone(),
        signer: None,
        ..Default::default()
    })
        .await?;

    // ── Subscribe to L2 snapshots ───────────────────────────────────────
    if args.l2_levels > 0 {
        for sym in &args.symbols {
            client
                .subscribe_l2_snapshot(sym, Some(args.l2_levels))
                .await?;
        }
    }

    // ── Register event handlers ─────────────────────────────────────────
    client
        .on(Topic::Ticker, |event| {
            if let Event::Ticker(t) = event {
                info!(
                    "[TICKER] {:<10} mark={:1.2} last={:1.2} funding={:.6} vol={:.2}",
                    t.symbol, t.mark_price, t.last_price, t.funding_rate, t.volume,
                );
            }
        })
        .await;

    client
        .on(Topic::L2Snapshot, |event| {
            if let Event::L2Snapshot(book) = event {
                let (bids, asks) = &book.levels;
                let best_bid = bids.first().map(|l| l.price).unwrap_or(0.0);
                let best_ask = asks.first().map(|l| l.price).unwrap_or(0.0);
                let spread = best_ask - best_bid;
                info!(
                    "[L2]     {:<10} bid={:1.2} ask={:1.2} spread={:.2} levels={}x{}",
                    book.symbol,
                    best_bid,
                    best_ask,
                    spread,
                    bids.len(),
                    asks.len(),
                );
            }
        })
        .await;

    client
        .on(Topic::Error, |event| {
            if let Event::Error(e) = event {
                error!("[ERROR] {e}");
            }
        })
        .await;

    // ── Wait for Ctrl-C ─────────────────────────────────────────────────
    info!("Listening... press Ctrl-C to stop.");

    tokio::signal::ctrl_c().await?;
    info!("\nShutting down...");

    client.shutdown().await;
    process::exit(0);
}

pub async fn closed(&self)

Wait for the actor to exit (e.g. on disconnect / error).

pub fn is_connected(&self) -> bool

Returns true if the WebSocket actor is still running.

This is a cheap, lock-free check — safe to call in hot loops. Once it returns false you must call BulkWsClient::connect again to establish a new connection.

pub fn get_ticker(&self, symbol: &str) -> Option<Ticker>

Get latest ticker for symbol, or None if not yet received.

Arguments

• symbol: symbol to retrieve for

Returns

• current ticker if available

pub fn get_price(&self, symbol: &str) -> Option<f64>

Get Latest mark price for symbol.

Arguments

• symbol: symbol to retrieve for

Returns

• current price if available

pub fn get_tickers(&self) -> HashMap<String, Ticker>

All current tickers, keyed by symbol.

Returns

• all current tickers

pub fn get_margin(&self) -> Margin

Get Current account margin.

pub fn get_position(&self, symbol: &str) -> Option<PositionInfo>

Get current position for symbol.

Arguments

• symbol: symbol to retrieve for

Returns

• current position for symbol if available

pub fn get_positions(&self) -> HashMap<String, PositionInfo>

Get all positions.

pub fn get_leverage(&self, symbol: &str) -> Option<f64>

Current leverage setting for symbol.

Arguments

• symbol: symbol to retrieve for

Returns

• current leverage if available

pub async fn wait_tickers_changed(&mut self) -> Result<HashMap<String, Ticker>>

Block until any ticker changes, then return the updated map.

pub async fn wait_account_changed(&mut self) -> Result<AccountState>

Block until account state changes (margin, positions, orders, leverage).

pub async fn open_orders(&self, symbol: Option<&str>) -> Result<Vec<OrderState>>

Open orders, optionally filtered by symbol.

Arguments

• symbol: optional symbol to retrieve for

Returns

• current orders or order status

pub async fn place_orders( &self, actions: Vec<Action>, account: Option<Pubkey>, nonce: Option<u64>, ) -> Result<Vec<Response>>

Place one or more actions (limit, market, cancel, cancel-all). Signs the bundle, sends through the actor, and awaits the exchange response with the configured timeout.

Arguments

• actions: list of orders, cancels, etc
• nonce: nonce to be used

Returns

• list of responses

pub async fn update_oracle( &self, actions: Vec<Price>, account: Option<Pubkey>, nonce: Option<u64>, ) -> Result<()>

Send oracle price updates

Arguments

• actions: list of oracle update
• nonce: nonce to be used

Returns

• list of responses

pub async fn place_limit_order( &self, symbol: &str, side: Side, price: f64, size: f64, tif: TimeInForce, reduce_only: bool, account: Option<Pubkey>, nonce: Option<u64>, ) -> Result<Response>

Place limit order

Arguments

• symbol: which market to execute in
• side: buy or sell
• price: limit price
• size: order size
• tif: time in force
• reduce_only: true if order is reduce only

Returns

• response for order placement

pub async fn place_market_order( &self, symbol: &str, side: Side, size: f64, reduce_only: bool, account: Option<Pubkey>, nonce: Option<u64>, ) -> Result<Response>

Place market order

Arguments

• symbol: which market to execute in
• side: buy or sell
• size: order size
• reduce_only: true if order is reduce only

Returns

• response for order placement

pub async fn cancel_order( &self, symbol: &str, order_id: &str, account: Option<Pubkey>, nonce: Option<u64>, ) -> Result<Response>

Cancel order

Arguments

• symbol: which market to execute in
• order_id: order ID to cancel

Returns

• response for order cancel

pub async fn cancel_all( &self, symbols: Vec<String>, account: Option<Pubkey>, nonce: Option<u64>, ) -> Result<Response>

Cancel all order

Arguments

• symbols: which symbols to cancel

Returns

• response for order cancel

pub fn subscribe_disconnect(&self) -> Receiver<String>

Subscribe to disconnect notifications.

The returned receiver fires exactly once, carrying the human-readable disconnect reason, when the actor exits for any reason (server close, network error, or explicit [shutdown]).

Use this as a poison pill for any tasks you spawned that should stop when the connection is lost:

let mut rx = client.subscribe_disconnect();
tokio::spawn(async move {
    let _ = rx.recv().await; // blocks until disconnect
    // clean up your task here
});

pub async fn subscribe_ticker(&self, symbol: &str) -> Result<()>

Subscribe to ticker for symbol.

Arguments

• symbol: symbol to subscrive to

pub async fn subscribe_trades(&self, symbols: &[&str]) -> Result<()>

Subscribe to fills for symbol.

Arguments

• symbols: list of symbol to subscribe to

pub async fn subscribe_l2_snapshot( &self, symbol: &str, nlevels: Option<u32>, ) -> Result<()>

Subscribe to L2 snapshots for symbol.

Arguments

• symbol: symbol to subscribe to

Examples found in repository

examples/md_listener.rs (line 56)

async fn main() -> eyre::Result<()> {
    tracing_subscriber::fmt()
        .with_env_filter(
            EnvFilter::from_default_env().add_directive(tracing::Level::INFO.into())
        )
        .init();

    let args = Args::parse();

    info!("Connecting to {} for symbols: {:?}", args.url, args.symbols);
    let client = BulkWsClient::connect(WSConfig {
        url: args.url,
        symbols: args.symbols.clone(),
        signer: None,
        ..Default::default()
    })
        .await?;

    // ── Subscribe to L2 snapshots ───────────────────────────────────────
    if args.l2_levels > 0 {
        for sym in &args.symbols {
            client
                .subscribe_l2_snapshot(sym, Some(args.l2_levels))
                .await?;
        }
    }

    // ── Register event handlers ─────────────────────────────────────────
    client
        .on(Topic::Ticker, |event| {
            if let Event::Ticker(t) = event {
                info!(
                    "[TICKER] {:<10} mark={:1.2} last={:1.2} funding={:.6} vol={:.2}",
                    t.symbol, t.mark_price, t.last_price, t.funding_rate, t.volume,
                );
            }
        })
        .await;

    client
        .on(Topic::L2Snapshot, |event| {
            if let Event::L2Snapshot(book) = event {
                let (bids, asks) = &book.levels;
                let best_bid = bids.first().map(|l| l.price).unwrap_or(0.0);
                let best_ask = asks.first().map(|l| l.price).unwrap_or(0.0);
                let spread = best_ask - best_bid;
                info!(
                    "[L2]     {:<10} bid={:1.2} ask={:1.2} spread={:.2} levels={}x{}",
                    book.symbol,
                    best_bid,
                    best_ask,
                    spread,
                    bids.len(),
                    asks.len(),
                );
            }
        })
        .await;

    client
        .on(Topic::Error, |event| {
            if let Event::Error(e) = event {
                error!("[ERROR] {e}");
            }
        })
        .await;

    // ── Wait for Ctrl-C ─────────────────────────────────────────────────
    info!("Listening... press Ctrl-C to stop.");

    tokio::signal::ctrl_c().await?;
    info!("\nShutting down...");

    client.shutdown().await;
    process::exit(0);
}

pub async fn subscribe_l2_delta(&self, symbol: &str) -> Result<()>

Subscribe to L2 deltas for symbol.

Arguments

• symbol: symbol to subscribe to

pub async fn subscribe_candles( &self, symbol: &str, interval: &str, ) -> Result<()>

Subscribe to candles for symbol.

Arguments

• symbol: symbol to subscribe to
• interval: bar period (“1min”, “5min”, …)

pub async fn on( &self, topic: Topic, handler: impl Fn(&Event) + Send + Sync + 'static, )

Register a callback for a topic. The callback runs synchronously inside the actor loop — keep it fast or tokio::spawn from within.

Argument

• topic: topic to subscribe to
• handler: callback for topic

Examples found in repository

examples/md_listener.rs (lines 63-70)

async fn main() -> eyre::Result<()> {
    tracing_subscriber::fmt()
        .with_env_filter(
            EnvFilter::from_default_env().add_directive(tracing::Level::INFO.into())
        )
        .init();

    let args = Args::parse();

    info!("Connecting to {} for symbols: {:?}", args.url, args.symbols);
    let client = BulkWsClient::connect(WSConfig {
        url: args.url,
        symbols: args.symbols.clone(),
        signer: None,
        ..Default::default()
    })
        .await?;

    // ── Subscribe to L2 snapshots ───────────────────────────────────────
    if args.l2_levels > 0 {
        for sym in &args.symbols {
            client
                .subscribe_l2_snapshot(sym, Some(args.l2_levels))
                .await?;
        }
    }

    // ── Register event handlers ─────────────────────────────────────────
    client
        .on(Topic::Ticker, |event| {
            if let Event::Ticker(t) = event {
                info!(
                    "[TICKER] {:<10} mark={:1.2} last={:1.2} funding={:.6} vol={:.2}",
                    t.symbol, t.mark_price, t.last_price, t.funding_rate, t.volume,
                );
            }
        })
        .await;

    client
        .on(Topic::L2Snapshot, |event| {
            if let Event::L2Snapshot(book) = event {
                let (bids, asks) = &book.levels;
                let best_bid = bids.first().map(|l| l.price).unwrap_or(0.0);
                let best_ask = asks.first().map(|l| l.price).unwrap_or(0.0);
                let spread = best_ask - best_bid;
                info!(
                    "[L2]     {:<10} bid={:1.2} ask={:1.2} spread={:.2} levels={}x{}",
                    book.symbol,
                    best_bid,
                    best_ask,
                    spread,
                    bids.len(),
                    asks.len(),
                );
            }
        })
        .await;

    client
        .on(Topic::Error, |event| {
            if let Event::Error(e) = event {
                error!("[ERROR] {e}");
            }
        })
        .await;

    // ── Wait for Ctrl-C ─────────────────────────────────────────────────
    info!("Listening... press Ctrl-C to stop.");

    tokio::signal::ctrl_c().await?;
    info!("\nShutting down...");

    client.shutdown().await;
    process::exit(0);
}

Trait Implementations

impl Clone for BulkWsClient

fn clone(&self) -> BulkWsClient

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.