xtb_client/

client.rs

use std::collections::HashMap;
use std::marker::PhantomData;
use std::str::FromStr;
use std::sync::Arc;
use std::time::Duration;

use async_trait::async_trait;
use derive_setters::Setters;
use serde::{Deserialize, Serialize};
use serde_json::{from_value, to_value, Value};
use thiserror::Error;
use tokio::spawn;
use tokio::sync::Mutex;
use tokio::task::JoinHandle;
use tokio::time::sleep;
use tracing::{debug, error};
use url::Url;

use crate::{BasicMessageStream, BasicXtbConnection, BasicXtbStreamConnection, DataMessageFilter, MessageStream, ResponsePromise, XtbConnection, BasicXtbConnectionError, XtbStreamConnection, BasicXtbStreamConnectionError};
use crate::message_processing::ProcessedMessage;
use 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};


/// Builder for `XtbClient`.
///
/// Configuration can be set with this class and when configuration step is finished, the `XtbClient`
/// can be created by the `build()` method.
///
/// Configurable fields are:
///
/// * `api_url` - url of the request/response API server.
/// * `stream_api_url` - url of the stream API server.
/// * `app_id` - application identifier (deprecated by the official API documentation)
/// * `app_name` - application name (deprecated by the official API documentation)
/// * `ping_period` - interval between ping commands. Default interval is 30s.
///
/// The required configuration values are `api_url` and `stream_api_url`. Other values are optional.
///
/// Official documentation says, the `ping_interval` should be less than 10 minutes. But real world
/// observation shows that the maximal interval must be less than 1 minute.
#[derive(Default, Setters)]
#[setters(into, prefix = "with_", strip_option)]
pub struct XtbClientBuilder {
    /// Url of the request/response API server
    api_url: Option<String>,
    /// Url of the stream API server
    stream_api_url: Option<String>,
    /// Identifier of the application. (Deprecated by the official api)
    app_id: Option<String>,
    /// Name of the application (deprecated by the official api)
    app_name: Option<String>,
    /// Interval between pings. Shouldn't be greater than 1 minute.
    ping_period: Option<u64>,
}


const DEFAULT_PING_INTERVAL_S: u64 = 30;

const DEFAULT_XTB_REAL: &'static str = "wss://ws.xtb.com/real";
const DEFAULT_XTB_REAL_STREAM: &'static str = "wss://ws.xtb.com/realStream";
const DEFAULT_XTB_DEMO: &'static str = "wss://ws.xtb.com/demo";
const DEFAULT_XTB_DEMO_STREAM: &'static str = "wss://ws.xtb.com/demoStream";


impl XtbClientBuilder {
    /// Create new builder using custom server urls.
    ///
    /// The resulting builder instance can be instantly built into the `XtbClient`
    pub fn new(api_url: &str, stream_api_url: &str) -> Self {
        Self {
            api_url: Some(api_url.to_string()),
            stream_api_url: Some(stream_api_url.to_string()),
            app_id: None,
            app_name: None,
            ping_period: None,
        }
    }

    /// Create new builder without any configuration.
    ///
    /// The `api_url` and the `stream_api_url` must be set at least.
    pub fn new_bare() -> Self {
        return Self {
            api_url: None,
            stream_api_url: None,
            app_id: None,
            app_name: None,
            ping_period: None,
        }
    }

    /// Shorthand for `XtbClientBuilder::new("wss://ws.xtb.com/real", "wss://ws.xtb.com/realStream")`
    pub fn new_real() -> Self {
        Self::new(DEFAULT_XTB_REAL, DEFAULT_XTB_REAL_STREAM)
    }


    /// Shorthand for `XtbClientBuilder::new("wss://ws.xtb.com/demo", "wss://ws.xtb.com/demoStream")`
    pub fn new_demo() -> Self {
        Self::new(DEFAULT_XTB_DEMO, DEFAULT_XTB_DEMO_STREAM)
    }

    /// Consume the builder instance and create instance of the `XtbClient`.
    ///
    /// The login is performed in this step.
    ///
    /// # Returns
    ///
    /// * `Ok(XtbClient)` - connected to the servers. The credentials given as arguments are
    /// used to log in an user.
    /// * `Err(XtbClintBuilderError)` - cannot connect to the servers.
    ///
    /// Common reasons of an errors are:
    ///
    /// * Incorrect credentials.
    /// * Malformed API servers urls.
    /// * Missing required configuration value.
    pub async fn build(self, user_id: &str, password: &str) -> Result<XtbClient, XtbClientBuilderError> {
        let api_url = Self::make_url(self.api_url)?;
        let stream_api_url = Self::make_url(self.stream_api_url)?;

        // create connection and perform login
        let mut connection = BasicXtbConnection::new(api_url).await.map_err(|err| XtbClientBuilderError::CannotMakeConnection(err))?;
        let mut login_request = LoginRequest::default().with_user_id(user_id).with_password(password);

        if let Some(app_id) = self.app_id {
            login_request = login_request.with_app_id(app_id);
        }
        if let Some(app_name) = self.app_name {
            login_request = login_request.with_app_name(app_name);
        }

        let login_request_value = to_value(login_request).map_err(|err| XtbClientBuilderError::UnexpectedError(format!("{:?}", err)))?;

        let response = connection
            .send_command(COMMAND_LOGIN, Some(login_request_value)).await
            .map_err(|err| XtbClientBuilderError::UnexpectedError(format!("{:?}", err)))?.await
            .map_err(|err| XtbClientBuilderError::UnexpectedError(format!("{:?}", err)))?;

        let stream_session_id = match response {
            ProcessedMessage::ErrorResponse(msg) => return Err(XtbClientBuilderError::LoginFailed { user_id: user_id.to_string(), extra_info: format!("{:?}", msg) }),
            ProcessedMessage::Response(response) => response.stream_session_id.unwrap(),
        };

        let stream_connection = BasicXtbStreamConnection::new(stream_api_url, stream_session_id).await.map_err(|err| XtbClientBuilderError::CannotMakeStreamConnection(err))?;

        Ok(XtbClient::new(connection, stream_connection, self.ping_period.unwrap_or(DEFAULT_PING_INTERVAL_S)))
    }

