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<Vec<Response>>

Send oracle price updates and return the exchange’s responses.

Waits for the exchange to acknowledge the transaction so that any rejection (e.g. invalid price, authorisation failure) is surfaced to the caller rather than silently dropped.

Arguments

• actions: list of oracle price updates
• account: optional override for the signing account
• nonce: optional nonce override; a fresh one is generated when None

Returns

• Ok(responses) — one Response per submitted price; callers should inspect each entry with Response::is_error to detect rejections.
• Err(_) — transport-level failure (send error, timeout, dropped channel).

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.