Struct StreamerClient 

pub struct StreamerClient { /* private fields */ }

Streaming client for IG Markets real-time data.

This client manages two Lightstreamer connections for different data types:

• Market streamer: Handles market data (prices, market state), trade updates (CONFIRMS, OPU, WOU), and account updates (positions, orders, balance). Uses the default adapter.
• Price streamer: Handles detailed price data (bid/ask levels, sizes, multiple currencies). Uses the “Pricing” adapter.

Each connection type can be managed independently and runs in parallel.

Implementations

impl StreamerClient

pub async fn new() -> Result<Self, AppError>

Creates a new streaming client instance with its own REST session.

This builds a fresh Client, logs in to obtain the Lightstreamer connection details, and initializes both streaming clients (market and price). No connection is established yet — that happens on connect().

When the caller already holds a Client with an active REST session, prefer with_client to reuse that session instead of performing a second login.

Errors

Returns AppError if the login / session lookup or Lightstreamer client initialization fails.

pub async fn with_client(client: &Client) -> Result<Self, AppError>

Creates a new streaming client that reuses the caller’s existing REST session.

Unlike new, this does not build a second HTTP client or perform a second login: it reuses client’s cached session (via Client::ws_info) to obtain the Lightstreamer endpoint and credentials. Both streaming clients (market and price) are initialized but no connection is established until connect() is called.

Errors

Returns AppError if the session lookup or Lightstreamer client initialization fails.

pub async fn market_subscribe( &mut self, epics: Vec<String>, fields: HashSet<StreamingMarketField>, ) -> Result<UnboundedReceiver<PriceData>, AppError>

Subscribes to market data updates for the specified instruments.

This method creates a subscription to receive real-time market data updates for the given EPICs and returns a channel receiver for consuming the updates.

Arguments

• epics - List of instrument EPICs to subscribe to
• fields - Set of market data fields to receive (e.g., BID, OFFER, etc.)

Returns

Returns a receiver channel for PriceData updates, or an error if the subscription setup failed.

Examples

let mut receiver = client.market_subscribe(
    vec!["IX.D.DAX.DAILY.IP".to_string()],
    fields
).await?;

tokio::spawn(async move {
    while let Some(price_data) = receiver.recv().await {
        println!("Price update: {:?}", price_data);
    }
});

pub async fn trade_subscribe( &mut self, ) -> Result<UnboundedReceiver<TradeFields>, AppError>

Subscribes to trade updates for the account.

This method creates a subscription to receive real-time trade confirmations, order updates (OPU), and working order updates (WOU) for the account, and returns a channel receiver for consuming the updates.

Returns

Returns a receiver channel for TradeFields updates, or an error if the subscription setup failed.

Examples

let mut receiver = client.trade_subscribe().await?;

tokio::spawn(async move {
    while let Some(trade_fields) = receiver.recv().await {
        println!("Trade update: {:?}", trade_fields);
    }
});

pub async fn account_subscribe( &mut self, fields: HashSet<StreamingAccountDataField>, ) -> Result<UnboundedReceiver<AccountFields>, AppError>

Subscribes to account data updates.

This method creates a subscription to receive real-time account updates including profit/loss, margin, equity, available funds, and other account metrics, and returns a channel receiver for consuming the updates.

Arguments

• fields - Set of account data fields to receive (e.g., PNL, MARGIN, EQUITY, etc.)

Returns

Returns a receiver channel for AccountFields updates, or an error if the subscription setup failed.

Examples

let mut receiver = client.account_subscribe(fields).await?;

tokio::spawn(async move {
    while let Some(account_fields) = receiver.recv().await {
        println!("Account update: {:?}", account_fields);
    }
});

pub async fn price_subscribe( &mut self, epics: Vec<String>, fields: HashSet<StreamingPriceField>, ) -> Result<UnboundedReceiver<PriceData>, AppError>

Subscribes to price data updates for the specified instruments.

This method creates a subscription to receive real-time price updates including bid/ask prices, sizes, and multiple currency levels for the given EPICs, and returns a channel receiver for consuming the updates.

Arguments

• epics - List of instrument EPICs to subscribe to
• fields - Set of price data fields to receive (e.g., BID_PRICE1, ASK_PRICE1, etc.)

Returns

Returns a receiver channel for PriceData updates, or an error if the subscription setup failed.

Examples

let mut receiver = client.price_subscribe(
    vec!["IX.D.DAX.DAILY.IP".to_string()],
    fields
).await?;

tokio::spawn(async move {
    while let Some(price_data) = receiver.recv().await {
        println!("Price update: {:?}", price_data);
    }
});

pub async fn chart_subscribe( &mut self, epics: Vec<String>, scale: ChartScale, fields: HashSet<StreamingChartField>, ) -> Result<UnboundedReceiver<ChartData>, AppError>

Subscribes to chart data updates for the specified instruments and scale.

This method creates a subscription to receive real-time chart updates including OHLC data, volume, and other chart metrics for the given EPICs and chart scale, and returns a channel receiver for consuming the updates.

Arguments

• epics - List of instrument EPICs to subscribe to.
• scale - Chart scale (e.g., Tick, 1Min, 5Min, etc.).
• fields - Set of chart data fields to receive (e.g., OPEN, HIGH, LOW, CLOSE, VOLUME).

Returns

Returns a receiver channel for ChartData updates, or an error if the subscription setup failed.

Examples

let mut receiver = client.chart_subscribe(
    vec!["IX.D.DAX.DAILY.IP".to_string()],
    ChartScale::OneMin,
    fields
).await?;

tokio::spawn(async move {
    while let Some(chart_data) = receiver.recv().await {
        println!("Chart update: {:?}", chart_data);
    }
});

pub async fn connect( &mut self, shutdown_signal: Option<Arc<Notify>>, ) -> Result<(), AppError>

Connects all active Lightstreamer clients and maintains the connections.

This method establishes connections for all streaming clients that have active subscriptions (market and price). Each client runs in its own task and all connections are maintained until a shutdown signal is received.

Arguments

• shutdown_signal - Optional signal to gracefully shutdown all connections. If None, a default signal handler for SIGINT/SIGTERM will be created.

Returns

Returns Ok(()) when all connections are closed gracefully, or an error if any connection fails after maximum retry attempts.

pub async fn disconnect(&mut self) -> Result<(), AppError>

Disconnects all active Lightstreamer clients.

This method gracefully closes all streaming connections (market and price) and tears down the per-subscription converter tasks so they do not idle for the process lifetime. Closing the Lightstreamer session closes the item channels feeding the converters, so they would exit on their own; aborting and awaiting them here guarantees a deterministic, leak-free shutdown. Calling disconnect more than once is safe: the converter list is drained and the client disconnect is idempotent.

Returns

Returns Ok(()) if all disconnections were successful.

Trait Implementations

impl Drop for StreamerClient

fn drop(&mut self)

Aborts any converter tasks that were not already torn down by StreamerClient::disconnect, so dropping the client never orphans a spawned task. Abort is synchronous, so no runtime is required here.

fn pin_drop(self: Pin<&mut Self>)

🔬This is a nightly-only experimental API. (pin_ergonomics)

Execute the destructor for this type, but different to Drop::drop, it requires self to be pinned.