    /// Convert string into an `Url` instance. This method is also used for validation of url presence.
    ///
    /// # Returns
    ///
    /// * `Ok(Url)` with correctly parsed url.
    /// * `Err(XtbClientBuilderError::RequiredFieldMissing)` if argument is `None`
    /// * `Err(XtbClientBuilderError::InvalidUrl)` if the url is malformed (cannot be parsed).
    fn make_url(source: Option<String>) -> Result<Url, XtbClientBuilderError> {
        let source_str = source.ok_or_else(|| XtbClientBuilderError::RequiredFieldMissing("api_url".to_owned()))?;
        Url::from_str(&source_str).map_err(|err| XtbClientBuilderError::InvalidUrl(source_str, err))
    }
}


#[derive(Debug, Error)]
pub enum XtbClientBuilderError {
    #[error("Required configuration field is missing: {0}")]
    RequiredFieldMissing(String),
    #[error("Url is invalid or malformed: {0} ({1})")]
    InvalidUrl(String, url::ParseError),
    #[error("Cannot connect to server")]
    CannotMakeConnection(BasicXtbConnectionError),
    #[error("Cannot connect to stream server")]
    CannotMakeStreamConnection(BasicXtbStreamConnectionError),
    #[error("Login failed for user: {user_id} ({extra_info:?})")]
    LoginFailed { user_id: String, extra_info: String },
    #[error("Something gets horribly wrong: {0}")]
    UnexpectedError(String),
}


/// Declaration of the Request/response API interface.
#[async_trait]
pub trait RequestResponseApi {
    /// Error returned from methods when command failed
    type Error;

    /// Returns array of all symbols available for the user.
    async fn get_all_symbols(&mut self, request: GetAllSymbolsRequest) -> Result<GetAllSymbolsResponse, Self::Error>;

    /// Returns calendar with market events.
    async fn get_calendar(&mut self, request: GetCalendarRequest) -> Result<GetCalendarResponse, Self::Error>;

    /// Please note that this function can be usually replaced by its streaming equivalent
    /// getCandles which is the preferred way of retrieving current candle data. Returns chart info,
    /// from start date to the current time. If the chosen period of CHART_LAST_INFO_RECORD is
    /// greater than 1 minute, the last candle returned by the API can change until the end of the
    /// period (the candle is being automatically updated every minute).
    ///
    /// Limitations: there are limitations in charts data availability. Detailed ranges for charts
    /// data, what can be accessed with specific period, are as follows:
    ///
    /// * PERIOD_M1 --- <0-1) month, i.e. one-month time
    /// * PERIOD_M30 --- <1-7) month, six months time
    /// * PERIOD_H4 --- <7-13) month, six months time
    /// * PERIOD_D1 --- 13 month, and earlier on
    ///
    /// Note, that specific PERIOD_ is the lowest (i.e. the most detailed) period, accessible
    /// in listed range. For instance, in months range <1-7) you can access periods: PERIOD_M30,
    /// PERIOD_H1, PERIOD_H4, PERIOD_D1, PERIOD_W1, PERIOD_MN1. Specific data ranges availability
    /// is guaranteed, however those ranges may be wider, e.g.: PERIOD_M1 may be accessible
    /// for 1.5 months back from now, where 1.0 months is guaranteed.
    ///
    /// Example scenario:
    ///
    /// * request charts of 5 minutes period, for 3 months time span, back from now;
    /// * response: you are guaranteed to get 1 month of 5 minutes charts; because, 5 minutes period
    /// charts are not accessible 2 months and 3 months back from now.
    async fn get_chart_last_request(&mut self, request: GetChartLastRequestRequest) -> Result<GetChartLastRequestResponse, Self::Error>;

    /// Please note that this function can be usually replaced by its streaming equivalent
    /// getCandles which is the preferred way of retrieving current candle data. Returns chart info
    /// with data between given start and end dates.
    ///
    /// Limitations: there are limitations in charts data availability. Detailed ranges for charts
    /// data, what can be accessed with specific period, are as follows:
    ///
    /// * PERIOD_M1 --- <0-1) month, i.e. one month time
    /// * PERIOD_M30 --- <1-7) month, six months time
    /// * PERIOD_H4 --- <7-13) month, six months time
    /// * PERIOD_D1 --- 13 month, and earlier on
    ///
    /// Note, that specific PERIOD_ is the lowest (i.e. the most detailed) period, accessible
    /// in listed range. For instance, in months range <1-7) you can access periods: PERIOD_M30,
    /// PERIOD_H1, PERIOD_H4, PERIOD_D1, PERIOD_W1, PERIOD_MN1. Specific data ranges availability
    /// is guaranteed, however those ranges may be wider, e.g.: PERIOD_M1 may be accessible
    /// for 1.5 months back from now, where 1.0 months is guaranteed.
    async fn get_chart_range_request(&mut self, request: GetChartRangeRequestRequest) -> Result<GetChartRangeRequestResponse, Self::Error>;

    /// Returns calculation of commission and rate of exchange. The value is calculated as expected
    /// value, and therefore might not be perfectly accurate.
    async fn get_commission_def(&mut self, request: GetCommissionDefRequest) -> Result<GetCommissionDefResponse, Self::Error>;

    /// Returns information about account currency, and account leverage.
    async fn get_current_user_data(&mut self, request: GetCurrentUserDataRequest) -> Result<GetCurrentUserDataResponse, Self::Error>;

