Struct WsSession 

pub struct WsSession { /* private fields */ }

A background WebSocket session for one channel. It reconnects by itself.

What the session does for you:

• It connects with a bearer token and takes a fresh token at every reconnect. If the broker refuses the connection with 403 or “Unauthorized”, it forces a token refresh.
• It sends a ping every 30 seconds and answers the broker’s pings. If no pong or ping arrives for 60 seconds it drops the link and reconnects.
• After a lost link it waits (500 ms, doubling up to 30 s, see WsOptions) and connects again. Market data subscriptions are sent again after each connect.
• It parses every text frame and pushes the IncomingMessage with try_send into your channel. If your channel is full, the frame is dropped and counted (WsSession::dropped_events): the session never waits for a slow reader. Size the channel for bursts and read it quickly.
• It stops when your channel is closed (the receiver dropped) or when the session value is dropped.

There is no explicit close method. Dropping the WsSession stops its background task at once, also when the session is waiting for the next reconnect or has never been able to connect.

The session never reports a lost link as an error value. Use WsSession::is_connected and, for an ordered signal, WsOptions::announce_connects.

Examples

use std::time::Duration;
use brokapi_bcs_protocol::http::client::BcsHttpClient;
use brokapi_bcs_protocol::models::instrument::InstrumentKey;
use brokapi_bcs_protocol::models::messages::IncomingMessage;
use brokapi_bcs_protocol::ws::client::BcsWsChannel;
use brokapi_bcs_protocol::ws::models::MarketDataSubscribe;
use brokapi_bcs_protocol::ws::session::{WsOptions, WsSession};

let http = BcsHttpClient::new(cfg)?;
let (tx, rx) = kanal::bounded_async::<IncomingMessage>(64);
let options = WsOptions { announce_connects: true, ..WsOptions::default() };
let session = WsSession::spawn_with(BcsWsChannel::MarketData, http, tx, options);

let sub = MarketDataSubscribe::order_book(vec![InstrumentKey::new("SBER", "TQBR")], 5);
session.subscribe(sub).await?;

// First the connect marker, then the answers of the broker.
let first = tokio::time::timeout(Duration::from_secs(5), rx.recv()).await??;
assert!(matches!(first, IncomingMessage::Reconnected));
assert_eq!(session.dropped_events(), 0);

Implementations

impl WsSession

pub fn spawn( channel: BcsWsChannel, http_client: BcsHttpClient, event_tx: AsyncSender<IncomingMessage>, ) -> Self

Starts a session with the default WsOptions. It returns at once, the connection is made in the background.

Panics

Panics if there is no Tokio runtime, because the session is a Tokio task.

Examples

use brokapi_bcs_protocol::http::client::BcsHttpClient;
use brokapi_bcs_protocol::models::messages::IncomingMessage;
use brokapi_bcs_protocol::ws::client::BcsWsChannel;
use brokapi_bcs_protocol::ws::session::WsSession;

let (tx, _rx) = kanal::bounded_async::<IncomingMessage>(16);
let session = WsSession::spawn(BcsWsChannel::Limits, BcsHttpClient::new(cfg)?, tx);
assert_eq!(session.channel(), BcsWsChannel::Limits);

pub fn spawn_with( channel: BcsWsChannel, http_client: BcsHttpClient, event_tx: AsyncSender<IncomingMessage>, options: WsOptions, ) -> Self

Starts a session with the given options. Otherwise the same as WsSession::spawn.

Panics

Panics if there is no Tokio runtime.

Examples

See the example on WsSession.

pub fn dropped_events(&self) -> u64

The number of parsed frames dropped because your event channel was full.

A growing number means the reader is too slow or the channel is too small. Frames that were dropped are gone. For order events that means states you must re-read over REST.

Examples

use brokapi_bcs_protocol::models::messages::IncomingMessage;
use brokapi_bcs_protocol::ws::client::BcsWsClient;

let ws = BcsWsClient::new(cfg)?;
let (tx, _rx) = kanal::bounded_async::<IncomingMessage>(16);
let session = ws.connect_limits(tx);
assert_eq!(session.dropped_events(), 0);

pub fn is_connected(&self) -> bool

true while the link is up and the subscriptions have been sent again.

It is false before the first connect, during a reconnect and after the session has ended. Do not use it to decide that data is flowing, it only says the socket is open.

Examples

use std::time::Duration;
use brokapi_bcs_protocol::models::messages::IncomingMessage;
use brokapi_bcs_protocol::ws::client::BcsWsClient;

let ws = BcsWsClient::new(cfg)?;
let (tx, _rx) = kanal::bounded_async::<IncomingMessage>(16);
let session = ws.connect_limits(tx);
tokio::time::timeout(Duration::from_secs(5), async {
    while !session.is_connected() {
        tokio::time::sleep(Duration::from_millis(10)).await;
    }
})
.await?;

