1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93

pub mod trades;
pub mod new_account_trades;
pub mod updated_account_orders;
pub mod updated_account_balances;
pub mod updated_orderbook;
pub mod updated_ticker;
use super::graphql::ResponseOrError;
use super::{NashProtocolSubscription, State};
use crate::errors::Result;
use async_trait::async_trait;
use tokio::sync::RwLock;
use std::sync::Arc;

/// Wrapper for all subscription request types supported on Nash. Required only for current version
/// of openlimits subscription logic
#[derive(Clone, Debug)]
pub enum SubscriptionRequest {
    Trades(trades::SubscribeTrades),
    Ticker(updated_ticker::SubscribeTicker),
    Orderbook(updated_orderbook::SubscribeOrderbook),
    AccountTrades(new_account_trades::SubscribeAccountTrades),
    AccountOrders(updated_account_orders::SubscribeAccountOrders),
    AccountBalances(updated_account_balances::SubscribeAccountBalances),
}

/// Wrapper for incoming subscription data supported on Nash. Required only for current version
/// of openlimits subscription logic
#[derive(Clone, Debug)]
pub enum SubscriptionResponse {
    Orderbook(updated_orderbook::SubscribeOrderbookResponse),
    Ticker(updated_ticker::SubscribeTickerResponse),
    Trades(trades::TradesResponse),
    AccountTrades(new_account_trades::AccountTradesResponse),
    AccountOrders(updated_account_orders::AccountOrdersResponse),
    AccountBalances(updated_account_balances::AccountBalancesResponse),
}

#[async_trait]
impl NashProtocolSubscription for SubscriptionRequest {
    type SubscriptionResponse = SubscriptionResponse;
    async fn graphql(&self, state: Arc<RwLock<State>>) -> Result<serde_json::Value> {
        match self {
            Self::Trades(trades_req) => trades_req.graphql(state).await,
            Self::Ticker(ticker_req) => ticker_req.graphql(state).await,
            Self::Orderbook(orderbook_req) => orderbook_req.graphql(state).await,
            Self::AccountTrades(account_trades_req) => account_trades_req.graphql(state).await,
            Self::AccountOrders(account_orders_req) => account_orders_req.graphql(state).await,
            Self::AccountBalances(account_balance_req) => account_balance_req.graphql(state).await
        }
    }
    async fn subscription_response_from_json(
        &self,
        response: serde_json::Value,
        state: Arc<RwLock<State>>,
    ) -> Result<ResponseOrError<Self::SubscriptionResponse>> {
        match self {
            Self::Trades(trades_req) => Ok(trades_req
                .subscription_response_from_json(response, state)
                .await?
                .map(Box::new(|res| SubscriptionResponse::Trades(res)))),
            Self::AccountTrades(account_trades_req) => Ok(account_trades_req
                .subscription_response_from_json(response, state)
                .await?
                .map(Box::new(|res| SubscriptionResponse::AccountTrades(res)))),
            Self::AccountOrders(account_orders_req) => Ok(account_orders_req
                .subscription_response_from_json(response, state)
                .await?
                .map(Box::new(|res| SubscriptionResponse::AccountOrders(res)))),
            Self::AccountBalances(account_balances_req) => Ok(account_balances_req
                .subscription_response_from_json(response, state)
                .await?
                .map(Box::new(|res| SubscriptionResponse::AccountBalances(res)))),
            Self::Ticker(ticker_req) => Ok(ticker_req
                .subscription_response_from_json(response, state)
                .await?
                .map(Box::new(|res| SubscriptionResponse::Ticker(res)))),
            Self::Orderbook(orderbook_req) => Ok(orderbook_req
                .subscription_response_from_json(response, state)
                .await?
                .map(Box::new(|res| SubscriptionResponse::Orderbook(res)))),
        }
    }
    async fn wrap_response_as_any_subscription(
        &self,
        response: serde_json::Value,
        state: Arc<RwLock<State>>,
    ) -> Result<ResponseOrError<SubscriptionResponse>> {
        let response = self
            .subscription_response_from_json(response, state)
            .await?;
        Ok(response)
    }
}