    /// Returns IBs data from the given time range.
    async fn get_ibs_history(&mut self, request: GetIbsHistoryRequest) -> Result<GetIbsHistoryResponse, Self::Error>;

    /// Please note that this function can be usually replaced by its streaming equivalent
    /// getBalance which is the preferred way of retrieving account indicators. Returns various
    /// account indicators.
    async fn get_margin_level(&mut self, request: GetMarginLevelRequest) -> Result<GetMarginLevelResponse, Self::Error>;

    /// Returns expected margin for given instrument and volume. The value is calculated as expected
    /// margin value, and therefore might not be perfectly accurate.
    async fn get_margin_trade(&mut self, request: GetMarginTradeRequest) -> Result<GetMarginTradeResponse, Self::Error>;

    /// Please note that this function can be usually replaced by its streaming equivalent getNews
    /// which is the preferred way of retrieving news data. Returns news from trading server which
    /// were sent within specified period of time.
    async fn get_news(&mut self, request: GetNewsRequest) -> Result<GetNewsResponse, Self::Error>;

    /// Calculates estimated profit for given deal data Should be used for calculator-like apps
    /// only. Profit for opened transactions should be taken from server, due to higher precision of
    /// server calculation.
    async fn get_profit_calculation(&mut self, request: GetProfitCalculationRequest) -> Result<GetProfitCalculationResponse, Self::Error>;

    /// Returns current time on trading server.
    async fn get_server_time(&mut self, request: GetServerTimeRequest) -> Result<GetServerTimeResponse, Self::Error>;

    /// Returns a list of step rules for DMAs.
    async fn get_step_rules(&mut self, request: GetStepRulesRequest) -> Result<GetStepRulesResponse, Self::Error>;

    /// Returns information about symbol available for the user.
    async fn get_symbol(&mut self, request: GetSymbolRequest) -> Result<GetSymbolResponse, Self::Error>;

    /// Please note that this function can be usually replaced by its streaming equivalent
    /// getTickPrices which is the preferred way of retrieving ticks data. Returns array of current
    /// quotations for given symbols, only quotations that changed from given timestamp are
    /// returned. New timestamp obtained from output will be used as an argument of the next call
    /// of this command.
    async fn get_tick_prices(&mut self, request: GetTickPricesRequest) -> Result<GetTickPricesResponse, Self::Error>;

    /// Returns array of trades listed in orders argument.
    async fn get_trade_records(&mut self, request: GetTradeRecordsRequest) -> Result<GetTradeRecordsResponse, Self::Error>;

    /// Please note that this function can be usually replaced by its streaming equivalent getTrades
    /// which is the preferred way of retrieving trades data. Returns array of user's trades.
    async fn get_trades(&mut self, request: GetTradesRequest) -> Result<GetTradesResponse, Self::Error>;

    /// Please note that this function can be usually replaced by its streaming equivalent getTrades
    /// which is the preferred way of retrieving trades data. Returns array of user's trades which
    /// were closed within specified period of time.
    async fn get_trades_history(&mut self, request: GetTradesHistoryRequest) -> Result<GetTradesHistoryResponse, Self::Error>;

    /// Returns quotes and trading times.
    async fn get_trading_hours(&mut self, request: GetTradingHoursRequest) -> Result<GetTradingHoursResponse, Self::Error>;

    /// Returns the current API version.
    async fn get_version(&mut self, request: GetVersionRequest) -> Result<GetVersionResponse, Self::Error>;

    /// Starts trade transaction. tradeTransaction sends main transaction information to the server.
    ///
    /// # Note
    ///
    /// How to verify that the trade request was accepted?
    /// The status field set to 'true' does not imply that the transaction was accepted. It only
    /// means, that the server acquired your request and began to process it. To analyse the status
    /// of the transaction (for example to verify if it was accepted or rejected) use the
    /// tradeTransactionStatus command with the order number, that came back with the response of
    /// the tradeTransaction command. You can find the example here:
    /// https://developers.xstore.pro/api/tutorials/opening_and_closing_trades2
    async fn trade_transaction(&mut self, request: TradeTransactionRequest) -> Result<TradeTransactionResponse, Self::Error>;

    /// Description: Please note that this function can be usually replaced by its streaming
    /// equivalent getTradeStatus which is the preferred way of retrieving transaction status data.
    /// Returns current transaction status. At any time of transaction processing client might check
    /// the status of transaction on server side. In order to do that client must provide unique
    /// order taken from tradeTransaction invocation.
    async fn trade_transaction_status(&mut self, request: TradeTransactionStatusRequest) -> Result<TradeTransactionStatusResponse, Self::Error>;
}


/// Declaration of the stream API interface
#[async_trait]
pub trait StreamApi {
    /// Error returned from the client when something went wrong
    type Error;

    type Stream<T: Send + Sync + for<'de> Deserialize<'de>>;

    /// Each streaming command takes as an argument streamSessionId which is sent in response
    /// message for login command performed in main connection. streamSessionId token allows to
    /// identify user in streaming connection. In one streaming connection multiple commands with
    /// different streamSessionId can be invoked. It will cause sending streaming data for multiple
    /// login sessions in one streaming connection. streamSessionId is valid until logout command is
    /// performed on main connection or main connection is disconnected.
    async fn subscribe_balance(&mut self, arguments: StreamGetBalanceSubscribe) -> Result<Self::Stream<StreamGetBalanceData>, Self::Error>;

    /// Subscribes for and unsubscribes from API chart candles. The interval of every candle
    /// is 1 minute. A new candle arrives every minute.
    async fn subscribe_candles(&mut self, arguments: StreamGetCandlesSubscribe) -> Result<Self::Stream<StreamGetCandlesData>, Self::Error>;

