brokapi_bcs_protocol/ws/client.rs
1//! The WebSocket client: it opens one background session per channel.
2
3use anyhow::Result;
4use kanal::AsyncSender;
5use serde::Deserialize;
6
7use crate::config::BcsConfig;
8use crate::error::BcsApiError;
9use crate::http::client::BcsHttpClient;
10use crate::models::instrument::InstrumentKey;
11use crate::models::messages::IncomingMessage;
12use crate::ws::models::MarketDataSubscribe;
13use crate::ws::session::{WsOptions, WsSession};
14
15pub use crate::ws::models::WsOutgoing;
16
17/// A WebSocket channel (service) of the BCS API. Each channel is its own URL and its own connection.
18///
19/// Only [`BcsWsChannel::MarketData`] takes subscription commands. The others push their data as soon as
20/// they are connected, with nothing to send.
21///
22/// The broker limits the number of simultaneous connections per account: 2 for portfolio, limits and
23/// margin indicators, 4 for orders, 20 for market data (broker docs). It also limits the client to 10
24/// messages per second and 100 instruments per connection.
25///
26/// # Examples
27///
28/// ```
29/// use brokapi_bcs_protocol::ws::client::BcsWsChannel;
30///
31/// let channel: BcsWsChannel = serde_json::from_str("\"market_data\"")?;
32/// assert_eq!(channel, BcsWsChannel::MarketData);
33/// # Ok::<(), serde_json::Error>(())
34/// ```
35#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)]
36#[serde(rename_all = "snake_case")]
37pub enum BcsWsChannel {
38 /// Account limits: positions, money, futures limits (`limits/ws`). Parsed into
39 /// [`IncomingMessage::Limits`].
40 Limits,
41 /// Portfolio positions with values and P/L (`portfolio/ws`). Parsed into
42 /// [`IncomingMessage::Portfolio`].
43 Portfolio,
44 /// Market data: quotes, order book, trades tape and last candle (`market-data/ws`). The only
45 /// channel that takes subscriptions. Parsed into [`IncomingMessage::MarketData`].
46 MarketData,
47 /// Deprecated order execution channel (`orders/execution/ws`). The broker docs mark it
48 /// deprecated: use [`BcsWsChannel::OrderEvents`]. Parsed into [`IncomingMessage::OrderExecution`].
49 OrderExecution,
50 /// Deprecated order transaction channel (`orders/transaction/ws`). The broker docs mark it
51 /// deprecated: use [`BcsWsChannel::OrderEvents`]. Parsed into [`IncomingMessage::OrderTransaction`].
52 OrderTransaction,
53 /// Current order status channel (`orders/events/ws`), replaces execution-status and
54 /// transaction-status, which the docs now mark deprecated. Parsed into
55 /// [`IncomingMessage::OrderExecution`].
56 OrderEvents,
57 /// Margin indicators: portfolio value, margin level, debts (`marginal-indicators/ws`). Parsed into
58 /// [`IncomingMessage::MarginalIndicators`].
59 MarginalIndicators,
60}
61
62impl BcsWsChannel {
63 /// The URL path of the channel, added to the WebSocket base URL.
64 pub(super) fn path(&self) -> &'static str {
65 match self {
66 BcsWsChannel::Limits => "/trade-api-bff-limit/api/v1/limits/ws",
67 BcsWsChannel::Portfolio => "/trade-api-bff-portfolio/api/v1/portfolio/ws",
68 BcsWsChannel::MarketData => "/trade-api-market-data-connector/api/v1/market-data/ws",
69 BcsWsChannel::OrderExecution => "/trade-api-bff-operations/api/v1/orders/execution/ws",
70 BcsWsChannel::OrderTransaction => {
71 "/trade-api-bff-operations/api/v1/orders/transaction/ws"
72 }
73 BcsWsChannel::OrderEvents => "/trade-api-bff-operations/api/v1/orders/events/ws",
74 BcsWsChannel::MarginalIndicators => {
75 "/trade-api-bff-marginal-indicators/api/v1/marginal-indicators/ws"
76 }
77 }
78 }
79}
80
81/// The WebSocket client of the BCS API: a factory of [`WsSession`]s that share one authorized HTTP
82/// client.
83///
84/// The client holds no connection itself. Each `connect_*` call starts a **background session** for one
85/// channel (see [`WsSession`] for what it does for you) and returns at once. Frames arrive in the
86/// `kanal` channel you pass in. The sessions of one client share the access token.
87///
88/// # Examples
89///
90/// ```
91/// # #[tokio::main(flavor = "current_thread")]
92/// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
93/// # let server = brokapi_mocks::bcs::MockServer::start(Default::default()).await;
94/// # let cfg = brokapi_bcs_protocol::config::BcsConfig::for_mock("mock-refresh-token", &server.http_base);
95/// use std::time::Duration;
96/// use brokapi_bcs_protocol::models::instrument::InstrumentKey;
97/// use brokapi_bcs_protocol::models::messages::IncomingMessage;
98/// use brokapi_bcs_protocol::ws::client::BcsWsClient;
99/// use brokapi_bcs_protocol::ws::models::MarketDataMessage;
100///
101/// let ws = BcsWsClient::new(cfg)?;
102/// let (tx, rx) = kanal::bounded_async::<IncomingMessage>(64);
103/// let session = ws.connect_market_data(tx);
104/// ws.subscribe_order_book(&session, vec![InstrumentKey::new("SBER", "TQBR")], 10).await?;
105///
106/// // Wait for the first book, with a timeout so a broken link cannot hang the program.
107/// let book = tokio::time::timeout(Duration::from_secs(5), async {
108/// loop {
109/// if let IncomingMessage::MarketData(MarketDataMessage::OrderBook(book)) = rx.recv().await? {
110/// return Ok::<_, kanal::ReceiveError>(book);
111/// }
112/// }
113/// })
114/// .await??;
115/// assert_eq!(book.ticker, "SBER");
116/// # Ok(())
117/// # }
118/// ```
119pub struct BcsWsClient {
120 pub(super) http_client: BcsHttpClient,
121 options: WsOptions,
122}
123
124impl BcsWsClient {
125 /// Creates a client. It builds the HTTP client for tokens and makes no network call.
126 ///
127 /// The sessions use [`WsOptions::default`] until you call [`BcsWsClient::with_options`].
128 ///
129 /// # Errors
130 ///
131 /// The same as [`BcsHttpClient::new`]: [`BcsApiError::HttpError`] if the HTTP stack cannot be built.
132 ///
133 /// # Examples
134 ///
135 /// ```
136 /// use brokapi_bcs_protocol::config::BcsConfig;
137 /// use brokapi_bcs_protocol::ws::client::BcsWsClient;
138 ///
139 /// let ws = BcsWsClient::new(BcsConfig::new("refresh-token".to_string()))?;
140 /// assert_eq!(ws.http_client().ws_base_url(), "wss://ws.broker.ru");
141 /// # Ok::<(), brokapi_bcs_protocol::error::BcsApiError>(())
142 /// ```
143 pub fn new(config: BcsConfig) -> Result<Self, BcsApiError> {
144 let http_client = BcsHttpClient::new(config)?;
145 Ok(Self {
146 http_client,
147 options: WsOptions::default(),
148 })
149 }
150
151 /// Sets the options of the sessions this client opens from now on. Sessions already open keep the
152 /// options they started with.
153 ///
154 /// # Examples
155 ///
156 /// ```
157 /// use brokapi_bcs_protocol::config::BcsConfig;
158 /// use brokapi_bcs_protocol::ws::client::BcsWsClient;
159 /// use brokapi_bcs_protocol::ws::session::WsOptions;
160 ///
161 /// let ws = BcsWsClient::new(BcsConfig::new("refresh-token".to_string()))?
162 /// .with_options(WsOptions { announce_connects: true, ..WsOptions::default() });
163 /// # let _ = ws;
164 /// # Ok::<(), brokapi_bcs_protocol::error::BcsApiError>(())
165 /// ```
166 pub fn with_options(mut self, options: WsOptions) -> Self {
167 self.options = options;
168 self
169 }
170
171 /// Starts a background session for a channel and returns it at once.
172 ///
173 /// The connection is made in the background: the returned session may not be connected yet
174 /// ([`WsSession::is_connected`]). Every parsed frame goes to `event_tx`.
175 ///
176 /// # Panics
177 ///
178 /// Panics if there is no Tokio runtime, because the session is a Tokio task.
179 ///
180 /// # Examples
181 ///
182 /// ```
183 /// # #[tokio::main(flavor = "current_thread")]
184 /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
185 /// # let server = brokapi_mocks::bcs::MockServer::start(Default::default()).await;
186 /// # let cfg = brokapi_bcs_protocol::config::BcsConfig::for_mock("mock-refresh-token", &server.http_base);
187 /// use brokapi_bcs_protocol::models::messages::IncomingMessage;
188 /// use brokapi_bcs_protocol::ws::client::{BcsWsChannel, BcsWsClient};
189 ///
190 /// let ws = BcsWsClient::new(cfg)?;
191 /// let (tx, _rx) = kanal::bounded_async::<IncomingMessage>(16);
192 /// let session = ws.spawn_session(BcsWsChannel::Portfolio, tx);
193 /// assert_eq!(session.channel(), BcsWsChannel::Portfolio);
194 /// # Ok(())
195 /// # }
196 /// ```
197 pub fn spawn_session(
198 &self,
199 service: BcsWsChannel,
200 event_tx: AsyncSender<IncomingMessage>,
201 ) -> WsSession {
202 WsSession::spawn_with(service, self.http_client.clone(), event_tx, self.options)
203 }
204
205 /// The HTTP client the sessions take their tokens from. It is a normal [`BcsHttpClient`], so you can
206 /// make REST calls with the same tokens.
207 pub fn http_client(&self) -> &BcsHttpClient {
208 &self.http_client
209 }
210
211 /// Starts the portfolio session (`portfolio/ws`). The broker sends an array with one record per
212 /// position. Frames arrive as [`IncomingMessage::Portfolio`]. The broker docs do not say whether a
213 /// first snapshot comes right at connect (the in-process imitation sends one).
214 ///
215 /// # Panics
216 ///
217 /// Panics if there is no Tokio runtime.
218 ///
219 /// # Examples
220 ///
221 /// ```
222 /// # #[tokio::main(flavor = "current_thread")]
223 /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
224 /// # let server = brokapi_mocks::bcs::MockServer::start(Default::default()).await;
225 /// # let cfg = brokapi_bcs_protocol::config::BcsConfig::for_mock("mock-refresh-token", &server.http_base);
226 /// use std::time::Duration;
227 /// use brokapi_bcs_protocol::models::messages::IncomingMessage;
228 /// use brokapi_bcs_protocol::ws::client::BcsWsClient;
229 ///
230 /// let ws = BcsWsClient::new(cfg)?;
231 /// let (tx, rx) = kanal::bounded_async::<IncomingMessage>(16);
232 /// let _session = ws.connect_portfolio(tx);
233 /// let first = tokio::time::timeout(Duration::from_secs(5), rx.recv()).await??;
234 /// match first {
235 /// IncomingMessage::Portfolio(positions) => assert!(!positions.is_empty()),
236 /// other => panic!("unexpected: {other:?}"),
237 /// }
238 /// # Ok(())
239 /// # }
240 /// ```
241 pub fn connect_portfolio(&self, event_tx: AsyncSender<IncomingMessage>) -> WsSession {
242 self.spawn_session(BcsWsChannel::Portfolio, event_tx)
243 }
244
245 /// Starts the **deprecated** transaction-status session (`orders/transaction/ws`). Prefer
246 /// [`BcsWsClient::connect_order_events`]. Frames arrive as [`IncomingMessage::OrderTransaction`].
247 /// The in-process imitation does not serve this channel with data, so there is no runnable example.
248 ///
249 /// # Panics
250 ///
251 /// Panics if there is no Tokio runtime.
252 pub fn connect_order_transaction(&self, event_tx: AsyncSender<IncomingMessage>) -> WsSession {
253 self.spawn_session(BcsWsChannel::OrderTransaction, event_tx)
254 }
255
256 /// Starts the limits session (`limits/ws`). The broker sends one object with the current limits:
257 /// positions, money, futures holdings and futures limits. Frames arrive as
258 /// [`IncomingMessage::Limits`]. The broker docs do not say whether a first snapshot comes right at
259 /// connect (the in-process imitation sends one).
260 ///
261 /// # Panics
262 ///
263 /// Panics if there is no Tokio runtime.
264 ///
265 /// # Examples
266 ///
267 /// ```
268 /// # #[tokio::main(flavor = "current_thread")]
269 /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
270 /// # let server = brokapi_mocks::bcs::MockServer::start(Default::default()).await;
271 /// # let cfg = brokapi_bcs_protocol::config::BcsConfig::for_mock("mock-refresh-token", &server.http_base);
272 /// use std::time::Duration;
273 /// use brokapi_bcs_protocol::models::messages::IncomingMessage;
274 /// use brokapi_bcs_protocol::ws::client::BcsWsClient;
275 ///
276 /// let ws = BcsWsClient::new(cfg)?;
277 /// let (tx, rx) = kanal::bounded_async::<IncomingMessage>(16);
278 /// let _session = ws.connect_limits(tx);
279 /// let first = tokio::time::timeout(Duration::from_secs(5), rx.recv()).await??;
280 /// match first {
281 /// IncomingMessage::Limits(limits) => assert!(limits.money_limits.is_some()),
282 /// other => panic!("unexpected: {other:?}"),
283 /// }
284 /// # Ok(())
285 /// # }
286 /// ```
287 pub fn connect_limits(&self, event_tx: AsyncSender<IncomingMessage>) -> WsSession {
288 self.spawn_session(BcsWsChannel::Limits, event_tx)
289 }
290
291 /// Starts the order events session (`orders/events/ws`), the current channel for order states and
292 /// fills. Frames arrive as [`IncomingMessage::OrderExecution`], in the same shape as the old
293 /// execution channel.
294 ///
295 /// The docs do not describe a replay of history, so treat an event that happened while you were not
296 /// connected as lost. After a reconnect ask for the state of your open orders with
297 /// [`BcsHttpClient::get_order`] (the session tells you about a reconnect with
298 /// [`IncomingMessage::Reconnected`], see [`WsOptions::announce_connects`]).
299 ///
300 /// # Panics
301 ///
302 /// Panics if there is no Tokio runtime.
303 ///
304 /// # Examples
305 ///
306 /// ```
307 /// # use rust_decimal::Decimal;
308 /// # use rust_decimal_macros::dec;
309 /// # #[tokio::main(flavor = "current_thread")]
310 /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
311 /// # let options = std::sync::Arc::new(brokapi_mocks::bcs::MockOptions::default());
312 /// # let server = brokapi_mocks::bcs::MockServer::start(options.clone()).await;
313 /// # let cfg = brokapi_bcs_protocol::config::BcsConfig::for_mock("mock-refresh-token", &server.http_base).with_write(true);
314 /// use std::time::Duration;
315 /// use brokapi_bcs_protocol::http::models::CreateOrderRequest;
316 /// use brokapi_bcs_protocol::models::enums::{OrderSide, OrderStatus};
317 /// use brokapi_bcs_protocol::models::messages::IncomingMessage;
318 /// use brokapi_bcs_protocol::ws::client::BcsWsClient;
319 /// use brokapi_bcs_protocol::ws::session::WsOptions;
320 ///
321 /// let ws = BcsWsClient::new(cfg)?
322 /// .with_options(WsOptions { announce_connects: true, ..WsOptions::default() });
323 /// let (tx, rx) = kanal::bounded_async::<IncomingMessage>(16);
324 /// let _session = ws.connect_order_events(tx);
325 ///
326 /// // The first message is the marker of the connect. Only after it do events follow.
327 /// let first = tokio::time::timeout(Duration::from_secs(5), rx.recv()).await??;
328 /// assert!(matches!(first, IncomingMessage::Reconnected));
329 /// # // The socket is registered by the imitation a moment after the marker.
330 /// # tokio::time::timeout(Duration::from_secs(5), async {
331 /// # while options.event_sockets() == 0 { tokio::time::sleep(Duration::from_millis(10)).await; }
332 /// # }).await?;
333 ///
334 /// // Place an order over REST. Its state arrives on the socket.
335 /// let cid = uuid::Uuid::new_v4();
336 /// ws.http_client()
337 /// .create_order(CreateOrderRequest::limit(cid, "SBER", "TQBR", OrderSide::Buy, 1, dec!(300.0)))
338 /// .await?;
339 /// let event = tokio::time::timeout(Duration::from_secs(5), rx.recv()).await??;
340 /// match event {
341 /// IncomingMessage::OrderExecution(m) => {
342 /// assert_eq!(m.data.unwrap().order_status, OrderStatus::New);
343 /// }
344 /// other => panic!("unexpected: {other:?}"),
345 /// }
346 /// # Ok(())
347 /// # }
348 /// ```
349 pub fn connect_order_events(&self, event_tx: AsyncSender<IncomingMessage>) -> WsSession {
350 self.spawn_session(BcsWsChannel::OrderEvents, event_tx)
351 }
352
353 /// Starts the **deprecated** execution-status session (`orders/execution/ws`). Prefer
354 /// [`BcsWsClient::connect_order_events`]. Frames arrive as [`IncomingMessage::OrderExecution`].
355 /// The in-process imitation does not serve this channel with data, so there is no runnable example.
356 ///
357 /// # Panics
358 ///
359 /// Panics if there is no Tokio runtime.
360 pub fn connect_order_execution(&self, event_tx: AsyncSender<IncomingMessage>) -> WsSession {
361 self.spawn_session(BcsWsChannel::OrderExecution, event_tx)
362 }
363
364 /// Starts the margin indicators session (`marginal-indicators/ws`): portfolio value, margin level,
365 /// debts. The broker sends all of it in one object. Frames arrive as
366 /// [`IncomingMessage::MarginalIndicators`]. The docs do not say whether a first snapshot comes right
367 /// at connect (the in-process imitation sends one).
368 ///
369 /// # Panics
370 ///
371 /// Panics if there is no Tokio runtime.
372 ///
373 /// # Examples
374 ///
375 /// ```
376 /// # #[tokio::main(flavor = "current_thread")]
377 /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
378 /// # let server = brokapi_mocks::bcs::MockServer::start(Default::default()).await;
379 /// # let cfg = brokapi_bcs_protocol::config::BcsConfig::for_mock("mock-refresh-token", &server.http_base);
380 /// use std::time::Duration;
381 /// use brokapi_bcs_protocol::models::messages::IncomingMessage;
382 /// use brokapi_bcs_protocol::ws::client::BcsWsClient;
383 ///
384 /// let ws = BcsWsClient::new(cfg)?;
385 /// let (tx, rx) = kanal::bounded_async::<IncomingMessage>(16);
386 /// let _session = ws.connect_marginal_indicators(tx);
387 /// let first = tokio::time::timeout(Duration::from_secs(5), rx.recv()).await??;
388 /// match first {
389 /// IncomingMessage::MarginalIndicators(m) => assert!(m.portfolio_current_value.is_some()),
390 /// other => panic!("unexpected: {other:?}"),
391 /// }
392 /// # Ok(())
393 /// # }
394 /// ```
395 pub fn connect_marginal_indicators(&self, event_tx: AsyncSender<IncomingMessage>) -> WsSession {
396 self.spawn_session(BcsWsChannel::MarginalIndicators, event_tx)
397 }
398
399 /// Starts the market data session (`market-data/ws`). It sends nothing until you subscribe with
400 /// [`BcsWsClient::subscribe_quotes`], [`BcsWsClient::subscribe_order_book`],
401 /// [`BcsWsClient::subscribe_last_trades`] or [`BcsWsClient::subscribe_last_candle`]. Frames arrive
402 /// as [`IncomingMessage::MarketData`].
403 ///
404 /// # Panics
405 ///
406 /// Panics if there is no Tokio runtime.
407 ///
408 /// # Examples
409 ///
410 /// See [`BcsWsClient`] for a full example with a subscription.
411 ///
412 /// ```
413 /// # #[tokio::main(flavor = "current_thread")]
414 /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
415 /// # let server = brokapi_mocks::bcs::MockServer::start(Default::default()).await;
416 /// # let cfg = brokapi_bcs_protocol::config::BcsConfig::for_mock("mock-refresh-token", &server.http_base);
417 /// use brokapi_bcs_protocol::models::messages::IncomingMessage;
418 /// use brokapi_bcs_protocol::ws::client::{BcsWsChannel, BcsWsClient};
419 ///
420 /// let ws = BcsWsClient::new(cfg)?;
421 /// let (tx, _rx) = kanal::bounded_async::<IncomingMessage>(16);
422 /// let session = ws.connect_market_data(tx);
423 /// assert_eq!(session.channel(), BcsWsChannel::MarketData);
424 /// # Ok(())
425 /// # }
426 /// ```
427 pub fn connect_market_data(&self, event_tx: AsyncSender<IncomingMessage>) -> WsSession {
428 self.spawn_session(BcsWsChannel::MarketData, event_tx)
429 }
430
431 /// Subscribes to quotes of the instruments (`dataType` 3).
432 ///
433 /// The command is kept by the session and sent again after every reconnect. The broker answers
434 /// with a subscription confirmation and then with quote frames.
435 ///
436 /// # Errors
437 ///
438 /// [`BcsApiError::SubscriptionError`] if the session is not a market data session, if the session
439 /// already holds 100 subscribe commands, or if the session task has ended. A refusal by the broker
440 /// is not an error here: it arrives later as a frame.
441 ///
442 /// # Examples
443 ///
444 /// ```
445 /// # #[tokio::main(flavor = "current_thread")]
446 /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
447 /// # let server = brokapi_mocks::bcs::MockServer::start(Default::default()).await;
448 /// # let cfg = brokapi_bcs_protocol::config::BcsConfig::for_mock("mock-refresh-token", &server.http_base);
449 /// use std::time::Duration;
450 /// use brokapi_bcs_protocol::models::instrument::InstrumentKey;
451 /// use brokapi_bcs_protocol::models::messages::IncomingMessage;
452 /// use brokapi_bcs_protocol::ws::client::BcsWsClient;
453 /// use brokapi_bcs_protocol::ws::models::MarketDataMessage;
454 ///
455 /// let ws = BcsWsClient::new(cfg)?;
456 /// let (tx, rx) = kanal::bounded_async::<IncomingMessage>(16);
457 /// let session = ws.connect_market_data(tx);
458 /// ws.subscribe_quotes(&session, vec![InstrumentKey::new("SBER", "TQBR")]).await?;
459 ///
460 /// let quote = tokio::time::timeout(Duration::from_secs(5), async {
461 /// loop {
462 /// if let IncomingMessage::MarketData(MarketDataMessage::Quotes(q)) = rx.recv().await? {
463 /// return Ok::<_, kanal::ReceiveError>(q);
464 /// }
465 /// }
466 /// })
467 /// .await??;
468 /// assert_eq!(quote.ticker, "SBER");
469 ///
470 /// // A subscription on a channel that takes none is refused locally.
471 /// let (tx2, _rx2) = kanal::bounded_async::<IncomingMessage>(16);
472 /// let portfolio = ws.connect_portfolio(tx2);
473 /// assert!(ws.subscribe_quotes(&portfolio, vec![]).await.is_err());
474 /// # Ok(())
475 /// # }
476 /// ```
477 pub async fn subscribe_quotes(
478 &self,
479 session: &WsSession,
480 instruments: Vec<InstrumentKey>,
481 ) -> Result<(), BcsApiError> {
482 session
483 .subscribe(MarketDataSubscribe::quotes(instruments))
484 .await
485 }
486
487 /// Subscribes to the order book of the instruments (`dataType` 0) with `depth` levels per side.
488 ///
489 /// The broker docs allow `depth` from 1 to 20 (default 20). The value is **not checked here**: a
490 /// bad depth comes back as an error frame from the broker. Errors and retention are as for
491 /// [`BcsWsClient::subscribe_quotes`].
492 ///
493 /// # Errors
494 ///
495 /// The same as [`BcsWsClient::subscribe_quotes`].
496 ///
497 /// # Examples
498 ///
499 /// See the example on [`BcsWsClient`].
500 pub async fn subscribe_order_book(
501 &self,
502 session: &WsSession,
503 instruments: Vec<InstrumentKey>,
504 depth: u32,
505 ) -> Result<(), BcsApiError> {
506 session
507 .subscribe(MarketDataSubscribe::order_book(instruments, depth))
508 .await
509 }
510
511 /// Subscribes to the anonymous trades tape of the instruments (`dataType` 2). Each trade arrives as
512 /// one frame, see [`LastTradesData`](crate::ws::models::LastTradesData).
513 ///
514 /// # Errors
515 ///
516 /// The same as [`BcsWsClient::subscribe_quotes`].
517 ///
518 /// # Examples
519 ///
520 /// ```
521 /// # use rust_decimal::Decimal;
522 /// # use rust_decimal_macros::dec;
523 /// # #[tokio::main(flavor = "current_thread")]
524 /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
525 /// # let server = brokapi_mocks::bcs::MockServer::start(Default::default()).await;
526 /// # let cfg = brokapi_bcs_protocol::config::BcsConfig::for_mock("mock-refresh-token", &server.http_base);
527 /// use std::time::Duration;
528 /// use brokapi_bcs_protocol::models::instrument::InstrumentKey;
529 /// use brokapi_bcs_protocol::models::messages::IncomingMessage;
530 /// use brokapi_bcs_protocol::ws::client::BcsWsClient;
531 /// use brokapi_bcs_protocol::ws::models::MarketDataMessage;
532 ///
533 /// let ws = BcsWsClient::new(cfg)?;
534 /// let (tx, rx) = kanal::bounded_async::<IncomingMessage>(16);
535 /// let session = ws.connect_market_data(tx);
536 /// ws.subscribe_last_trades(&session, vec![InstrumentKey::new("SBER", "TQBR")]).await?;
537 ///
538 /// let trade = tokio::time::timeout(Duration::from_secs(5), async {
539 /// loop {
540 /// if let IncomingMessage::MarketData(MarketDataMessage::LastTrades(t)) = rx.recv().await? {
541 /// return Ok::<_, kanal::ReceiveError>(t);
542 /// }
543 /// }
544 /// })
545 /// .await??;
546 /// assert!(trade.price > dec!(0.0));
547 /// # Ok(())
548 /// # }
549 /// ```
550 pub async fn subscribe_last_trades(
551 &self,
552 session: &WsSession,
553 instruments: Vec<InstrumentKey>,
554 ) -> Result<(), BcsApiError> {
555 session
556 .subscribe(MarketDataSubscribe::last_trades(instruments))
557 .await
558 }
559
560 /// Subscribes to the last candle of the instruments (`dataType` 1) for a time frame.
561 ///
562 /// `time_frame` is one of `M1 M5 M15 M30 H1 H4 D W MN` (broker docs). It is not checked here.
563 /// Note `D` and `W`: the names `D1` and `W1` are not in the broker's list.
564 ///
565 /// # Errors
566 ///
567 /// The same as [`BcsWsClient::subscribe_quotes`].
568 ///
569 /// # Examples
570 ///
571 /// ```
572 /// # #[tokio::main(flavor = "current_thread")]
573 /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
574 /// # let server = brokapi_mocks::bcs::MockServer::start(Default::default()).await;
575 /// # let cfg = brokapi_bcs_protocol::config::BcsConfig::for_mock("mock-refresh-token", &server.http_base);
576 /// use std::time::Duration;
577 /// use brokapi_bcs_protocol::models::instrument::InstrumentKey;
578 /// use brokapi_bcs_protocol::models::messages::IncomingMessage;
579 /// use brokapi_bcs_protocol::ws::client::BcsWsClient;
580 /// use brokapi_bcs_protocol::ws::models::MarketDataMessage;
581 ///
582 /// let ws = BcsWsClient::new(cfg)?;
583 /// let (tx, rx) = kanal::bounded_async::<IncomingMessage>(16);
584 /// let session = ws.connect_market_data(tx);
585 /// ws.subscribe_last_candle(&session, vec![InstrumentKey::new("SBER", "TQBR")], "M1").await?;
586 ///
587 /// let candle = tokio::time::timeout(Duration::from_secs(5), async {
588 /// loop {
589 /// if let IncomingMessage::MarketData(MarketDataMessage::LastCandle(c)) = rx.recv().await? {
590 /// return Ok::<_, kanal::ReceiveError>(c);
591 /// }
592 /// }
593 /// })
594 /// .await??;
595 /// assert_eq!(candle.time_frame.as_deref(), Some("M1"));
596 /// # Ok(())
597 /// # }
598 /// ```
599 pub async fn subscribe_last_candle(
600 &self,
601 session: &WsSession,
602 instruments: Vec<InstrumentKey>,
603 time_frame: &str,
604 ) -> Result<(), BcsApiError> {
605 session
606 .subscribe(MarketDataSubscribe::last_candle(instruments, time_frame))
607 .await
608 }
609}