Struct StreamerClient 

pub struct StreamerClient { /* private fields */ }

Available on crate feature streaming only.

Streaming client for IG Markets real-time data.

One Lightstreamer session carries every IG channel: market data (MARKET:), detailed prices (PRICE:, served by the Pricing data adapter), trade confirmations (TRADE:), account balances (ACCOUNT:) and candles (CHART:). The data adapter is a property of the subscription, so one session is enough — the pair of connections this type used to open was a workaround for the previous client library.

Lifecycle

The session is opened lazily by the first *_subscribe call and lives until disconnect or Drop. connect does not open it; it consumes the session event stream and blocks until the shutdown signal fires or the session ends for good, which is what makes it usable as the “run until stopped” body of a streaming binary.

Channels

Each *_subscribe returns an unbounded receiver of decoded DTOs. The sender is owned by a converter task spawned per subscription; when the caller drops the receiver that task logs and exits, and when the session ends the subscription stream closes and the task exits. Every one of those tasks is tracked and joined by disconnect, so none outlives the client.

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 and logs in to obtain the Lightstreamer endpoint and credentials. No connection is established yet — the session opens on the first subscription.

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 fails, or AppError::InvalidInput if IG returned an endpoint the Lightstreamer client rejects.

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.

Errors

Returns AppError if the session lookup fails, or AppError::InvalidInput if IG returned an endpoint the Lightstreamer client rejects.

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.

Errors

Returns AppError::InvalidInput if epics or fields is empty or contains a name Lightstreamer rejects, and AppError::WebSocketError if the session cannot be opened or the subscription cannot be sent.

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.

Errors

Returns AppError::WebSocketError if the session cannot be opened or the subscription cannot be sent.

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.

Errors

Returns AppError::InvalidInput if fields is empty or contains a name Lightstreamer rejects, and AppError::WebSocketError if the session cannot be opened or the subscription cannot be sent.

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.

Errors

Returns AppError::InvalidInput if epics or fields is empty or contains a name Lightstreamer rejects, and AppError::WebSocketError if the session cannot be opened or the subscription cannot be sent.

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.

Errors

Returns AppError::InvalidInput if epics or fields is empty or contains a name Lightstreamer rejects, and AppError::WebSocketError if the session cannot be opened or the subscription cannot be sent.

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>

Consumes the session event stream and blocks until shutdown.

The Lightstreamer session is already open by the time this is called (the first subscription opened it) and reconnection is handled by lightstreamer-rs itself, with bounded jittered backoff. What this method adds is observation: it reports what every reconnection meant — in particular a session that was replaced rather than preserved, after which every subscription has been re-executed and a fresh snapshot is on its way — and it returns when the session ends for good.

Arguments

• shutdown_signal - Signalled by the caller to stop. When None, this waits for SIGINT / SIGTERM instead.

Returns

Ok(()) when the shutdown signal fired or the session was closed by this client.

Errors

Returns AppError::WebSocketError when the session ended for a reason this client did not ask for: refused by IG, reconnection budget exhausted, or an internal failure in the streaming crate.

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

Disconnects the Lightstreamer session and tears down every converter task.

The order matters: the converters are signalled and joined first, which drops their subscription streams and so unsubscribes, and only then is the session closed. Calling this more than once is safe — the task list is drained and the session handle is taken.

Errors

Returns AppError::WebSocketError if closing the session failed. The converter tasks are stopped either way.

Trait Implementations

impl Drop for StreamerClient

fn drop(&mut self)

Signals and then abandons any converter task that StreamerClient::disconnect did not already join, so dropping the client never leaves one running. Dropping the session handle closes the Lightstreamer session; neither step can await, which is why disconnect is still the way to observe the close completing.

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.