    /// Subscribes for and unsubscribes from 'keep alive' messages. A new 'keep alive' message
    /// is sent by the API every 3 seconds.
    async fn subscribe_keep_alive(&mut self, arguments: StreamGetKeepAliveSubscribe) -> Result<Self::Stream<StreamGetKeepAliveData>, Self::Error>;

    /// Subscribes for and unsubscribes from news.
    async fn subscribe_news(&mut self, arguments: StreamGetNewsSubscribe) -> Result<Self::Stream<StreamGetNewsData>, Self::Error>;

    /// Subscribes for and unsubscribes from profits.
    async fn subscribe_profits(&mut self, arguments: StreamGetProfitSubscribe) -> Result<Self::Stream<StreamGetProfitData>, Self::Error>;

    /// Establishes subscription for quotations and allows to obtain the relevant information
    /// in real-time, as soon as it is available in the system. The getTickPrices command can
    /// be invoked many times for the same symbol, but only one subscription for a given symbol
    /// will be created. Please beware that when multiple records are available, the order in which
    /// they are received is not guaranteed.
    async fn subscribe_tick_prices(&mut self, arguments: StreamGetTickPricesSubscribe) -> Result<Self::Stream<StreamGetTickPricesData>, Self::Error>;

    /// Establishes subscription for user trade status data and allows to obtain the relevant
    /// information in real-time, as soon as it is available in the system. Please beware that when
    /// multiple records are available, the order in which they are received is not guaranteed.
    async fn subscribe_trades(&mut self, arguments: StreamGetTradesSubscribe) -> Result<Self::Stream<StreamGetTradesData>, Self::Error>;

    /// Allows to get status for sent trade requests in real-time, as soon as it is available
    /// in the system. Please beware that when multiple records are available, the order in which
    /// they are received is not guaranteed.
    async fn subscribe_trade_status(&mut self, arguments: StreamGetTradeStatusSubscribe) -> Result<Self::Stream<StreamGetTradeStatusData>, Self::Error>;
}


/// Implementor of the API traits.
///
/// This struct is designed to be an interface between user (application) and XTB API servers.
///
/// The `XtbClient` is responsible for sending and receiving pings and logout when instance is dropped.
pub struct XtbClient {
    /// Connection to the request/response server
    connection: Arc<Mutex<BasicXtbConnection>>,
    /// Connection to the stream server
    stream_manager: StreamManager,
    /// handle of the request/response server ping worker
    ping_join_handle: JoinHandle<()>,
    /// handle of the stream server ping worker
    stream_ping_join_handle: JoinHandle<()>,
}


impl XtbClient {
    /// Create builder for the `XtbClient`.
    ///
    /// When builder is configured, the `build()` method can be called with credentials passed to the
    /// method arguments. This call create new `XtbClient` instance.
    pub fn builder() -> XtbClientBuilder {
        XtbClientBuilder::default()
    }

    /// Create new instance of the `XtbClient`. it expects connected and logged connections to the
    /// request/response and the stream server.
    ///
    /// # Note
    ///
    /// The login is performed by the builder because the stream server implementation needs to know
    /// a stream session id which is provided by the `login` command.
    pub fn new(connection: BasicXtbConnection, stream_connection: BasicXtbStreamConnection, ping_period: u64) -> Self {
        let connection = Arc::new(Mutex::new(connection));

        let ping_join_handle = spawn_ping(connection.clone(), ping_period);

        let stream_manager = StreamManager::new(stream_connection);
        let stream_ping_join_handle = spawn_stream_ping(stream_manager.clone(), ping_period);

        Self {
            connection,
            stream_manager,
            ping_join_handle,
            stream_ping_join_handle,
        }
    }

    /// Send command to the server and wait for response.
    ///
    /// If command does not return any response, create default one with type of `RESP`.
    async fn send_and_wait_or_default<REQ, RESP>(&mut self, command: &str, request: REQ) -> Result<RESP, XtbClientError>
        where
            REQ: Serialize,
            RESP: for<'de> Deserialize<'de> + Default {
        self.send_and_wait(command, request).await.map(|val| val.unwrap_or_default())
    }

    /// Send the command and wait for a response.
    async fn send_and_wait<REQ, RESP>(&mut self, command: &str, request: REQ) -> Result<Option<RESP>, XtbClientError>
        where
            REQ: Serialize,
            RESP: for<'de> Deserialize<'de>
    {
        let promise = self.send(command, request).await?;
        let response = promise.await.map_err(|err| {
            error!("Unexpected error: {:?}", err);
            XtbClientError::UnexpectedError
        })?;
        match response {
            ProcessedMessage::Response(response) => {
                match response.return_data {
                    Some(data) => from_value(data).map_err(|err| XtbClientError::DeserializationFailed(err)).map(|v| Some(v)),
                    None => Ok(None)
                }
            }
            ProcessedMessage::ErrorResponse(err) => Err(XtbClientError::CommandFailed(err)),
        }
    }

    /// Send a command request to the server and return `Ok(ResponsePromise)` o
    async fn send<A>(&mut self, command: &str, request: A) -> Result<ResponsePromise, XtbClientError>
        where
            A: Serialize
    {
        let mut conn = self.connection.lock().await;
        let payload = Self::convert_data_to_value(request)?;
        conn.send_command(command, Some(payload)).await.map_err(|err| {
            match err {
                BasicXtbConnectionError::SerializationError(err) => XtbClientError::SerializationFailed(err),
                BasicXtbConnectionError::CannotSendRequest(err) => XtbClientError::CannotSendCommand(err),
                _ => XtbClientError::UnexpectedError,
            }
        })
    }