pub async fn wait_connected(&self)

Waits until the link is up and the subscriptions have been sent again (is_connected is true). Returns at once when it already is.

It has no limit of its own: it waits for a connect that may never come (the broker is unreachable). Put your own timeout around it, as [tokio::time::timeout]. The session keeps trying in the background either way. A session that has ended for good never comes up, so the wait then ends only with your timeout.

Examples

use std::time::Duration;
use brokapi_bcs_protocol::models::messages::IncomingMessage;
use brokapi_bcs_protocol::ws::client::BcsWsClient;

let ws = BcsWsClient::new(cfg)?;
let (tx, _rx) = kanal::bounded_async::<IncomingMessage>(16);
let session = ws.connect_limits(tx);
tokio::time::timeout(Duration::from_secs(5), session.wait_connected()).await?;
assert!(session.is_connected());

pub fn channel(&self) -> BcsWsChannel

The channel this session is on.

pub async fn subscribe( &self, sub: MarketDataSubscribe, ) -> Result<(), BcsApiError>

Subscribes on the market data channel. The command is kept and sent again after each reconnect.

The command is put in the restore list first and then queued for sending. Something to know:

• The session counts commands: at most 100 per session. The broker’s limit (broker docs) is 100 instruments per connection, so a command with many instruments can pass the local check and still be refused by the broker (as an error frame, not as a return value).
• If you subscribe before the first connect, the command waits in the restore list and goes out once, when the session connects. The same holds for a command given while the session is connecting.

The answer of the broker (subscription confirmation, data, or an error frame) arrives in your event channel, not as the return value.

Errors

BcsApiError::SubscriptionError if the session is not on BcsWsChannel::MarketData, if it already holds 100 subscribe commands, or if the session task has ended.

Examples

use brokapi_bcs_protocol::error::BcsApiError;
use brokapi_bcs_protocol::http::client::BcsHttpClient;
use brokapi_bcs_protocol::models::instrument::InstrumentKey;
use brokapi_bcs_protocol::models::messages::IncomingMessage;
use brokapi_bcs_protocol::ws::client::BcsWsChannel;
use brokapi_bcs_protocol::ws::models::MarketDataSubscribe;
use brokapi_bcs_protocol::ws::session::WsSession;

let (tx, _rx) = kanal::bounded_async::<IncomingMessage>(16);
let sber = vec![InstrumentKey::new("SBER", "TQBR")];

// A portfolio session takes no subscriptions.
let portfolio = WsSession::spawn(BcsWsChannel::Portfolio, BcsHttpClient::new(cfg.clone())?, tx.clone());
let err = portfolio.subscribe(MarketDataSubscribe::quotes(sber.clone())).await.unwrap_err();
assert!(matches!(err, BcsApiError::SubscriptionError { .. }));

// A market data session does.
let md = WsSession::spawn(BcsWsChannel::MarketData, BcsHttpClient::new(cfg)?, tx);
md.subscribe(MarketDataSubscribe::quotes(sber)).await?;

pub async fn unsubscribe( &self, sub: MarketDataSubscribe, ) -> Result<(), BcsApiError>

Unsubscribes on the market data channel. Pass the same MarketDataSubscribe you subscribed with: the session turns it into the unsubscribe command itself.

The matching entry is removed from the restore list only if it is equal to the one you pass (same instruments in the same order, same depth, same time frame). A different grouping of the same instruments is sent to the broker as an unsubscribe, but the original stays in the restore list and comes back after a reconnect. The broker docs describe no answer to an unsubscribe.

Errors

BcsApiError::SubscriptionError if the session is not on BcsWsChannel::MarketData or if the session task has ended.

Examples

use brokapi_bcs_protocol::http::client::BcsHttpClient;
use brokapi_bcs_protocol::models::instrument::InstrumentKey;
use brokapi_bcs_protocol::models::messages::IncomingMessage;
use brokapi_bcs_protocol::ws::client::BcsWsChannel;
use brokapi_bcs_protocol::ws::models::MarketDataSubscribe;
use brokapi_bcs_protocol::ws::session::WsSession;

let (tx, _rx) = kanal::bounded_async::<IncomingMessage>(16);
let md = WsSession::spawn(BcsWsChannel::MarketData, BcsHttpClient::new(cfg)?, tx);

let sub = MarketDataSubscribe::order_book(vec![InstrumentKey::new("SBER", "TQBR")], 5);
md.subscribe(sub.clone()).await?;
md.unsubscribe(sub).await?; // the same value as at subscribe

Trait Implementations

impl Drop for WsSession

fn drop(&mut self)

Stops the background task, wherever it is: connected, waiting for a reconnect or never connected.

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.