projectx_client/realtime/
session_handle.rs

1// SPDX-FileCopyrightText: 2026 Kevin Monaghan
2// SPDX-License-Identifier: MIT-0
3
4use super::{
5    AccountId, ContractId, Hub, RealtimeError, RealtimeGeneration, RealtimeInner, Value, Weak,
6};
7
8/// Subscription and invocation admission bound to one exact ready socket.
9/// This handle cannot stop or reconnect the transport and does not keep its owner alive.
10#[derive(Clone)]
11pub struct RealtimeSession {
12    pub(super) inner: Weak<RealtimeInner>,
13    pub(super) generation: RealtimeGeneration,
14}
15
16impl RealtimeSession {
17    /// The socket to which every invocation through this handle is bound.
18    #[must_use]
19    pub const fn generation(&self) -> RealtimeGeneration {
20        self.generation
21    }
22
23    /// Invokes an arbitrary provider target and waits for its completion frame.
24    ///
25    /// # Errors
26    ///
27    /// Returns an error when disconnected, a bounded capacity is exhausted,
28    /// the provider rejects the invocation, or completion times out. A timeout
29    /// or cancellation after queue admission has an unknown outcome and reclaims
30    /// only that invocation slot. It never closes the connection or retries the call.
31    pub async fn invoke(
32        &self,
33        target: impl Into<String>,
34        arguments: Vec<Value>,
35    ) -> Result<(), RealtimeError> {
36        self.send_invocation(target.into(), arguments).await
37    }
38
39    /// Subscribes to market trades for a contract.
40    ///
41    /// # Errors
42    ///
43    /// Returns a real-time invocation error.
44    pub async fn subscribe_contract_trades(
45        &self,
46        contract: &ContractId,
47    ) -> Result<(), RealtimeError> {
48        self.ensure_hub(Hub::Market)?;
49        self.contract_invocation("SubscribeContractTrades", contract)
50            .await
51    }
52
53    /// Unsubscribes from market trades for a contract.
54    ///
55    /// # Errors
56    ///
57    /// Returns a real-time invocation error.
58    pub async fn unsubscribe_contract_trades(
59        &self,
60        contract: &ContractId,
61    ) -> Result<(), RealtimeError> {
62        self.ensure_hub(Hub::Market)?;
63        self.contract_invocation("UnsubscribeContractTrades", contract)
64            .await
65    }
66
67    /// Subscribes to market quotes for a contract.
68    ///
69    /// # Errors
70    ///
71    /// Returns a real-time invocation error.
72    pub async fn subscribe_contract_quotes(
73        &self,
74        contract: &ContractId,
75    ) -> Result<(), RealtimeError> {
76        self.ensure_hub(Hub::Market)?;
77        self.contract_invocation("SubscribeContractQuotes", contract)
78            .await
79    }
80
81    /// Unsubscribes from market quotes for a contract.
82    ///
83    /// # Errors
84    ///
85    /// Returns a real-time invocation error.
86    pub async fn unsubscribe_contract_quotes(
87        &self,
88        contract: &ContractId,
89    ) -> Result<(), RealtimeError> {
90        self.ensure_hub(Hub::Market)?;
91        self.contract_invocation("UnsubscribeContractQuotes", contract)
92            .await
93    }
94
95    /// Subscribes to market depth for a contract.
96    ///
97    /// # Errors
98    ///
99    /// Returns a real-time invocation error.
100    pub async fn subscribe_contract_depth(
101        &self,
102        contract: &ContractId,
103    ) -> Result<(), RealtimeError> {
104        self.ensure_hub(Hub::Market)?;
105        self.contract_invocation("SubscribeContractMarketDepth", contract)
106            .await
107    }
108
109    /// Unsubscribes from market depth for a contract.
110    ///
111    /// # Errors
112    ///
113    /// Returns a real-time invocation error.
114    pub async fn unsubscribe_contract_depth(
115        &self,
116        contract: &ContractId,
117    ) -> Result<(), RealtimeError> {
118        self.ensure_hub(Hub::Market)?;
119        self.contract_invocation("UnsubscribeContractMarketDepth", contract)
120            .await
121    }
122
123    /// Subscribes to account updates.
124    ///
125    /// # Errors
126    ///
127    /// Returns a real-time invocation error.
128    pub async fn subscribe_accounts(&self) -> Result<(), RealtimeError> {
129        self.ensure_hub(Hub::User)?;
130        self.send_invocation("SubscribeAccounts".to_owned(), Vec::new())
131            .await
132    }
133
134    /// Unsubscribes from account updates.
135    ///
136    /// # Errors
137    ///
138    /// Returns a real-time invocation error.
139    pub async fn unsubscribe_accounts(&self) -> Result<(), RealtimeError> {
140        self.ensure_hub(Hub::User)?;
141        self.send_invocation("UnsubscribeAccounts".to_owned(), Vec::new())
142            .await
143    }
144
145    /// Subscribes to order updates for an account.
146    ///
147    /// # Errors
148    ///
149    /// Returns a real-time invocation error.
150    pub async fn subscribe_orders(&self, account: AccountId) -> Result<(), RealtimeError> {
151        self.ensure_hub(Hub::User)?;
152        self.account_invocation("SubscribeOrders", account).await
153    }
154
155    /// Unsubscribes from order updates for an account.
156    ///
157    /// # Errors
158    ///
159    /// Returns a real-time invocation error.
160    pub async fn unsubscribe_orders(&self, account: AccountId) -> Result<(), RealtimeError> {
161        self.ensure_hub(Hub::User)?;
162        self.account_invocation("UnsubscribeOrders", account).await
163    }
164
165    /// Subscribes to position updates for an account.
166    ///
167    /// # Errors
168    ///
169    /// Returns a real-time invocation error.
170    pub async fn subscribe_positions(&self, account: AccountId) -> Result<(), RealtimeError> {
171        self.ensure_hub(Hub::User)?;
172        self.account_invocation("SubscribePositions", account).await
173    }
174
175    /// Unsubscribes from position updates for an account.
176    ///
177    /// # Errors
178    ///
179    /// Returns a real-time invocation error.
180    pub async fn unsubscribe_positions(&self, account: AccountId) -> Result<(), RealtimeError> {
181        self.ensure_hub(Hub::User)?;
182        self.account_invocation("UnsubscribePositions", account)
183            .await
184    }
185
186    /// Subscribes to trade updates for an account.
187    ///
188    /// # Errors
189    ///
190    /// Returns a real-time invocation error.
191    pub async fn subscribe_trades(&self, account: AccountId) -> Result<(), RealtimeError> {
192        self.ensure_hub(Hub::User)?;
193        self.account_invocation("SubscribeTrades", account).await
194    }
195
196    /// Unsubscribes from trade updates for an account.
197    ///
198    /// # Errors
199    ///
200    /// Returns a real-time invocation error.
201    pub async fn unsubscribe_trades(&self, account: AccountId) -> Result<(), RealtimeError> {
202        self.ensure_hub(Hub::User)?;
203        self.account_invocation("UnsubscribeTrades", account).await
204    }
205
206    fn ensure_hub(&self, expected: Hub) -> Result<(), RealtimeError> {
207        if self.inner.upgrade().ok_or(RealtimeError::NotConnected)?.hub == expected {
208            Ok(())
209        } else {
210            Err(RealtimeError::WrongHub)
211        }
212    }
213
214    async fn contract_invocation(
215        &self,
216        target: &str,
217        contract: &ContractId,
218    ) -> Result<(), RealtimeError> {
219        self.inner
220            .upgrade()
221            .ok_or(RealtimeError::NotConnected)?
222            .send_invocation(
223                Some(self.generation),
224                target.to_owned(),
225                vec![Value::String(contract.to_string())],
226            )
227            .await
228    }
229
230    async fn account_invocation(
231        &self,
232        target: &str,
233        account: AccountId,
234    ) -> Result<(), RealtimeError> {
235        self.inner
236            .upgrade()
237            .ok_or(RealtimeError::NotConnected)?
238            .send_invocation(
239                Some(self.generation),
240                target.to_owned(),
241                vec![Value::from(account.get())],
242            )
243            .await
244    }
245
246    async fn send_invocation(
247        &self,
248        target: String,
249        arguments: Vec<Value>,
250    ) -> Result<(), RealtimeError> {
251        self.inner
252            .upgrade()
253            .ok_or(RealtimeError::NotConnected)?
254            .send_invocation(Some(self.generation), target, arguments)
255            .await
256    }
257}
258
259impl std::fmt::Debug for RealtimeSession {
260    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
261        f.debug_struct("RealtimeSession")
262            .field("generation", &self.generation)
263            .finish_non_exhaustive()
264    }
265}
projectx_client/realtime/session_handle.rs

projectx_client/realtime/
session_handle.rs