    /// Serialize payload data into value.
    ///
    /// # Returns
    ///
    /// * `Ok(Value)` when data was serialized successfully.
    /// * `Err(XtbClientError::SerializationFailed)` - data cannot be serialized.
    fn convert_data_to_value<T: Serialize>(data: T) -> Result<Value, XtbClientError> {
        to_value(data).map_err(|err| XtbClientError::SerializationFailed(err))
    }

    /// Send stream command to the stream API server.
    ///
    /// # Parameters
    ///
    /// * `subscribe_command` - command name of the subscribe command (e.g. `getCandles`)
    /// * `subscribe_arguments` - arguments for the subscribe command
    /// * `unsubscribe_command` - command name of the unsubscribe command (e.g. `stopCandles`)
    /// * `unsubscribe_arguments` - arguments for the unsubscribe command
    /// * `data_command` - command name in data messages (e.g. `candle`)
    ///
    /// # Returns
    ///
    /// * `Ok<DataStream<T>>` - data stream with filter set to messages related to sent command
    /// * `Err<XtbClientError>` - unable to send command
    async fn send_simple_stream_command<T, SA, UA>(
        &mut self,
        subscribe_command: &str,
        subscribe_arguments: SA,
        unsubscribe_command: &str,
        unsubscribe_arguments: UA,
        data_command: &str,
    ) -> Result<DataStream<T>, XtbClientError>
        where
            T: for<'de> Deserialize<'de> + Send + Sync,
            SA: Serialize,
            UA: Serialize,
    {
        let unsubscribe_arguments = Self::convert_data_to_value(unsubscribe_arguments)?;
        let filter = DataMessageFilter::Command(data_command.to_owned());
        let subscribe_arguments = Self::convert_data_to_value(subscribe_arguments)?;
        self.stream_manager.subscribe(subscribe_command, Some(subscribe_arguments), unsubscribe_command, Some(unsubscribe_arguments), data_command, filter).await
    }

    /// Send stream command to the stream API server and add filter by the `symbol` field to the
    /// data stream.
    ///
    /// # Parameters
    ///
    /// * `subscribe_command` - command name of the subscribe command (e.g. `getCandles`)
    /// * `subscribe_arguments` - arguments for the subscribe command
    /// * `unsubscribe_command` - command name of the unsubscribe command (e.g. `stopCandles`)
    /// * `unsubscribe_arguments` - arguments for the unsubscribe command
    /// * `data_command` - command name in data messages (e.g. `candle`)
    ///
    /// # Returns
    ///
    /// * `Ok<DataStream<T>>` - data stream with filter set to messages related to sent command
    /// * `Err<XtbClientError>` - unable to send command
    async fn send_symbol_scoped_stream_command<T, SA, UA>(
        &mut self,
        subscribe_command: &str,
        subscribe_arguments: SA,
        unsubscribe_command: &str,
        unsubscribe_arguments: UA,
        data_command: &str,
        symbol: &str,
    ) -> Result<DataStream<T>, XtbClientError>
        where
            T: for<'de> Deserialize<'de> + Send + Sync,
            SA: Serialize,
            UA: Serialize,
    {
        let unsubscribe_arguments = Self::convert_data_to_value(unsubscribe_arguments)?;
        let subscribe_arguments = Self::convert_data_to_value(subscribe_arguments)?;
        let subscription_key = format!("{}.{}", data_command, symbol);

        let filter = DataMessageFilter::All(vec![
            DataMessageFilter::Command(data_command.to_owned()),
            DataMessageFilter::FieldValue { name: "symbol".to_owned(), value: Value::String(symbol.to_owned()) },
        ]);
        self.stream_manager.subscribe(subscribe_command, Some(subscribe_arguments), unsubscribe_command, Some(unsubscribe_arguments), &subscription_key, filter).await
    }
}


impl Drop for XtbClient {
    fn drop(&mut self) {
        self.ping_join_handle.abort();
        self.stream_ping_join_handle.abort();
    }
}


#[async_trait]
impl RequestResponseApi for XtbClient {
    type Error = XtbClientError;

    async fn get_all_symbols(&mut self, request: GetAllSymbolsRequest) -> Result<GetAllSymbolsResponse, Self::Error> {
        self.send_and_wait_or_default(COMMAND_GET_ALL_SYMBOLS, request).await
    }

    async fn get_calendar(&mut self, request: GetCalendarRequest) -> Result<GetCalendarResponse, Self::Error> {
        self.send_and_wait_or_default(COMMAND_GET_CALENDAR, request).await
    }

    async fn get_chart_last_request(&mut self, request: GetChartLastRequestRequest) -> Result<GetChartLastRequestResponse, Self::Error> {
        self.send_and_wait_or_default(COMMAND_GET_CHART_LAST_REQUEST, request).await
    }

    async fn get_chart_range_request(&mut self, request: GetChartRangeRequestRequest) -> Result<GetChartRangeRequestResponse, Self::Error> {
        self.send_and_wait_or_default(COMMAND_GET_CHART_RANGE_REQUEST, request).await
    }

    async fn get_commission_def(&mut self, request: GetCommissionDefRequest) -> Result<GetCommissionDefResponse, Self::Error> {
        self.send_and_wait_or_default(COMMAND_GET_COMMISSION_DEF, request).await
    }

    async fn get_current_user_data(&mut self, request: GetCurrentUserDataRequest) -> Result<GetCurrentUserDataResponse, Self::Error> {
        self.send_and_wait_or_default(COMMAND_GET_CURRENT_USER_DATA, request).await
    }

    async fn get_ibs_history(&mut self, request: GetIbsHistoryRequest) -> Result<GetIbsHistoryResponse, Self::Error> {
        self.send_and_wait_or_default(COMMAND_GET_IBS_HISTORY, request).await
    }

    async fn get_margin_level(&mut self, request: GetMarginLevelRequest) -> Result<GetMarginLevelResponse, Self::Error> {
        self.send_and_wait_or_default(COMMAND_GET_MARGIN_LEVEL, request).await
    }

