rust-okx 0.5.0

Async Rust client for the OKX v5 REST API
Documentation 
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
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136

//! WebSocket connection traits and the default tungstenite connector.

use std::future::Future;

use bytes::Bytes;

use super::error::WsError;

/// A WebSocket frame handled by the OKX WebSocket client.
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum WsFrame {
    /// A text frame.
    Text(String),
    /// A ping frame.
    Ping(Bytes),
    /// A pong frame.
    Pong(Bytes),
    /// A close frame.
    Close,
}

/// A full-duplex WebSocket connection.
///
/// This trait is intentionally minimal so tests can use a fake connection
/// without depending on a concrete WebSocket implementation.
pub trait WsConn: Send {
    /// Send a text frame.
    fn send_text(&mut self, text: String) -> impl Future<Output = Result<(), WsError>> + Send;

    /// Send a pong control frame with the supplied ping payload.
    fn send_pong(&mut self, payload: Bytes) -> impl Future<Output = Result<(), WsError>> + Send;

    /// Receive the next frame.
    fn recv(&mut self) -> impl Future<Output = Result<Option<WsFrame>, WsError>> + Send;

    /// Close the connection.
    fn close(&mut self) -> impl Future<Output = Result<(), WsError>> + Send;
}

/// Creates WebSocket connections for an OKX WebSocket endpoint URL.
pub trait WsConnector: Send + Sync {
    /// The connection type returned by this connector.
    type Conn: WsConn;

    /// Connect to a WebSocket URL.
    fn connect(&self, url: &str) -> impl Future<Output = Result<Self::Conn, WsError>> + Send;
}

#[cfg(feature = "websocket")]
mod tungstenite_impl {
    use futures_util::{SinkExt, StreamExt};
    use tokio::net::TcpStream;
    use tokio_tungstenite::tungstenite::Message;
    use tokio_tungstenite::{MaybeTlsStream, WebSocketStream, connect_async};

    use super::*;
    use crate::TransportError;

    /// Default WebSocket connector backed by `tokio-tungstenite`.
    #[derive(Debug, Clone, Copy, Default)]
    pub struct TungsteniteConnector;

    /// WebSocket connection backed by `tokio-tungstenite`.
    pub struct TungsteniteConn {
        inner: WebSocketStream<MaybeTlsStream<TcpStream>>,
    }

    impl WsConnector for TungsteniteConnector {
        type Conn = TungsteniteConn;

        fn connect(&self, url: &str) -> impl Future<Output = Result<Self::Conn, WsError>> + Send {
            let url = url.to_owned();
            async move {
                let (inner, _) = connect_async(url)
                    .await
                    .map_err(|e| WsError::from(TransportError::new(e)))?;
                Ok(TungsteniteConn { inner })
            }
        }
    }

    #[allow(clippy::manual_async_fn)]
    impl WsConn for TungsteniteConn {
        fn send_text(&mut self, text: String) -> impl Future<Output = Result<(), WsError>> + Send {
            async move {
                self.inner
                    .send(Message::Text(text))
                    .await
                    .map_err(|e| WsError::from(TransportError::new(e)))
            }
        }

        fn send_pong(
            &mut self,
            payload: Bytes,
        ) -> impl Future<Output = Result<(), WsError>> + Send {
            async move {
                self.inner
                    .send(Message::Pong(payload.to_vec()))
                    .await
                    .map_err(|e| WsError::from(TransportError::new(e)))
            }
        }

        fn recv(&mut self) -> impl Future<Output = Result<Option<WsFrame>, WsError>> + Send {
            async move {
                loop {
                    let Some(message) = self.inner.next().await else {
                        return Ok(None);
                    };
                    let message = message.map_err(|e| WsError::from(TransportError::new(e)))?;
                    match message {
                        Message::Text(text) => return Ok(Some(WsFrame::Text(text))),
                        Message::Ping(bytes) => return Ok(Some(WsFrame::Ping(Bytes::from(bytes)))),
                        Message::Pong(bytes) => return Ok(Some(WsFrame::Pong(Bytes::from(bytes)))),
                        Message::Close(_) => return Ok(Some(WsFrame::Close)),
                        Message::Binary(_) | Message::Frame(_) => continue,
                    }
                }
            }
        }

        fn close(&mut self) -> impl Future<Output = Result<(), WsError>> + Send {
            async move {
                self.inner
                    .close(None)
                    .await
                    .map_err(|e| WsError::from(TransportError::new(e)))
            }
        }
    }
}

#[cfg(feature = "websocket")]
pub use tungstenite_impl::{TungsteniteConn, TungsteniteConnector};