fmp_rust_sdk/websocket/
types.rs

1use std::{fmt, sync::Arc, time::Duration};
2
3use super::{Heartbeat, StreamQuote};
4use serde::Deserialize;
5use thiserror::Error;
6
7/// The FMP market cluster to connect to.
8#[derive(Debug, Clone, Copy, PartialEq, Eq)]
9pub enum WebSocketCluster {
10    /// The stock trade and quote feed.
11    Stocks,
12    /// The cryptocurrency trade and quote feed.
13    Crypto,
14    /// The forex trade and quote feed.
15    Forex,
16}
17
18impl WebSocketCluster {
19    /// Returns the endpoint published in FMP's public WebSocket documentation.
20    pub const fn endpoint(self) -> &'static str {
21        match self {
22            Self::Stocks => "wss://websockets.financialmodelingprep.com",
23            Self::Crypto => "wss://crypto.financialmodelingprep.com",
24            Self::Forex => "wss://forex.financialmodelingprep.com",
25        }
26    }
27}
28
29/// A validated JSON control envelope with its original, credential-redacted JSON.
30///
31/// FMP's public documentation does not prescribe every control payload.
32/// `status` and `message` are populated when present at the top level; additional
33/// fields can be decoded from `json` into an account-specific strict model.
34#[derive(Debug, Clone, PartialEq, Eq)]
35pub struct ControlEvent {
36    /// The required nonempty event name.
37    pub event: String,
38    /// The server status, from a top-level control or nested login data.
39    pub status: Option<u16>,
40    /// The server message, from a top-level control or nested login data.
41    pub message: Option<String>,
42    /// The complete control envelope, with occurrences of the API key redacted.
43    pub json: String,
44}
45
46/// Validates an account's login acknowledgment.
47///
48/// Return `Ok(true)` only for a positively validated authentication success,
49/// `Ok(false)` for other controls, and an error for a malformed acknowledgment.
50/// An unknown event or absence of an error is not authentication success.
51pub type LoginConfirmation =
52    Arc<dyn Fn(&ControlEvent) -> Result<bool, WebSocketError> + Send + Sync>;
53
54/// Explicit timeouts and connection retry settings.
55#[derive(Clone)]
56pub struct WebSocketConfig {
57    pub(super) handshake_timeout: Duration,
58    pub(super) receive_timeout: Duration,
59    pub(super) connect_attempts: u32,
60    pub(super) retry_delay: Duration,
61    pub(super) confirm_login: LoginConfirmation,
62}
63
64impl fmt::Debug for WebSocketConfig {
65    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
66        formatter
67            .debug_struct("WebSocketConfig")
68            .field("handshake_timeout", &self.handshake_timeout)
69            .field("receive_timeout", &self.receive_timeout)
70            .field("connect_attempts", &self.connect_attempts)
71            .field("retry_delay", &self.retry_delay)
72            .finish_non_exhaustive()
73    }
74}
75
76impl WebSocketConfig {
77    /// Creates a configuration; timeouts and attempt count must be positive.
78    ///
79    /// Only transient HTTP upgrade/transport failures are retried, before login.
80    /// Authentication, protocol, and active connection failures are returned.
81    pub fn new(
82        handshake_timeout: Duration,
83        receive_timeout: Duration,
84        connect_attempts: u32,
85        retry_delay: Duration,
86        confirm_login: impl Fn(&ControlEvent) -> Result<bool, WebSocketError> + Send + Sync + 'static,
87    ) -> Result<Self, WebSocketError> {
88        if handshake_timeout.is_zero() || receive_timeout.is_zero() || connect_attempts == 0 {
89            return Err(WebSocketError::Configuration {
90                reason: "handshake_timeout, receive_timeout, and connect_attempts must be positive"
91                    .to_owned(),
92            });
93        }
94        Ok(Self {
95            handshake_timeout,
96            receive_timeout,
97            connect_attempts,
98            retry_delay,
99            confirm_login: Arc::new(confirm_login),
100        })
101    }
102}
103
104/// Required fields of a trade or trade-break update.
105#[derive(Debug, Clone, PartialEq, Deserialize)]
106pub struct TradeUpdate {
107    /// The asset's symbol.
108    #[serde(rename = "s")]
109    pub symbol: String,
110    /// The provider timestamp, preserved without assuming a time unit.
111    #[serde(rename = "t")]
112    pub timestamp: u64,
113    /// The execution price.
114    #[serde(rename = "lp")]
115    pub price: f64,
116    /// The traded quantity; crypto quantities can be fractional.
117    #[serde(rename = "ls")]
118    pub size: f64,
119}
120
121/// Required fields of a top-of-book quote update.
122#[derive(Debug, Clone, PartialEq, Deserialize)]
123pub struct QuoteUpdate {
124    /// The asset's symbol.
125    #[serde(rename = "s")]
126    pub symbol: String,
127    /// The provider timestamp, preserved without assuming a time unit.
128    #[serde(rename = "t")]
129    pub timestamp: u64,
130    /// The lowest ask price.
131    #[serde(rename = "ap")]
132    pub ask_price: f64,
133    /// The quantity at the ask price.
134    #[serde(rename = "as")]
135    pub ask_size: f64,
136    /// The highest bid price.
137    #[serde(rename = "bp")]
138    pub bid_price: f64,
139    /// The quantity at the bid price.
140    #[serde(rename = "bs")]
141    pub bid_size: f64,
142}
143
144/// A strictly typed update using FMP's documented T/Q/B discriminator.
145#[derive(Debug, Clone, PartialEq, Deserialize)]
146#[serde(tag = "type")]
147pub enum MarketUpdate {
148    /// A last-trade execution.
149    #[serde(rename = "T")]
150    Trade(TradeUpdate),
151    /// A top-of-book quote.
152    #[serde(rename = "Q")]
153    Quote(QuoteUpdate),
154    /// A trade-break update.
155    #[serde(rename = "B")]
156    Break(TradeUpdate),
157}
158
159/// The remote peer's completed close handshake.
160#[derive(Debug, Clone, PartialEq, Eq)]
161pub struct CloseInfo {
162    /// The close code, if the peer supplied a close frame payload.
163    pub code: Option<u16>,
164    /// The peer's reason, with credentials redacted.
165    pub reason: String,
166}
167
168/// An incoming application, pong, or close event.
169#[derive(Debug, Clone, PartialEq)]
170pub enum WebSocketEvent {
171    /// A validated trade, quote, or trade-break update.
172    Market(MarketUpdate),
173    /// A current market stream's validated quote snapshot.
174    StreamQuote(Box<StreamQuote>),
175    /// The current server's required timestamped heartbeat.
176    Heartbeat(Heartbeat),
177    /// A control message; server error controls instead return an error.
178    Control(ControlEvent),
179    /// The peer's pong payload.
180    Pong(Vec<u8>),
181    /// A cleanly closed connection.
182    Closed(CloseInfo),
183}
184
185/// A specific connection, configuration, server, or protocol failure.
186#[derive(Debug, Error)]
187pub enum WebSocketError {
188    /// Invalid local configuration or subscription symbols.
189    #[error("invalid WebSocket configuration: {reason}")]
190    Configuration {
191        /// The actionable validation failure.
192        reason: String,
193    },
194    /// The account has no configured grant for this cluster.
195    #[error(
196        "no explicit WebSocket entitlement configured for {cluster:?}; confirm the cluster grant with FMP"
197    )]
198    Entitlement {
199        /// The requested cluster.
200        cluster: WebSocketCluster,
201    },
202    /// The requested current market stream has no configured grant.
203    #[error(
204        "no explicit WebSocket entitlement configured for stream {stream}; confirm the stream grant with FMP"
205    )]
206    StreamEntitlement {
207        /// The requested stream identifier.
208        stream: String,
209    },
210    /// A bounded operation exceeded its configured timeout.
211    #[error("WebSocket {operation} timed out for {endpoint}")]
212    Timeout {
213        /// The operation that exceeded its deadline.
214        operation: &'static str,
215        /// The requested endpoint.
216        endpoint: String,
217    },
218    /// An HTTP upgrade or active transport operation failed.
219    #[error("WebSocket {operation} failed for {endpoint}: {reason}")]
220    Transport {
221        /// The failed operation.
222        operation: &'static str,
223        /// The requested endpoint.
224        endpoint: String,
225        /// The underlying error, including HTTP status/body when available.
226        reason: String,
227    },
228    /// A server control explicitly rejected a request or authentication.
229    #[error("FMP WebSocket rejected {event} at {endpoint} (status {status:?}): {body}")]
230    Server {
231        /// The failed control event.
232        event: String,
233        /// The provider's top-level status code, if supplied.
234        status: Option<u16>,
235        /// The requested endpoint.
236        endpoint: String,
237        /// The complete redacted server control payload.
238        body: String,
239    },
240    /// The remote payload failed strict runtime validation.
241    #[error("invalid WebSocket response at {endpoint}: {reason}; body: {body}")]
242    Protocol {
243        /// The requested endpoint.
244        endpoint: String,
245        /// The exact validation failure.
246        reason: String,
247        /// The credential-redacted payload.
248        body: String,
249    },
250    /// A command was attempted after a completed remote close.
251    #[error(
252        "WebSocket connection is closed at {endpoint}; reconnect explicitly before issuing commands"
253    )]
254    Closed {
255        /// The closed connection's endpoint.
256        endpoint: String,
257    },
258}
fmp_rust_sdk/websocket/types.rs

fmp_rust_sdk/websocket/
types.rs