    async fn get_margin_trade(&mut self, request: GetMarginTradeRequest) -> Result<GetMarginTradeResponse, Self::Error> {
        self.send_and_wait_or_default(COMMAND_GET_MARGIN_TRADE, request).await
    }

    async fn get_news(&mut self, request: GetNewsRequest) -> Result<GetNewsResponse, Self::Error> {
        self.send_and_wait_or_default(COMMAND_GET_NEWS, request).await
    }

    async fn get_profit_calculation(&mut self, request: GetProfitCalculationRequest) -> Result<GetProfitCalculationResponse, Self::Error> {
        self.send_and_wait_or_default(COMMAND_GET_PROFIT_CALCULATION, request).await
    }

    async fn get_server_time(&mut self, request: GetServerTimeRequest) -> Result<GetServerTimeResponse, Self::Error> {
        self.send_and_wait_or_default(COMMAND_GET_SERVER_TIME, request).await
    }

    async fn get_step_rules(&mut self, request: GetStepRulesRequest) -> Result<GetStepRulesResponse, Self::Error> {
        self.send_and_wait_or_default(COMMAND_GET_STEP_RULES, request).await
    }

    async fn get_symbol(&mut self, request: GetSymbolRequest) -> Result<GetSymbolResponse, Self::Error> {
        self.send_and_wait_or_default(COMMAND_GET_SYMBOL, request).await
    }

    async fn get_tick_prices(&mut self, request: GetTickPricesRequest) -> Result<GetTickPricesResponse, Self::Error> {
        self.send_and_wait_or_default(COMMAND_GET_TICK_PRICES, request).await
    }

    async fn get_trade_records(&mut self, request: GetTradeRecordsRequest) -> Result<GetTradeRecordsResponse, Self::Error> {
        self.send_and_wait_or_default(COMMAND_GET_TRADE_RECORDS, request).await
    }

    async fn get_trades(&mut self, request: GetTradesRequest) -> Result<GetTradesResponse, Self::Error> {
        self.send_and_wait_or_default(COMMAND_GET_TRADES, request).await
    }

    async fn get_trades_history(&mut self, request: GetTradesHistoryRequest) -> Result<GetTradesHistoryResponse, Self::Error> {
        self.send_and_wait_or_default(COMMAND_GET_TRADES_HISTORY, request).await
    }

    async fn get_trading_hours(&mut self, request: GetTradingHoursRequest) -> Result<GetTradingHoursResponse, Self::Error> {
        self.send_and_wait_or_default(COMMAND_GET_TRADING_HOURS, request).await
    }

    async fn get_version(&mut self, request: GetVersionRequest) -> Result<GetVersionResponse, Self::Error> {
        self.send_and_wait_or_default(COMMAND_GET_VERSION, request).await
    }

    async fn trade_transaction(&mut self, request: TradeTransactionRequest) -> Result<TradeTransactionResponse, Self::Error> {
        self.send_and_wait_or_default(COMMAND_TRADE_TRANSACTION, request).await
    }

    async fn trade_transaction_status(&mut self, request: TradeTransactionStatusRequest) -> Result<TradeTransactionStatusResponse, Self::Error> {
        self.send_and_wait_or_default(COMMAND_TRADE_TRANSACTION_STATUS, request).await
    }
}


#[async_trait]
impl StreamApi for XtbClient {
    type Error = XtbClientError;

    type Stream<T: Send + Sync + for<'de> Deserialize<'de>> = DataStream<T>;

    async fn subscribe_balance(&mut self, arguments: StreamGetBalanceSubscribe) -> Result<Self::Stream<StreamGetBalanceData>, Self::Error> {
        let stop_arguments = Self::convert_data_to_value(StreamGetBalanceUnsubscribe::default())?;
        self.send_simple_stream_command(STREAM_BALANCE_SUBSCRIBE, arguments, STREAM_BALANCE_UNSUBSCRIBE, stop_arguments, STREAM_BALANCE).await
    }

    async fn subscribe_candles(&mut self, arguments: StreamGetCandlesSubscribe) -> Result<Self::Stream<StreamGetCandlesData>, Self::Error> {
        let stop_arguments = Self::convert_data_to_value(StreamGetCandlesUnsubscribe::default().with_symbol(&arguments.symbol))?;
        let symbol = arguments.symbol.clone();
        self.send_symbol_scoped_stream_command(STREAM_CANDLES_SUBSCRIBE, arguments, STREAM_CANDLES_UNSUBSCRIBE, stop_arguments, STREAM_CANDLES, &symbol).await
    }

    async fn subscribe_keep_alive(&mut self, arguments: StreamGetKeepAliveSubscribe) -> Result<Self::Stream<StreamGetKeepAliveData>, Self::Error> {
        let stop_arguments = Self::convert_data_to_value(StreamGetKeepAliveUnsubscribe::default())?;
        self.send_simple_stream_command(STREAM_KEEP_ALIVE_SUBSCRIBE, arguments, STREAM_KEEP_ALIVE_UNSUBSCRIBE, stop_arguments, STREAM_KEEP_ALIVE).await
    }

    async fn subscribe_news(&mut self, arguments: StreamGetNewsSubscribe) -> Result<Self::Stream<StreamGetNewsData>, Self::Error> {
        let stop_arguments = Self::convert_data_to_value(StreamGetNewsUnsubscribe::default())?;
        self.send_simple_stream_command(STREAM_NEWS_SUBSCRIBE, arguments, STREAM_NEWS_UNSUBSCRIBE, stop_arguments, STREAM_NEWS).await
    }

