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

use crate::error::OkxError;
use crate::websocket::types::{request::Request, response::Response};
use exc_core::ExchangeError;
use futures::{future::BoxFuture, FutureExt, TryFutureExt};
use tower::{buffer::Buffer, util::BoxService, Service, ServiceExt};

/// Okx websocket channel.
#[derive(Clone)]
pub struct Channel {
    pub(crate) svc: Buffer<BoxService<Request, Response, OkxError>, Request>,
}

impl Channel {
    /// Send request.
    pub async fn request(
        &mut self,
        request: Request,
    ) -> Result<<Self as Service<Request>>::Future, ExchangeError> {
        ServiceExt::<Request>::ready(self).await?;
        let fut = self.call(request);
        Ok(fut)
    }
}

impl tower::Service<Request> for Channel {
    type Response = Response;
    type Error = ExchangeError;
    type Future = BoxFuture<'static, Result<Self::Response, Self::Error>>;

    fn poll_ready(
        &mut self,
        cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<Result<(), Self::Error>> {
        self.svc
            .poll_ready(cx)
            .map_err(OkxError::Buffer)
            .map_err(|err| ExchangeError::Other(err.into()))
    }

    fn call(&mut self, req: Request) -> Self::Future {
        self.svc
            .call(req)
            .map_err(OkxError::Buffer)
            .map_err(|err| ExchangeError::Other(err.into()))
            .boxed()
    }
}