fmp_rust_sdk/websocket/
standard.rs

1use serde::Serialize;
2use url::Url;
3
4use crate::plan::Subscription;
5
6use super::{
7    Connection, MarketStream, StreamConfig, StreamEntitlements, WebSocketError, WebSocketEvent,
8};
9
10/// FMP's current native WebSocket connector for approved market streams.
11///
12/// The [current FMP documentation](https://site.financialmodelingprep.com/developer/docs#websockets)
13/// publishes `wss://socket.financialmodelingprep.com` and requires approved
14/// declaration forms for direct streaming. Each stream must be explicitly
15/// declared through [`StreamEntitlements`]; a billing plan alone grants none.
16/// Authentication is confirmed using the documented nested login response.
17#[derive(Debug)]
18pub struct StandardWebSocketClient {
19    connection: Connection,
20    subscription: Subscription,
21    entitlements: StreamEntitlements,
22}
23
24impl StandardWebSocketClient {
25    /// Connects to the current native endpoint and waits for confirmed login.
26    pub async fn connect(
27        api_key: &str,
28        subscription: &Subscription,
29        entitlements: StreamEntitlements,
30        config: StreamConfig,
31    ) -> Result<Self, WebSocketError> {
32        let endpoint = Url::parse("wss://socket.financialmodelingprep.com").map_err(|error| {
33            WebSocketError::Configuration {
34                reason: format!("invalid documented standard WebSocket URL: {error}"),
35            }
36        })?;
37        Self::connect_to(api_key, subscription, entitlements, endpoint, config).await
38    }
39
40    /// Connects to a trusted explicit current endpoint; ws is loopback-only.
41    ///
42    /// Use this for FMP-assigned endpoints or real local integration servers.
43    /// URL credentials, query parameters, and fragments are rejected.
44    pub async fn connect_to(
45        api_key: &str,
46        subscription: &Subscription,
47        entitlements: StreamEntitlements,
48        endpoint: Url,
49        config: StreamConfig,
50    ) -> Result<Self, WebSocketError> {
51        Ok(Self {
52            connection: Connection::connect(api_key, endpoint, config.config).await?,
53            subscription: *subscription,
54            entitlements,
55        })
56    }
57
58    /// Returns the configured subscription without claiming server verification.
59    pub const fn subscription(&self) -> Subscription {
60        self.subscription
61    }
62
63    /// Subscribes to an explicitly granted exchange, synthetic, or delayed stream.
64    pub async fn subscribe(&mut self, stream: &MarketStream) -> Result<(), WebSocketError> {
65        if !self.entitlements.allows(stream) {
66            return Err(WebSocketError::StreamEntitlement {
67                stream: stream.as_str().to_owned(),
68            });
69        }
70        let command = StreamCommand {
71            event: "subscribe",
72            data: StreamData {
73                stream: stream.as_str(),
74            },
75        };
76        let json = serde_json::to_string(&command).map_err(|error| {
77            self.connection
78                .protocol_error(&error.to_string(), "stream subscribe serialization")
79        })?;
80        self.connection.send_text(json, "stream subscribe").await
81    }
82
83    /// Stops an explicitly granted stream using the documented unsubscribe event.
84    pub async fn unsubscribe(&mut self, stream: &MarketStream) -> Result<(), WebSocketError> {
85        if !self.entitlements.allows(stream) {
86            return Err(WebSocketError::StreamEntitlement {
87                stream: stream.as_str().to_owned(),
88            });
89        }
90        let command = StreamCommand {
91            event: "unsubscribe",
92            data: StreamData {
93                stream: stream.as_str(),
94            },
95        };
96        let json = serde_json::to_string(&command).map_err(|error| {
97            self.connection
98                .protocol_error(&error.to_string(), "stream unsubscribe serialization")
99        })?;
100        self.connection.send_text(json, "stream unsubscribe").await
101    }
102
103    /// Receives a strictly validated snapshot, heartbeat, control, pong, or close.
104    pub async fn next_event(&mut self) -> Result<WebSocketEvent, WebSocketError> {
105        self.connection.next_event().await
106    }
107
108    /// Completes a close handshake within the configured handshake timeout.
109    pub async fn close(self) -> Result<(), WebSocketError> {
110        self.connection.close().await
111    }
112}
113
114impl crate::Client {
115    /// Opens an authenticated current market-stream connection for this client.
116    ///
117    /// Explicit stream grants are independent of the client's configured plan.
118    pub async fn connect_streams(
119        &self,
120        entitlements: StreamEntitlements,
121        config: StreamConfig,
122    ) -> Result<StandardWebSocketClient, WebSocketError> {
123        StandardWebSocketClient::connect(self.api_key(), self.subscription(), entitlements, config)
124            .await
125    }
126}
127
128#[derive(Serialize)]
129struct StreamCommand<'a> {
130    event: &'static str,
131    data: StreamData<'a>,
132}
133
134#[derive(Serialize)]
135struct StreamData<'a> {
136    stream: &'a str,
137}
fmp_rust_sdk/websocket/standard.rs

fmp_rust_sdk/websocket/
standard.rs