    async fn subscribe_profits(&mut self, arguments: StreamGetProfitSubscribe) -> Result<Self::Stream<StreamGetProfitData>, Self::Error> {
        let stop_arguments = Self::convert_data_to_value(StreamGetProfitUnsubscribe::default())?;
        self.send_simple_stream_command(STREAM_PROFITS_SUBSCRIBE, arguments, STREAM_PROFITS_UNSUBSCRIBE, stop_arguments, STREAM_PROFITS).await
    }

    async fn subscribe_tick_prices(&mut self, arguments: StreamGetTickPricesSubscribe) -> Result<Self::Stream<StreamGetTickPricesData>, Self::Error> {
        let stop_arguments = Self::convert_data_to_value(StreamGetTickPricesUnsubscribe::default().with_symbol(&arguments.symbol))?;
        let symbol = arguments.symbol.clone();
        self.send_symbol_scoped_stream_command(STREAM_TICK_PRICES_SUBSCRIBE, arguments, STREAM_TICK_PRICES_UNSUBSCRIBE, stop_arguments, STREAM_TICK_PRICES, &symbol).await
    }

    async fn subscribe_trades(&mut self, arguments: StreamGetTradesSubscribe) -> Result<Self::Stream<StreamGetTradesData>, Self::Error> {
        let stop_arguments = Self::convert_data_to_value(StreamGetTradesUnsubscribe::default())?;
        self.send_simple_stream_command(STREAM_TRADES_SUBSCRIBE, arguments, STREAM_TRADES_UNSUBSCRIBE, stop_arguments, STREAM_TRADES).await
    }

    async fn subscribe_trade_status(&mut self, arguments: StreamGetTradeStatusSubscribe) -> Result<Self::Stream<StreamGetTradeStatusData>, Self::Error> {
        let stop_arguments = Self::convert_data_to_value(StreamGetTradeStatusUnsubscribe::default())?;
        self.send_simple_stream_command(STREAM_TRADE_STATUS_SUBSCRIBE, arguments, STREAM_TRADE_STATUS_UNSUBSCRIBE, stop_arguments, STREAM_TRADE_STATUS).await
    }
}


#[derive(Debug, Error)]
pub enum XtbClientError {
    #[error("Cannot serialize arguments")]
    SerializationFailed(serde_json::Error),
    #[error("Cannot send command to server")]
    CannotSendCommand(tokio_tungstenite::tungstenite::Error),
    #[error("Cannot send stream command")]
    CannotSendStreamCommand(BasicXtbStreamConnectionError),
    #[error("Unexpected error.")]
    UnexpectedError,
    #[error("Cannot deserialize data")]
    DeserializationFailed(serde_json::Error),
    #[error("Command failed and an error response was returned")]
    CommandFailed(ErrorResponse),
}


/// Shared inner state of the `StreamManager`
#[derive(Debug)]
struct StreamManagerState {
    /// The stream connection
    connection: BasicXtbStreamConnection,
    /// subscription counter
    subscriptions: HashMap<String, usize>,
}


impl StreamManagerState {
    /// Create new instance of the struct
    pub fn new(connection: BasicXtbStreamConnection) -> Self {
        Self {
            connection,
            subscriptions: HashMap::new(),
        }
    }
}


/// Manage stream subscriptions across application. All instances cloned from same origin share
/// its internal state.
#[derive(Clone, Debug)]
struct StreamManager {
    /// The inner state shared between instances of the `StreamManager`
    state: Arc<Mutex<StreamManagerState>>,
}


impl StreamManager {
    /// Create new instance of the `StreamManager` struct.
    pub fn new(connection: BasicXtbStreamConnection) -> Self {
        let state = Arc::new(Mutex::new(StreamManagerState::new(connection)));
        Self {
            state
        }
    }

    /// Subscribe for a stream from the stream API server.
    ///
    /// # Parameters
    ///
    /// * `subscribe_command` - command name of the subscribe command (e.g. `getCandles`)
    /// * `subscribe_arguments` - arguments for the subscribe command
    /// * `unsubscribe_command` - command name of the unsubscribe command (e.g. `stopCandles`)
    /// * `unsubscribe_arguments` - arguments for the unsubscribe command
    /// * `subscription_key` - key used to track number of subscribers of the data stream
    /// * `filter` - the filter predicate for message routing.
    ///
    /// # Returns
    ///
    /// * `Ok<DataStream<T>>` - data stream with filter set to messages related to sent command
    /// * `Err<XtbClientError>` - unable to send command
    pub async fn subscribe<T: for<'de> Deserialize<'de> + Send + Sync>(
        &mut self,
        subscribe_command: &str,
        subscribe_arguments: Option<Value>,
        unsubscribe_command: &str,
        unsubscribe_arguments: Option<Value>,
        subscription_key: &str,
        filter: DataMessageFilter,
    ) -> Result<DataStream<T>, XtbClientError> {
        let mut state = self.state.lock().await;
        let stream = state.connection.make_message_stream(filter).await;
        state.connection.subscribe(subscribe_command, subscribe_arguments).await.map_err(|err| XtbClientError::CannotSendStreamCommand(err))?;
        *state.subscriptions.entry(subscription_key.to_owned()).or_default() += 1;
        Ok(DataStream::new(stream, self.clone(), subscription_key.to_owned(), unsubscribe_command.to_owned(), unsubscribe_arguments))
    }

    /// Unsubscribe from a stream.
    ///
    /// # Parameters
    ///
    /// * `subscription_key` - the subscription key where number of subscribers is tracked
    /// * `command` - an unsubscribe command to be sent to the server
    /// * `arguments` - the unsubscribe command arguments
    ///
    /// # Returns
    ///
    /// * `Ok(())` - success
    /// * `Err(XtbClientError::CannotSendStreamCommand)` - fail
    pub async fn unsubscribe(&mut self, subscription_key: &str, command: &str, arguments: Option<Value>) -> Result<(), XtbClientError> {
        let mut state = self.state.lock().await;
        let mut entry = state.subscriptions.entry(subscription_key.to_owned()).or_default();
        if *entry > 0 {
            *entry -= 1;
        }
        if *entry == 0 {
            state.connection.unsubscribe(command, arguments).await.map_err(|err| XtbClientError::CannotSendStreamCommand(err))?;
        }
        Ok(())
    }
}


