brokapi_finam/core/
mod.rs

1//! Sans-io cores: frames in, normalized events out. No sockets, timers or tasks here.
2//!
3//! A stream task reads gRPC messages and wraps them into frames ([`MdFrame`], [`UdFrame`]). The core
4//! turns a frame into events of `brokapi-core` (order books, quotes, orders...) and keeps the state that
5//! needs memory: the local order book, the M10 candle being built, the id of seen trades, the
6//! order lifecycle. Because the core does no input and output, tests drive it with plain values and a
7//! fixed clock.
8//!
9//! # Examples
10//!
11//! A quote subscription walked through by hand: no network, no clock besides `Instant::now()`.
12//!
13//! ```
14//! use std::time::Instant;
15//! use brokapi_core::{Event, SubscriptionId};
16//! use brokapi_finam::core::{MdCore, MdEvent, MdFrame, SubStatus};
17//! use brokapi_finam::util::instrument_from_symbol;
18//! use brokapi_sansio::{BrokerCore, DesiredSubscription, EventSink, SubscriptionKind, SubscriptionSet};
19//! use brokapi_finam_protocol::pb::google::r#type::Decimal;
20//! use brokapi_finam_protocol::pb::grpc::tradeapi::v1::marketdata as pb;
21//!
22//! let sub = SubscriptionId::new(1);
23//! let sber = instrument_from_symbol("SBER@MISX");
24//! let mut core = MdCore::new();
25//! core.apply(&SubscriptionSet::with_subscriptions(vec![DesiredSubscription {
26//!     id: sub,
27//!     kind: SubscriptionKind::Quotes { instrument: sber },
28//! }]));
29//! assert_eq!(core.status(sub), Some(SubStatus::Connecting));
30//!
31//! let mut out = EventSink::new();
32//! core.on_frame(MdFrame { sub, event: MdEvent::Connected }, Instant::now(), &mut out);
33//! assert_eq!(core.status(sub), Some(SubStatus::Active));
34//!
35//! let price = |v: &str| Some(Decimal { value: v.into() });
36//! let quote = pb::Quote { symbol: "SBER@MISX".into(), bid: price("250.1"), ask: price("250.2"), ..Default::default() };
37//! let message = pb::SubscribeQuoteResponse { quote: vec![quote], error: None };
38//! core.on_frame(MdFrame { sub, event: MdEvent::Quotes(message) }, Instant::now(), &mut out);
39//!
40//! let events = out.drain();
41//! assert!(matches!(events[0].payload, Event::Quote(_)));
42//! ```
43
44pub mod md;
45pub mod ud;
46
47pub use md::{MdCore, MdEvent, MdFrame};
48pub use ud::{UdCore, UdEvent, UdFrame};
49
50/// State of one subscription as seen by the core.
51#[derive(Debug, Clone, PartialEq, Eq)]
52pub enum SubStatus {
53    /// Registered, stream not open yet.
54    Connecting,
55    /// Stream open.
56    Active,
57    /// Stream dropped, being reopened with backoff.
58    Reconnecting(String),
59    /// Server refuses the subscription (bad symbol, no permission...); retried slowly.
60    Failed(String),
61    /// Stream is open but carried an in-band error message.
62    Erroring(String),
63}
64
65/// What a stream task reports about its stream (before it is wrapped into a core frame).
66#[derive(Debug, Clone)]
67pub enum StreamEvent<M> {
68    /// The stream was opened (also after a reopen).
69    Connected,
70    /// One message of the stream.
71    Msg(M),
72    /// The stream broke or ended. It is being reopened. The text says why.
73    Disconnected(String),
74    /// The broker refuses the stream (bad symbol, no permission). It is retried slowly. The text says why.
75    Failed(String),
76}
brokapi_finam/core/mod.rs

brokapi_finam/core/
mod.rs
