xtb_client/
client.rs

1use std::collections::HashMap;
2use std::marker::PhantomData;
3use std::str::FromStr;
4use std::sync::Arc;
5use std::time::Duration;
6
7use async_trait::async_trait;
8use derive_setters::Setters;
9use serde::{Deserialize, Serialize};
10use serde_json::{from_value, to_value, Value};
11use thiserror::Error;
12use tokio::spawn;
13use tokio::sync::Mutex;
14use tokio::task::JoinHandle;
15use tokio::time::sleep;
16use tracing::{debug, error};
17use url::Url;
18
19use crate::{BasicMessageStream, BasicXtbConnection, BasicXtbStreamConnection, DataMessageFilter, MessageStream, ResponsePromise, XtbConnection, BasicXtbConnectionError, XtbStreamConnection, BasicXtbStreamConnectionError};
20use crate::message_processing::ProcessedMessage;
21use crate::schema::{COMMAND_GET_ALL_SYMBOLS, COMMAND_GET_CALENDAR, COMMAND_GET_CHART_LAST_REQUEST, COMMAND_GET_CHART_RANGE_REQUEST, COMMAND_GET_COMMISSION_DEF, COMMAND_GET_CURRENT_USER_DATA, COMMAND_GET_IBS_HISTORY, COMMAND_GET_MARGIN_LEVEL, COMMAND_GET_MARGIN_TRADE, COMMAND_GET_NEWS, COMMAND_GET_PROFIT_CALCULATION, COMMAND_GET_SERVER_TIME, COMMAND_GET_STEP_RULES, COMMAND_GET_SYMBOL, COMMAND_GET_TICK_PRICES, COMMAND_GET_TRADE_RECORDS, COMMAND_GET_TRADES, COMMAND_GET_TRADES_HISTORY, COMMAND_GET_TRADING_HOURS, COMMAND_GET_VERSION, COMMAND_LOGIN, COMMAND_PING, COMMAND_TRADE_TRANSACTION, COMMAND_TRADE_TRANSACTION_STATUS, ErrorResponse, GetAllSymbolsRequest, GetAllSymbolsResponse, GetCalendarRequest, GetCalendarResponse, GetChartLastRequestRequest, GetChartLastRequestResponse, GetChartRangeRequestRequest, GetChartRangeRequestResponse, GetCommissionDefRequest, GetCommissionDefResponse, GetCurrentUserDataRequest, GetCurrentUserDataResponse, GetIbsHistoryRequest, GetIbsHistoryResponse, GetMarginLevelRequest, GetMarginLevelResponse, GetMarginTradeRequest, GetMarginTradeResponse, GetNewsRequest, GetNewsResponse, GetProfitCalculationRequest, GetProfitCalculationResponse, GetServerTimeRequest, GetServerTimeResponse, GetStepRulesRequest, GetStepRulesResponse, GetSymbolRequest, GetSymbolResponse, GetTickPricesRequest, GetTickPricesResponse, GetTradeRecordsRequest, GetTradeRecordsResponse, GetTradesHistoryRequest, GetTradesHistoryResponse, GetTradesRequest, GetTradesResponse, GetTradingHoursRequest, GetTradingHoursResponse, GetVersionRequest, GetVersionResponse, LoginRequest, PingRequest, STREAM_BALANCE, STREAM_CANDLES, STREAM_BALANCE_SUBSCRIBE, STREAM_CANDLES_SUBSCRIBE, STREAM_KEEP_ALIVE_SUBSCRIBE, STREAM_NEWS_SUBSCRIBE, STREAM_PROFITS_SUBSCRIBE, STREAM_TICK_PRICES_SUBSCRIBE, STREAM_TRADE_STATUS_SUBSCRIBE, STREAM_TRADES_SUBSCRIBE, STREAM_KEEP_ALIVE, STREAM_NEWS, STREAM_PING, STREAM_PROFITS, STREAM_BALANCE_UNSUBSCRIBE, STREAM_CANDLES_UNSUBSCRIBE, STREAM_KEEP_ALIVE_UNSUBSCRIBE, STREAM_NEWS_UNSUBSCRIBE, STREAM_PROFITS_UNSUBSCRIBE, STREAM_TICK_PRICES_UNSUBSCRIBE, STREAM_TRADE_STATUS_UNSUBSCRIBE, STREAM_TRADES_UNSUBSCRIBE, STREAM_TICK_PRICES, STREAM_TRADE_STATUS, STREAM_TRADES, StreamDataMessage, StreamGetBalanceData, StreamGetBalanceSubscribe, StreamGetBalanceUnsubscribe, StreamGetCandlesData, StreamGetCandlesSubscribe, StreamGetCandlesUnsubscribe, StreamGetKeepAliveData, StreamGetKeepAliveSubscribe, StreamGetKeepAliveUnsubscribe, StreamGetNewsData, StreamGetNewsSubscribe, StreamGetNewsUnsubscribe, StreamGetProfitData, StreamGetProfitSubscribe, StreamGetProfitUnsubscribe, StreamGetTickPricesData, StreamGetTickPricesSubscribe, StreamGetTickPricesUnsubscribe, StreamGetTradesData, StreamGetTradesSubscribe, StreamGetTradeStatusData, StreamGetTradeStatusSubscribe, StreamGetTradeStatusUnsubscribe, StreamGetTradesUnsubscribe, StreamPingSubscribe, TradeTransactionRequest, TradeTransactionResponse, TradeTransactionStatusRequest, TradeTransactionStatusResponse};
22
23
24/// Builder for `XtbClient`.
25///
26/// Configuration can be set with this class and when configuration step is finished, the `XtbClient`
27/// can be created by the `build()` method.
28///
29/// Configurable fields are:
30///
31/// * `api_url` - url of the request/response API server.
32/// * `stream_api_url` - url of the stream API server.
33/// * `app_id` - application identifier (deprecated by the official API documentation)
34/// * `app_name` - application name (deprecated by the official API documentation)
35/// * `ping_period` - interval between ping commands. Default interval is 30s.
36///
37/// The required configuration values are `api_url` and `stream_api_url`. Other values are optional.
38///
39/// Official documentation says, the `ping_interval` should be less than 10 minutes. But real world
40/// observation shows that the maximal interval must be less than 1 minute.
41#[derive(Default, Setters)]
42#[setters(into, prefix = "with_", strip_option)]
43pub struct XtbClientBuilder {
44    /// Url of the request/response API server
45    api_url: Option<String>,
46    /// Url of the stream API server
47    stream_api_url: Option<String>,
48    /// Identifier of the application. (Deprecated by the official api)
49    app_id: Option<String>,
50    /// Name of the application (deprecated by the official api)
51    app_name: Option<String>,
52    /// Interval between pings. Shouldn't be greater than 1 minute.
53    ping_period: Option<u64>,
54}
55
56
57const DEFAULT_PING_INTERVAL_S: u64 = 30;
58
59const DEFAULT_XTB_REAL: &'static str = "wss://ws.xtb.com/real";
60const DEFAULT_XTB_REAL_STREAM: &'static str = "wss://ws.xtb.com/realStream";
61const DEFAULT_XTB_DEMO: &'static str = "wss://ws.xtb.com/demo";
62const DEFAULT_XTB_DEMO_STREAM: &'static str = "wss://ws.xtb.com/demoStream";
63
64
65impl XtbClientBuilder {
66    /// Create new builder using custom server urls.
67    ///
68    /// The resulting builder instance can be instantly built into the `XtbClient`
69    pub fn new(api_url: &str, stream_api_url: &str) -> Self {
70        Self {
71            api_url: Some(api_url.to_string()),
72            stream_api_url: Some(stream_api_url.to_string()),
73            app_id: None,
74            app_name: None,
75            ping_period: None,
76        }
77    }
78
79    /// Create new builder without any configuration.
80    ///
81    /// The `api_url` and the `stream_api_url` must be set at least.
82    pub fn new_bare() -> Self {
83        return Self {
84            api_url: None,
85            stream_api_url: None,
86            app_id: None,
87            app_name: None,
88            ping_period: None,
89        }
90    }
91
92    /// Shorthand for `XtbClientBuilder::new("wss://ws.xtb.com/real", "wss://ws.xtb.com/realStream")`
93    pub fn new_real() -> Self {
94        Self::new(DEFAULT_XTB_REAL, DEFAULT_XTB_REAL_STREAM)
95    }
96
97
98    /// Shorthand for `XtbClientBuilder::new("wss://ws.xtb.com/demo", "wss://ws.xtb.com/demoStream")`
99    pub fn new_demo() -> Self {
100        Self::new(DEFAULT_XTB_DEMO, DEFAULT_XTB_DEMO_STREAM)
101    }
102
103    /// Consume the builder instance and create instance of the `XtbClient`.
104    ///
105    /// The login is performed in this step.
106    ///
107    /// # Returns
108    ///
109    /// * `Ok(XtbClient)` - connected to the servers. The credentials given as arguments are
110    /// used to log in an user.
111    /// * `Err(XtbClintBuilderError)` - cannot connect to the servers.
112    ///
113    /// Common reasons of an errors are:
114    ///
115    /// * Incorrect credentials.
116    /// * Malformed API servers urls.
117    /// * Missing required configuration value.
118    pub async fn build(self, user_id: &str, password: &str) -> Result<XtbClient, XtbClientBuilderError> {
119        let api_url = Self::make_url(self.api_url)?;
120        let stream_api_url = Self::make_url(self.stream_api_url)?;
121
122        // create connection and perform login
123        let mut connection = BasicXtbConnection::new(api_url).await.map_err(|err| XtbClientBuilderError::CannotMakeConnection(err))?;
124        let mut login_request = LoginRequest::default().with_user_id(user_id).with_password(password);
125
126        if let Some(app_id) = self.app_id {
127            login_request = login_request.with_app_id(app_id);
128        }
129        if let Some(app_name) = self.app_name {
130            login_request = login_request.with_app_name(app_name);
131        }
132
133        let login_request_value = to_value(login_request).map_err(|err| XtbClientBuilderError::UnexpectedError(format!("{:?}", err)))?;
134
135        let response = connection
136            .send_command(COMMAND_LOGIN, Some(login_request_value)).await
137            .map_err(|err| XtbClientBuilderError::UnexpectedError(format!("{:?}", err)))?.await
138            .map_err(|err| XtbClientBuilderError::UnexpectedError(format!("{:?}", err)))?;
139
140        let stream_session_id = match response {
141            ProcessedMessage::ErrorResponse(msg) => return Err(XtbClientBuilderError::LoginFailed { user_id: user_id.to_string(), extra_info: format!("{:?}", msg) }),
142            ProcessedMessage::Response(response) => response.stream_session_id.unwrap(),
143        };
144
145        let stream_connection = BasicXtbStreamConnection::new(stream_api_url, stream_session_id).await.map_err(|err| XtbClientBuilderError::CannotMakeStreamConnection(err))?;
146
147        Ok(XtbClient::new(connection, stream_connection, self.ping_period.unwrap_or(DEFAULT_PING_INTERVAL_S)))
148    }
149
150    /// Convert string into an `Url` instance. This method is also used for validation of url presence.
151    ///
152    /// # Returns
153    ///
154    /// * `Ok(Url)` with correctly parsed url.
155    /// * `Err(XtbClientBuilderError::RequiredFieldMissing)` if argument is `None`
156    /// * `Err(XtbClientBuilderError::InvalidUrl)` if the url is malformed (cannot be parsed).
157    fn make_url(source: Option<String>) -> Result<Url, XtbClientBuilderError> {
158        let source_str = source.ok_or_else(|| XtbClientBuilderError::RequiredFieldMissing("api_url".to_owned()))?;
159        Url::from_str(&source_str).map_err(|err| XtbClientBuilderError::InvalidUrl(source_str, err))
160    }
161}
162
163
164#[derive(Debug, Error)]
165pub enum XtbClientBuilderError {
166    #[error("Required configuration field is missing: {0}")]
167    RequiredFieldMissing(String),
168    #[error("Url is invalid or malformed: {0} ({1})")]
169    InvalidUrl(String, url::ParseError),
170    #[error("Cannot connect to server")]
171    CannotMakeConnection(BasicXtbConnectionError),
172    #[error("Cannot connect to stream server")]
173    CannotMakeStreamConnection(BasicXtbStreamConnectionError),
174    #[error("Login failed for user: {user_id} ({extra_info:?})")]
175    LoginFailed { user_id: String, extra_info: String },
176    #[error("Something gets horribly wrong: {0}")]
177    UnexpectedError(String),
178}
179
180
181/// Declaration of the Request/response API interface.
182#[async_trait]
183pub trait RequestResponseApi {
184    /// Error returned from methods when command failed
185    type Error;
186
187    /// Returns array of all symbols available for the user.
188    async fn get_all_symbols(&mut self, request: GetAllSymbolsRequest) -> Result<GetAllSymbolsResponse, Self::Error>;
189
190    /// Returns calendar with market events.
191    async fn get_calendar(&mut self, request: GetCalendarRequest) -> Result<GetCalendarResponse, Self::Error>;
192
193    /// Please note that this function can be usually replaced by its streaming equivalent
194    /// getCandles which is the preferred way of retrieving current candle data. Returns chart info,
195    /// from start date to the current time. If the chosen period of CHART_LAST_INFO_RECORD is
196    /// greater than 1 minute, the last candle returned by the API can change until the end of the
197    /// period (the candle is being automatically updated every minute).
198    ///
199    /// Limitations: there are limitations in charts data availability. Detailed ranges for charts
200    /// data, what can be accessed with specific period, are as follows:
201    ///
202    /// * PERIOD_M1 --- <0-1) month, i.e. one-month time
203    /// * PERIOD_M30 --- <1-7) month, six months time
204    /// * PERIOD_H4 --- <7-13) month, six months time
205    /// * PERIOD_D1 --- 13 month, and earlier on
206    ///
207    /// Note, that specific PERIOD_ is the lowest (i.e. the most detailed) period, accessible
208    /// in listed range. For instance, in months range <1-7) you can access periods: PERIOD_M30,
209    /// PERIOD_H1, PERIOD_H4, PERIOD_D1, PERIOD_W1, PERIOD_MN1. Specific data ranges availability
210    /// is guaranteed, however those ranges may be wider, e.g.: PERIOD_M1 may be accessible
211    /// for 1.5 months back from now, where 1.0 months is guaranteed.
212    ///
213    /// Example scenario:
214    ///
215    /// * request charts of 5 minutes period, for 3 months time span, back from now;
216    /// * response: you are guaranteed to get 1 month of 5 minutes charts; because, 5 minutes period
217    /// charts are not accessible 2 months and 3 months back from now.
218    async fn get_chart_last_request(&mut self, request: GetChartLastRequestRequest) -> Result<GetChartLastRequestResponse, Self::Error>;
219
220    /// Please note that this function can be usually replaced by its streaming equivalent
221    /// getCandles which is the preferred way of retrieving current candle data. Returns chart info
222    /// with data between given start and end dates.
223    ///
224    /// Limitations: there are limitations in charts data availability. Detailed ranges for charts
225    /// data, what can be accessed with specific period, are as follows:
226    ///
227    /// * PERIOD_M1 --- <0-1) month, i.e. one month time
228    /// * PERIOD_M30 --- <1-7) month, six months time
229    /// * PERIOD_H4 --- <7-13) month, six months time
230    /// * PERIOD_D1 --- 13 month, and earlier on
231    ///
232    /// Note, that specific PERIOD_ is the lowest (i.e. the most detailed) period, accessible
233    /// in listed range. For instance, in months range <1-7) you can access periods: PERIOD_M30,
234    /// PERIOD_H1, PERIOD_H4, PERIOD_D1, PERIOD_W1, PERIOD_MN1. Specific data ranges availability
235    /// is guaranteed, however those ranges may be wider, e.g.: PERIOD_M1 may be accessible
236    /// for 1.5 months back from now, where 1.0 months is guaranteed.
237    async fn get_chart_range_request(&mut self, request: GetChartRangeRequestRequest) -> Result<GetChartRangeRequestResponse, Self::Error>;
238
239    /// Returns calculation of commission and rate of exchange. The value is calculated as expected
240    /// value, and therefore might not be perfectly accurate.
241    async fn get_commission_def(&mut self, request: GetCommissionDefRequest) -> Result<GetCommissionDefResponse, Self::Error>;
242
243    /// Returns information about account currency, and account leverage.
244    async fn get_current_user_data(&mut self, request: GetCurrentUserDataRequest) -> Result<GetCurrentUserDataResponse, Self::Error>;
245
246    /// Returns IBs data from the given time range.
247    async fn get_ibs_history(&mut self, request: GetIbsHistoryRequest) -> Result<GetIbsHistoryResponse, Self::Error>;
248
249    /// Please note that this function can be usually replaced by its streaming equivalent
250    /// getBalance which is the preferred way of retrieving account indicators. Returns various
251    /// account indicators.
252    async fn get_margin_level(&mut self, request: GetMarginLevelRequest) -> Result<GetMarginLevelResponse, Self::Error>;
253
254    /// Returns expected margin for given instrument and volume. The value is calculated as expected
255    /// margin value, and therefore might not be perfectly accurate.
256    async fn get_margin_trade(&mut self, request: GetMarginTradeRequest) -> Result<GetMarginTradeResponse, Self::Error>;
257
258    /// Please note that this function can be usually replaced by its streaming equivalent getNews
259    /// which is the preferred way of retrieving news data. Returns news from trading server which
260    /// were sent within specified period of time.
261    async fn get_news(&mut self, request: GetNewsRequest) -> Result<GetNewsResponse, Self::Error>;
262
263    /// Calculates estimated profit for given deal data Should be used for calculator-like apps
264    /// only. Profit for opened transactions should be taken from server, due to higher precision of
265    /// server calculation.
266    async fn get_profit_calculation(&mut self, request: GetProfitCalculationRequest) -> Result<GetProfitCalculationResponse, Self::Error>;
267
268    /// Returns current time on trading server.
269    async fn get_server_time(&mut self, request: GetServerTimeRequest) -> Result<GetServerTimeResponse, Self::Error>;
270
271    /// Returns a list of step rules for DMAs.
272    async fn get_step_rules(&mut self, request: GetStepRulesRequest) -> Result<GetStepRulesResponse, Self::Error>;
273
274    /// Returns information about symbol available for the user.
275    async fn get_symbol(&mut self, request: GetSymbolRequest) -> Result<GetSymbolResponse, Self::Error>;
276
277    /// Please note that this function can be usually replaced by its streaming equivalent
278    /// getTickPrices which is the preferred way of retrieving ticks data. Returns array of current
279    /// quotations for given symbols, only quotations that changed from given timestamp are
280    /// returned. New timestamp obtained from output will be used as an argument of the next call
281    /// of this command.
282    async fn get_tick_prices(&mut self, request: GetTickPricesRequest) -> Result<GetTickPricesResponse, Self::Error>;
283
284    /// Returns array of trades listed in orders argument.
285    async fn get_trade_records(&mut self, request: GetTradeRecordsRequest) -> Result<GetTradeRecordsResponse, Self::Error>;
286
287    /// Please note that this function can be usually replaced by its streaming equivalent getTrades
288    /// which is the preferred way of retrieving trades data. Returns array of user's trades.
289    async fn get_trades(&mut self, request: GetTradesRequest) -> Result<GetTradesResponse, Self::Error>;
290
291    /// Please note that this function can be usually replaced by its streaming equivalent getTrades
292    /// which is the preferred way of retrieving trades data. Returns array of user's trades which
293    /// were closed within specified period of time.
294    async fn get_trades_history(&mut self, request: GetTradesHistoryRequest) -> Result<GetTradesHistoryResponse, Self::Error>;
295
296    /// Returns quotes and trading times.
297    async fn get_trading_hours(&mut self, request: GetTradingHoursRequest) -> Result<GetTradingHoursResponse, Self::Error>;
298
299    /// Returns the current API version.
300    async fn get_version(&mut self, request: GetVersionRequest) -> Result<GetVersionResponse, Self::Error>;
301
302    /// Starts trade transaction. tradeTransaction sends main transaction information to the server.
303    ///
304    /// # Note
305    ///
306    /// How to verify that the trade request was accepted?
307    /// The status field set to 'true' does not imply that the transaction was accepted. It only
308    /// means, that the server acquired your request and began to process it. To analyse the status
309    /// of the transaction (for example to verify if it was accepted or rejected) use the
310    /// tradeTransactionStatus command with the order number, that came back with the response of
311    /// the tradeTransaction command. You can find the example here:
312    /// https://developers.xstore.pro/api/tutorials/opening_and_closing_trades2
313    async fn trade_transaction(&mut self, request: TradeTransactionRequest) -> Result<TradeTransactionResponse, Self::Error>;
314
315    /// Description: Please note that this function can be usually replaced by its streaming
316    /// equivalent getTradeStatus which is the preferred way of retrieving transaction status data.
317    /// Returns current transaction status. At any time of transaction processing client might check
318    /// the status of transaction on server side. In order to do that client must provide unique
319    /// order taken from tradeTransaction invocation.
320    async fn trade_transaction_status(&mut self, request: TradeTransactionStatusRequest) -> Result<TradeTransactionStatusResponse, Self::Error>;
321}
322
323
324/// Declaration of the stream API interface
325#[async_trait]
326pub trait StreamApi {
327    /// Error returned from the client when something went wrong
328    type Error;
329
330    type Stream<T: Send + Sync + for<'de> Deserialize<'de>>;
331
332    /// Each streaming command takes as an argument streamSessionId which is sent in response
333    /// message for login command performed in main connection. streamSessionId token allows to
334    /// identify user in streaming connection. In one streaming connection multiple commands with
335    /// different streamSessionId can be invoked. It will cause sending streaming data for multiple
336    /// login sessions in one streaming connection. streamSessionId is valid until logout command is
337    /// performed on main connection or main connection is disconnected.
338    async fn subscribe_balance(&mut self, arguments: StreamGetBalanceSubscribe) -> Result<Self::Stream<StreamGetBalanceData>, Self::Error>;
339
340    /// Subscribes for and unsubscribes from API chart candles. The interval of every candle
341    /// is 1 minute. A new candle arrives every minute.
342    async fn subscribe_candles(&mut self, arguments: StreamGetCandlesSubscribe) -> Result<Self::Stream<StreamGetCandlesData>, Self::Error>;
343
344    /// Subscribes for and unsubscribes from 'keep alive' messages. A new 'keep alive' message
345    /// is sent by the API every 3 seconds.
346    async fn subscribe_keep_alive(&mut self, arguments: StreamGetKeepAliveSubscribe) -> Result<Self::Stream<StreamGetKeepAliveData>, Self::Error>;
347
348    /// Subscribes for and unsubscribes from news.
349    async fn subscribe_news(&mut self, arguments: StreamGetNewsSubscribe) -> Result<Self::Stream<StreamGetNewsData>, Self::Error>;
350
351    /// Subscribes for and unsubscribes from profits.
352    async fn subscribe_profits(&mut self, arguments: StreamGetProfitSubscribe) -> Result<Self::Stream<StreamGetProfitData>, Self::Error>;
353
354    /// Establishes subscription for quotations and allows to obtain the relevant information
355    /// in real-time, as soon as it is available in the system. The getTickPrices command can
356    /// be invoked many times for the same symbol, but only one subscription for a given symbol
357    /// will be created. Please beware that when multiple records are available, the order in which
358    /// they are received is not guaranteed.
359    async fn subscribe_tick_prices(&mut self, arguments: StreamGetTickPricesSubscribe) -> Result<Self::Stream<StreamGetTickPricesData>, Self::Error>;
360
361    /// Establishes subscription for user trade status data and allows to obtain the relevant
362    /// information in real-time, as soon as it is available in the system. Please beware that when
363    /// multiple records are available, the order in which they are received is not guaranteed.
364    async fn subscribe_trades(&mut self, arguments: StreamGetTradesSubscribe) -> Result<Self::Stream<StreamGetTradesData>, Self::Error>;
365
366    /// Allows to get status for sent trade requests in real-time, as soon as it is available
367    /// in the system. Please beware that when multiple records are available, the order in which
368    /// they are received is not guaranteed.
369    async fn subscribe_trade_status(&mut self, arguments: StreamGetTradeStatusSubscribe) -> Result<Self::Stream<StreamGetTradeStatusData>, Self::Error>;
370}
371
372
373/// Implementor of the API traits.
374///
375/// This struct is designed to be an interface between user (application) and XTB API servers.
376///
377/// The `XtbClient` is responsible for sending and receiving pings and logout when instance is dropped.
378pub struct XtbClient {
379    /// Connection to the request/response server
380    connection: Arc<Mutex<BasicXtbConnection>>,
381    /// Connection to the stream server
382    stream_manager: StreamManager,
383    /// handle of the request/response server ping worker
384    ping_join_handle: JoinHandle<()>,
385    /// handle of the stream server ping worker
386    stream_ping_join_handle: JoinHandle<()>,
387}
388
389
390impl XtbClient {
391    /// Create builder for the `XtbClient`.
392    ///
393    /// When builder is configured, the `build()` method can be called with credentials passed to the
394    /// method arguments. This call create new `XtbClient` instance.
395    pub fn builder() -> XtbClientBuilder {
396        XtbClientBuilder::default()
397    }
398
399    /// Create new instance of the `XtbClient`. it expects connected and logged connections to the
400    /// request/response and the stream server.
401    ///
402    /// # Note
403    ///
404    /// The login is performed by the builder because the stream server implementation needs to know
405    /// a stream session id which is provided by the `login` command.
406    pub fn new(connection: BasicXtbConnection, stream_connection: BasicXtbStreamConnection, ping_period: u64) -> Self {
407        let connection = Arc::new(Mutex::new(connection));
408
409        let ping_join_handle = spawn_ping(connection.clone(), ping_period);
410
411        let stream_manager = StreamManager::new(stream_connection);
412        let stream_ping_join_handle = spawn_stream_ping(stream_manager.clone(), ping_period);
413
414        Self {
415            connection,
416            stream_manager,
417            ping_join_handle,
418            stream_ping_join_handle,
419        }
420    }
421
422    /// Send command to the server and wait for response.
423    ///
424    /// If command does not return any response, create default one with type of `RESP`.
425    async fn send_and_wait_or_default<REQ, RESP>(&mut self, command: &str, request: REQ) -> Result<RESP, XtbClientError>
426        where
427            REQ: Serialize,
428            RESP: for<'de> Deserialize<'de> + Default {
429        self.send_and_wait(command, request).await.map(|val| val.unwrap_or_default())
430    }
431
432    /// Send the command and wait for a response.
433    async fn send_and_wait<REQ, RESP>(&mut self, command: &str, request: REQ) -> Result<Option<RESP>, XtbClientError>
434        where
435            REQ: Serialize,
436            RESP: for<'de> Deserialize<'de>
437    {
438        let promise = self.send(command, request).await?;
439        let response = promise.await.map_err(|err| {
440            error!("Unexpected error: {:?}", err);
441            XtbClientError::UnexpectedError
442        })?;
443        match response {
444            ProcessedMessage::Response(response) => {
445                match response.return_data {
446                    Some(data) => from_value(data).map_err(|err| XtbClientError::DeserializationFailed(err)).map(|v| Some(v)),
447                    None => Ok(None)
448                }
449            }
450            ProcessedMessage::ErrorResponse(err) => Err(XtbClientError::CommandFailed(err)),
451        }
452    }
453
454    /// Send a command request to the server and return `Ok(ResponsePromise)` o
455    async fn send<A>(&mut self, command: &str, request: A) -> Result<ResponsePromise, XtbClientError>
456        where
457            A: Serialize
458    {
459        let mut conn = self.connection.lock().await;
460        let payload = Self::convert_data_to_value(request)?;
461        conn.send_command(command, Some(payload)).await.map_err(|err| {
462            match err {
463                BasicXtbConnectionError::SerializationError(err) => XtbClientError::SerializationFailed(err),
464                BasicXtbConnectionError::CannotSendRequest(err) => XtbClientError::CannotSendCommand(err),
465                _ => XtbClientError::UnexpectedError,
466            }
467        })
468    }
469
470    /// Serialize payload data into value.
471    ///
472    /// # Returns
473    ///
474    /// * `Ok(Value)` when data was serialized successfully.
475    /// * `Err(XtbClientError::SerializationFailed)` - data cannot be serialized.
476    fn convert_data_to_value<T: Serialize>(data: T) -> Result<Value, XtbClientError> {
477        to_value(data).map_err(|err| XtbClientError::SerializationFailed(err))
478    }
479
480    /// Send stream command to the stream API server.
481    ///
482    /// # Parameters
483    ///
484    /// * `subscribe_command` - command name of the subscribe command (e.g. `getCandles`)
485    /// * `subscribe_arguments` - arguments for the subscribe command
486    /// * `unsubscribe_command` - command name of the unsubscribe command (e.g. `stopCandles`)
487    /// * `unsubscribe_arguments` - arguments for the unsubscribe command
488    /// * `data_command` - command name in data messages (e.g. `candle`)
489    ///
490    /// # Returns
491    ///
492    /// * `Ok<DataStream<T>>` - data stream with filter set to messages related to sent command
493    /// * `Err<XtbClientError>` - unable to send command
494    async fn send_simple_stream_command<T, SA, UA>(
495        &mut self,
496        subscribe_command: &str,
497        subscribe_arguments: SA,
498        unsubscribe_command: &str,
499        unsubscribe_arguments: UA,
500        data_command: &str,
501    ) -> Result<DataStream<T>, XtbClientError>
502        where
503            T: for<'de> Deserialize<'de> + Send + Sync,
504            SA: Serialize,
505            UA: Serialize,
506    {
507        let unsubscribe_arguments = Self::convert_data_to_value(unsubscribe_arguments)?;
508        let filter = DataMessageFilter::Command(data_command.to_owned());
509        let subscribe_arguments = Self::convert_data_to_value(subscribe_arguments)?;
510        self.stream_manager.subscribe(subscribe_command, Some(subscribe_arguments), unsubscribe_command, Some(unsubscribe_arguments), data_command, filter).await
511    }
512
513    /// Send stream command to the stream API server and add filter by the `symbol` field to the
514    /// data stream.
515    ///
516    /// # Parameters
517    ///
518    /// * `subscribe_command` - command name of the subscribe command (e.g. `getCandles`)
519    /// * `subscribe_arguments` - arguments for the subscribe command
520    /// * `unsubscribe_command` - command name of the unsubscribe command (e.g. `stopCandles`)
521    /// * `unsubscribe_arguments` - arguments for the unsubscribe command
522    /// * `data_command` - command name in data messages (e.g. `candle`)
523    ///
524    /// # Returns
525    ///
526    /// * `Ok<DataStream<T>>` - data stream with filter set to messages related to sent command
527    /// * `Err<XtbClientError>` - unable to send command
528    async fn send_symbol_scoped_stream_command<T, SA, UA>(
529        &mut self,
530        subscribe_command: &str,
531        subscribe_arguments: SA,
532        unsubscribe_command: &str,
533        unsubscribe_arguments: UA,
534        data_command: &str,
535        symbol: &str,
536    ) -> Result<DataStream<T>, XtbClientError>
537        where
538            T: for<'de> Deserialize<'de> + Send + Sync,
539            SA: Serialize,
540            UA: Serialize,
541    {
542        let unsubscribe_arguments = Self::convert_data_to_value(unsubscribe_arguments)?;
543        let subscribe_arguments = Self::convert_data_to_value(subscribe_arguments)?;
544        let subscription_key = format!("{}.{}", data_command, symbol);
545
546        let filter = DataMessageFilter::All(vec![
547            DataMessageFilter::Command(data_command.to_owned()),
548            DataMessageFilter::FieldValue { name: "symbol".to_owned(), value: Value::String(symbol.to_owned()) },
549        ]);
550        self.stream_manager.subscribe(subscribe_command, Some(subscribe_arguments), unsubscribe_command, Some(unsubscribe_arguments), &subscription_key, filter).await
551    }
552}
553
554
555impl Drop for XtbClient {
556    fn drop(&mut self) {
557        self.ping_join_handle.abort();
558        self.stream_ping_join_handle.abort();
559    }
560}
561
562
563#[async_trait]
564impl RequestResponseApi for XtbClient {
565    type Error = XtbClientError;
566
567    async fn get_all_symbols(&mut self, request: GetAllSymbolsRequest) -> Result<GetAllSymbolsResponse, Self::Error> {
568        self.send_and_wait_or_default(COMMAND_GET_ALL_SYMBOLS, request).await
569    }
570
571    async fn get_calendar(&mut self, request: GetCalendarRequest) -> Result<GetCalendarResponse, Self::Error> {
572        self.send_and_wait_or_default(COMMAND_GET_CALENDAR, request).await
573    }
574
575    async fn get_chart_last_request(&mut self, request: GetChartLastRequestRequest) -> Result<GetChartLastRequestResponse, Self::Error> {
576        self.send_and_wait_or_default(COMMAND_GET_CHART_LAST_REQUEST, request).await
577    }
578
579    async fn get_chart_range_request(&mut self, request: GetChartRangeRequestRequest) -> Result<GetChartRangeRequestResponse, Self::Error> {
580        self.send_and_wait_or_default(COMMAND_GET_CHART_RANGE_REQUEST, request).await
581    }
582
583    async fn get_commission_def(&mut self, request: GetCommissionDefRequest) -> Result<GetCommissionDefResponse, Self::Error> {
584        self.send_and_wait_or_default(COMMAND_GET_COMMISSION_DEF, request).await
585    }
586
587    async fn get_current_user_data(&mut self, request: GetCurrentUserDataRequest) -> Result<GetCurrentUserDataResponse, Self::Error> {
588        self.send_and_wait_or_default(COMMAND_GET_CURRENT_USER_DATA, request).await
589    }
590
591    async fn get_ibs_history(&mut self, request: GetIbsHistoryRequest) -> Result<GetIbsHistoryResponse, Self::Error> {
592        self.send_and_wait_or_default(COMMAND_GET_IBS_HISTORY, request).await
593    }
594
595    async fn get_margin_level(&mut self, request: GetMarginLevelRequest) -> Result<GetMarginLevelResponse, Self::Error> {
596        self.send_and_wait_or_default(COMMAND_GET_MARGIN_LEVEL, request).await
597    }
598
599    async fn get_margin_trade(&mut self, request: GetMarginTradeRequest) -> Result<GetMarginTradeResponse, Self::Error> {
600        self.send_and_wait_or_default(COMMAND_GET_MARGIN_TRADE, request).await
601    }
602
603    async fn get_news(&mut self, request: GetNewsRequest) -> Result<GetNewsResponse, Self::Error> {
604        self.send_and_wait_or_default(COMMAND_GET_NEWS, request).await
605    }
606
607    async fn get_profit_calculation(&mut self, request: GetProfitCalculationRequest) -> Result<GetProfitCalculationResponse, Self::Error> {
608        self.send_and_wait_or_default(COMMAND_GET_PROFIT_CALCULATION, request).await
609    }
610
611    async fn get_server_time(&mut self, request: GetServerTimeRequest) -> Result<GetServerTimeResponse, Self::Error> {
612        self.send_and_wait_or_default(COMMAND_GET_SERVER_TIME, request).await
613    }
614
615    async fn get_step_rules(&mut self, request: GetStepRulesRequest) -> Result<GetStepRulesResponse, Self::Error> {
616        self.send_and_wait_or_default(COMMAND_GET_STEP_RULES, request).await
617    }
618
619    async fn get_symbol(&mut self, request: GetSymbolRequest) -> Result<GetSymbolResponse, Self::Error> {
620        self.send_and_wait_or_default(COMMAND_GET_SYMBOL, request).await
621    }
622
623    async fn get_tick_prices(&mut self, request: GetTickPricesRequest) -> Result<GetTickPricesResponse, Self::Error> {
624        self.send_and_wait_or_default(COMMAND_GET_TICK_PRICES, request).await
625    }
626
627    async fn get_trade_records(&mut self, request: GetTradeRecordsRequest) -> Result<GetTradeRecordsResponse, Self::Error> {
628        self.send_and_wait_or_default(COMMAND_GET_TRADE_RECORDS, request).await
629    }
630
631    async fn get_trades(&mut self, request: GetTradesRequest) -> Result<GetTradesResponse, Self::Error> {
632        self.send_and_wait_or_default(COMMAND_GET_TRADES, request).await
633    }
634
635    async fn get_trades_history(&mut self, request: GetTradesHistoryRequest) -> Result<GetTradesHistoryResponse, Self::Error> {
636        self.send_and_wait_or_default(COMMAND_GET_TRADES_HISTORY, request).await
637    }
638
639    async fn get_trading_hours(&mut self, request: GetTradingHoursRequest) -> Result<GetTradingHoursResponse, Self::Error> {
640        self.send_and_wait_or_default(COMMAND_GET_TRADING_HOURS, request).await
641    }
642
643    async fn get_version(&mut self, request: GetVersionRequest) -> Result<GetVersionResponse, Self::Error> {
644        self.send_and_wait_or_default(COMMAND_GET_VERSION, request).await
645    }
646
647    async fn trade_transaction(&mut self, request: TradeTransactionRequest) -> Result<TradeTransactionResponse, Self::Error> {
648        self.send_and_wait_or_default(COMMAND_TRADE_TRANSACTION, request).await
649    }
650
651    async fn trade_transaction_status(&mut self, request: TradeTransactionStatusRequest) -> Result<TradeTransactionStatusResponse, Self::Error> {
652        self.send_and_wait_or_default(COMMAND_TRADE_TRANSACTION_STATUS, request).await
653    }
654}
655
656
657#[async_trait]
658impl StreamApi for XtbClient {
659    type Error = XtbClientError;
660
661    type Stream<T: Send + Sync + for<'de> Deserialize<'de>> = DataStream<T>;
662
663    async fn subscribe_balance(&mut self, arguments: StreamGetBalanceSubscribe) -> Result<Self::Stream<StreamGetBalanceData>, Self::Error> {
664        let stop_arguments = Self::convert_data_to_value(StreamGetBalanceUnsubscribe::default())?;
665        self.send_simple_stream_command(STREAM_BALANCE_SUBSCRIBE, arguments, STREAM_BALANCE_UNSUBSCRIBE, stop_arguments, STREAM_BALANCE).await
666    }
667
668    async fn subscribe_candles(&mut self, arguments: StreamGetCandlesSubscribe) -> Result<Self::Stream<StreamGetCandlesData>, Self::Error> {
669        let stop_arguments = Self::convert_data_to_value(StreamGetCandlesUnsubscribe::default().with_symbol(&arguments.symbol))?;
670        let symbol = arguments.symbol.clone();
671        self.send_symbol_scoped_stream_command(STREAM_CANDLES_SUBSCRIBE, arguments, STREAM_CANDLES_UNSUBSCRIBE, stop_arguments, STREAM_CANDLES, &symbol).await
672    }
673
674    async fn subscribe_keep_alive(&mut self, arguments: StreamGetKeepAliveSubscribe) -> Result<Self::Stream<StreamGetKeepAliveData>, Self::Error> {
675        let stop_arguments = Self::convert_data_to_value(StreamGetKeepAliveUnsubscribe::default())?;
676        self.send_simple_stream_command(STREAM_KEEP_ALIVE_SUBSCRIBE, arguments, STREAM_KEEP_ALIVE_UNSUBSCRIBE, stop_arguments, STREAM_KEEP_ALIVE).await
677    }
678
679    async fn subscribe_news(&mut self, arguments: StreamGetNewsSubscribe) -> Result<Self::Stream<StreamGetNewsData>, Self::Error> {
680        let stop_arguments = Self::convert_data_to_value(StreamGetNewsUnsubscribe::default())?;
681        self.send_simple_stream_command(STREAM_NEWS_SUBSCRIBE, arguments, STREAM_NEWS_UNSUBSCRIBE, stop_arguments, STREAM_NEWS).await
682    }
683
684    async fn subscribe_profits(&mut self, arguments: StreamGetProfitSubscribe) -> Result<Self::Stream<StreamGetProfitData>, Self::Error> {
685        let stop_arguments = Self::convert_data_to_value(StreamGetProfitUnsubscribe::default())?;
686        self.send_simple_stream_command(STREAM_PROFITS_SUBSCRIBE, arguments, STREAM_PROFITS_UNSUBSCRIBE, stop_arguments, STREAM_PROFITS).await
687    }
688
689    async fn subscribe_tick_prices(&mut self, arguments: StreamGetTickPricesSubscribe) -> Result<Self::Stream<StreamGetTickPricesData>, Self::Error> {
690        let stop_arguments = Self::convert_data_to_value(StreamGetTickPricesUnsubscribe::default().with_symbol(&arguments.symbol))?;
691        let symbol = arguments.symbol.clone();
692        self.send_symbol_scoped_stream_command(STREAM_TICK_PRICES_SUBSCRIBE, arguments, STREAM_TICK_PRICES_UNSUBSCRIBE, stop_arguments, STREAM_TICK_PRICES, &symbol).await
693    }
694
695    async fn subscribe_trades(&mut self, arguments: StreamGetTradesSubscribe) -> Result<Self::Stream<StreamGetTradesData>, Self::Error> {
696        let stop_arguments = Self::convert_data_to_value(StreamGetTradesUnsubscribe::default())?;
697        self.send_simple_stream_command(STREAM_TRADES_SUBSCRIBE, arguments, STREAM_TRADES_UNSUBSCRIBE, stop_arguments, STREAM_TRADES).await
698    }
699
700    async fn subscribe_trade_status(&mut self, arguments: StreamGetTradeStatusSubscribe) -> Result<Self::Stream<StreamGetTradeStatusData>, Self::Error> {
701        let stop_arguments = Self::convert_data_to_value(StreamGetTradeStatusUnsubscribe::default())?;
702        self.send_simple_stream_command(STREAM_TRADE_STATUS_SUBSCRIBE, arguments, STREAM_TRADE_STATUS_UNSUBSCRIBE, stop_arguments, STREAM_TRADE_STATUS).await
703    }
704}
705
706
707#[derive(Debug, Error)]
708pub enum XtbClientError {
709    #[error("Cannot serialize arguments")]
710    SerializationFailed(serde_json::Error),
711    #[error("Cannot send command to server")]
712    CannotSendCommand(tokio_tungstenite::tungstenite::Error),
713    #[error("Cannot send stream command")]
714    CannotSendStreamCommand(BasicXtbStreamConnectionError),
715    #[error("Unexpected error.")]
716    UnexpectedError,
717    #[error("Cannot deserialize data")]
718    DeserializationFailed(serde_json::Error),
719    #[error("Command failed and an error response was returned")]
720    CommandFailed(ErrorResponse),
721}
722
723
724/// Shared inner state of the `StreamManager`
725#[derive(Debug)]
726struct StreamManagerState {
727    /// The stream connection
728    connection: BasicXtbStreamConnection,
729    /// subscription counter
730    subscriptions: HashMap<String, usize>,
731}
732
733
734impl StreamManagerState {
735    /// Create new instance of the struct
736    pub fn new(connection: BasicXtbStreamConnection) -> Self {
737        Self {
738            connection,
739            subscriptions: HashMap::new(),
740        }
741    }
742}
743
744
745/// Manage stream subscriptions across application. All instances cloned from same origin share
746/// its internal state.
747#[derive(Clone, Debug)]
748struct StreamManager {
749    /// The inner state shared between instances of the `StreamManager`
750    state: Arc<Mutex<StreamManagerState>>,
751}
752
753
754impl StreamManager {
755    /// Create new instance of the `StreamManager` struct.
756    pub fn new(connection: BasicXtbStreamConnection) -> Self {
757        let state = Arc::new(Mutex::new(StreamManagerState::new(connection)));
758        Self {
759            state
760        }
761    }
762
763    /// Subscribe for a stream from the stream API server.
764    ///
765    /// # Parameters
766    ///
767    /// * `subscribe_command` - command name of the subscribe command (e.g. `getCandles`)
768    /// * `subscribe_arguments` - arguments for the subscribe command
769    /// * `unsubscribe_command` - command name of the unsubscribe command (e.g. `stopCandles`)
770    /// * `unsubscribe_arguments` - arguments for the unsubscribe command
771    /// * `subscription_key` - key used to track number of subscribers of the data stream
772    /// * `filter` - the filter predicate for message routing.
773    ///
774    /// # Returns
775    ///
776    /// * `Ok<DataStream<T>>` - data stream with filter set to messages related to sent command
777    /// * `Err<XtbClientError>` - unable to send command
778    pub async fn subscribe<T: for<'de> Deserialize<'de> + Send + Sync>(
779        &mut self,
780        subscribe_command: &str,
781        subscribe_arguments: Option<Value>,
782        unsubscribe_command: &str,
783        unsubscribe_arguments: Option<Value>,
784        subscription_key: &str,
785        filter: DataMessageFilter,
786    ) -> Result<DataStream<T>, XtbClientError> {
787        let mut state = self.state.lock().await;
788        let stream = state.connection.make_message_stream(filter).await;
789        state.connection.subscribe(subscribe_command, subscribe_arguments).await.map_err(|err| XtbClientError::CannotSendStreamCommand(err))?;
790        *state.subscriptions.entry(subscription_key.to_owned()).or_default() += 1;
791        Ok(DataStream::new(stream, self.clone(), subscription_key.to_owned(), unsubscribe_command.to_owned(), unsubscribe_arguments))
792    }
793
794    /// Unsubscribe from a stream.
795    ///
796    /// # Parameters
797    ///
798    /// * `subscription_key` - the subscription key where number of subscribers is tracked
799    /// * `command` - an unsubscribe command to be sent to the server
800    /// * `arguments` - the unsubscribe command arguments
801    ///
802    /// # Returns
803    ///
804    /// * `Ok(())` - success
805    /// * `Err(XtbClientError::CannotSendStreamCommand)` - fail
806    pub async fn unsubscribe(&mut self, subscription_key: &str, command: &str, arguments: Option<Value>) -> Result<(), XtbClientError> {
807        let mut state = self.state.lock().await;
808        let entry = state.subscriptions.entry(subscription_key.to_owned()).or_default();
809        if *entry > 0 {
810            *entry -= 1;
811        }
812        if *entry == 0 {
813            state.connection.unsubscribe(command, arguments).await.map_err(|err| XtbClientError::CannotSendStreamCommand(err))?;
814        }
815        Ok(())
816    }
817}
818
819
820/// Stream of messages delivered to a consumer.
821///
822/// The message data is deserialized and typed to data type related to a command.
823pub struct DataStream<T>
824    where
825        T: for<'de> Deserialize<'de> + Send + Sync
826{
827    /// The message stream with raw messages
828    message_stream: BasicMessageStream,
829    /// The stream manager used to unsubscribe from a stream when struct is dropped
830    stream_manager: StreamManager,
831    /// Internal subscription key for subscriber tracking
832    subscription_key: String,
833    /// Unsubscribe command to be sent to the XTB server
834    unsubscribe_command: String,
835    /// Unsubscribe command arguments
836    unsubscribe_arguments: Option<Value>,
837    /// Data type returned to a consumer
838    type_: PhantomData<T>,
839}
840
841impl<T> DataStream<T>
842    where
843        T: for<'de> Deserialize<'de> + Send + Sync
844{
845    /// Create new instance of the stream.
846    fn new(message_stream: BasicMessageStream, stream_manager: StreamManager, subscription_key: String, unsubscribe_command: String, unsubscribe_arguments: Option<Value>) -> Self {
847        Self {
848            message_stream,
849            stream_manager,
850            subscription_key,
851            unsubscribe_command,
852            unsubscribe_arguments,
853            type_: PhantomData::<T>,
854        }
855    }
856
857    /// Wait and get next message from the stream.
858    ///
859    /// # Returns
860    ///
861    /// * `Ok(Some(T))` - next message in stream.
862    /// * `Ok(None)` - there is no message left
863    /// * `Err(DataStreamError)` - message was recived but cannot be processed. A next message can be ok.
864    pub async fn next(&mut self) -> Result<Option<T>, DataStreamError> {
865        let message = self.message_stream.next().await;
866        match message {
867            Some(msg) => Self::process_message(msg).map(|r| Some(r)),
868            None => Ok(None),
869        }
870    }
871
872    /// Deserialize serialized data representation to actual type `T`.
873    fn process_message(msg: StreamDataMessage) -> Result<T, DataStreamError> {
874        from_value(msg.data).map_err(|err| DataStreamError::CannotDeserializeValue(err))
875    }
876}
877
878impl<T> Drop for DataStream<T>
879    where
880        T: for<'de> Deserialize<'de> + Send + Sync
881{
882    fn drop(&mut self) {
883        let mut manager = self.stream_manager.clone();
884        let unsubscribe_command = self.unsubscribe_command.clone();
885        let unsubscribe_arguments = self.unsubscribe_arguments.take();
886        let subscription_key = self.subscription_key.clone();
887        spawn(async move {
888            let result = manager.unsubscribe(&subscription_key, &unsubscribe_command, unsubscribe_arguments.clone()).await;
889            match result {
890                Err(err) => error!("Cannot unsubscribe command '{unsubscribe_command}' ({unsubscribe_arguments:?}). The subscription key was: '{subscription_key}'. The error was: {err:?}"),
891                _ => (),
892            };
893        });
894    }
895}
896
897#[derive(Debug, Error)]
898pub enum DataStreamError {
899    #[error("Cannot deserialize value: {0}")]
900    CannotDeserializeValue(serde_json::Error)
901}
902
903
904/// Spawn tokio green thread and to send ping periodically to sync connection
905///
906/// # Arguments
907///
908/// * conn - the stream connection
909/// * ping_secs - number of seconds between each ping
910///
911/// # Panics
912///
913/// The ping message cannot be serialized. The serialization is done before the green thread is run
914///
915/// # Returns
916///
917/// `JoinHandle` of the green thread
918fn spawn_ping(conn: Arc<Mutex<BasicXtbConnection>>, ping_secs: u64) -> JoinHandle<()> {
919    let ping_value = to_value(PingRequest::default()).expect("Cannot serialize ping message");
920    spawn(async move {
921        let mut idx = 1u64;
922        loop {
923            let response_promise = {
924                let mut conn = conn.lock().await;
925                debug!("Sending ping #{} to connection", idx);
926                match conn.send_command(COMMAND_PING, Some(ping_value.clone())).await {
927                    Ok(resp) => Some(resp),
928                    Err(err) => {
929                        error!("Cannot send ping #{}: {:?}", idx, err);
930                        None
931                    }
932                }
933            };
934            if let Some(response_promise) = response_promise {
935                match response_promise.await {
936                    Ok(_) => (),
937                    Err(err) => error!("Cannot await the ping response #{}: {:?}", idx, err)
938                }
939            }
940            idx += 1;
941            sleep(Duration::from_secs(ping_secs)).await;
942        }
943    })
944}
945
946
947/// Spawn tokio green thread and to send ping periodically to stream connection
948///
949/// # Arguments
950///
951/// * conn - the stream connection
952/// * ping_secs - number of seconds between each ping
953///
954/// # Panics
955///
956/// The ping message cannot be serialized. The serialization is done before the green thread is run
957///
958/// # Returns
959///
960/// `JoinHandle` of the green thread
961fn spawn_stream_ping(stream_manager: StreamManager, ping_secs: u64) -> JoinHandle<()> {
962    let ping_value = to_value(StreamPingSubscribe::default()).expect("Cannot serialize the stream ping message");
963    spawn(async move {
964        let mut idx = 1u64;
965        loop {
966            {
967                debug!("Sending ping #{} to stream connection", idx);
968                let mut inner_state = stream_manager.state.lock().await;
969                match inner_state.connection.subscribe(STREAM_PING, Some(ping_value.clone())).await {
970                    Ok(_) => (),
971                    Err(err) => error!("Cannot send ping #{}: {:?}", idx, err)
972                }
973            }
974            idx += 1;
975            sleep(Duration::from_secs(ping_secs)).await;
976        }
977    })
978}
xtb_client/client.rs

xtb_client/
client.rs