/// Stream of messages delivered to a consumer.
///
/// The message data is deserialized and typed to data type related to a command.
pub struct DataStream<T>
    where
        T: for<'de> Deserialize<'de> + Send + Sync
{
    /// The message stream with raw messages
    message_stream: BasicMessageStream,
    /// The stream manager used to unsubscribe from a stream when struct is dropped
    stream_manager: StreamManager,
    /// Internal subscription key for subscriber tracking
    subscription_key: String,
    /// Unsubscribe command to be sent to the XTB server
    unsubscribe_command: String,
    /// Unsubscribe command arguments
    unsubscribe_arguments: Option<Value>,
    /// Data type returned to a consumer
    type_: PhantomData<T>,
}

impl<T> DataStream<T>
    where
        T: for<'de> Deserialize<'de> + Send + Sync
{
    /// Create new instance of the stream.
    fn new(message_stream: BasicMessageStream, stream_manager: StreamManager, subscription_key: String, unsubscribe_command: String, unsubscribe_arguments: Option<Value>) -> Self {
        Self {
            message_stream,
            stream_manager,
            subscription_key,
            unsubscribe_command,
            unsubscribe_arguments,
            type_: PhantomData::<T>,
        }
    }

    /// Wait and get next message from the stream.
    ///
    /// # Returns
    ///
    /// * `Ok(Some(T))` - next message in stream.
    /// * `Ok(None)` - there is no message left
    /// * `Err(DataStreamError)` - message was recived but cannot be processed. A next message can be ok.
    pub async fn next(&mut self) -> Result<Option<T>, DataStreamError> {
        let message = self.message_stream.next().await;
        match message {
            Some(msg) => Self::process_message(msg).map(|r| Some(r)),
            None => Ok(None),
        }
    }

    /// Deserialize serialized data representation to actual type `T`.
    fn process_message(msg: StreamDataMessage) -> Result<T, DataStreamError> {
        from_value(msg.data).map_err(|err| DataStreamError::CannotDeserializeValue(err))
    }
}

impl<T> Drop for DataStream<T>
    where
        T: for<'de> Deserialize<'de> + Send + Sync
{
    fn drop(&mut self) {
        let mut manager = self.stream_manager.clone();
        let unsubscribe_command = self.unsubscribe_command.clone();
        let unsubscribe_arguments = self.unsubscribe_arguments.take();
        let subscription_key = self.subscription_key.clone();
        spawn(async move {
            let result = manager.unsubscribe(&subscription_key, &unsubscribe_command, unsubscribe_arguments.clone()).await;
            match result {
                Err(err) => error!("Cannot unsubscribe command '{unsubscribe_command}' ({unsubscribe_arguments:?}). The subscription key was: '{subscription_key}'. The error was: {err:?}"),
                _ => (),
            };
        });
    }
}

#[derive(Debug, Error)]
pub enum DataStreamError {
    #[error("Cannot deserialize value: {0}")]
    CannotDeserializeValue(serde_json::Error)
}


/// Spawn tokio green thread and to send ping periodically to sync connection
///
/// # Arguments
///
/// * conn - the stream connection
/// * ping_secs - number of seconds between each ping
///
/// # Panics
///
/// The ping message cannot be serialized. The serialization is done before the green thread is run
///
/// # Returns
///
/// `JoinHandle` of the green thread
fn spawn_ping(conn: Arc<Mutex<BasicXtbConnection>>, ping_secs: u64) -> JoinHandle<()> {
    let ping_value = to_value(PingRequest::default()).expect("Cannot serialize ping message");
    spawn(async move {
        let mut idx = 1u64;
        loop {
            let response_promise = {
                let mut conn = conn.lock().await;
                debug!("Sending ping #{} to connection", idx);
                match conn.send_command(COMMAND_PING, Some(ping_value.clone())).await {
                    Ok(resp) => Some(resp),
                    Err(err) => {
                        error!("Cannot send ping #{}: {:?}", idx, err);
                        None
                    }
                }
            };
            if let Some(response_promise) = response_promise {
                match response_promise.await {
                    Ok(_) => (),
                    Err(err) => error!("Cannot await the ping response #{}: {:?}", idx, err)
                }
            }
            idx += 1;
            sleep(Duration::from_secs(ping_secs)).await;
        }
    })
}


/// Spawn tokio green thread and to send ping periodically to stream connection
///
/// # Arguments
///
/// * conn - the stream connection
/// * ping_secs - number of seconds between each ping
///
/// # Panics
///
/// The ping message cannot be serialized. The serialization is done before the green thread is run
///
/// # Returns
///
/// `JoinHandle` of the green thread
fn spawn_stream_ping(stream_manager: StreamManager, ping_secs: u64) -> JoinHandle<()> {
    let ping_value = to_value(StreamPingSubscribe::default()).expect("Cannot serialize the stream ping message");
    spawn(async move {
        let mut idx = 1u64;
        loop {
            {
                debug!("Sending ping #{} to stream connection", idx);
                let mut inner_state = stream_manager.state.lock().await;
                match inner_state.connection.subscribe(STREAM_PING, Some(ping_value.clone())).await {
                    Ok(_) => (),
                    Err(err) => error!("Cannot send ping #{}: {:?}", idx, err)
                }
            }
            idx += 1;
            sleep(Duration::from_secs(ping_secs)).await;
        }
    })
}