brokapi_alor_protocol/ws/cws.rs
1//! The trading channel `/cws`: order commands over one WebSocket connection.
2//!
3//! The same commands exist over HTTP, and the broker says the WebSocket gives only a small gain in time
4//! (up to 5 ms), so the connectors use HTTP. This client is here for those who want the channel anyway. It is
5//! plain on purpose: one connection, one command at a time, no reconnect. A command that gets no answer in time
6//! is an unknown outcome, as at every broker: find out before you send it again.
7//!
8//! The connection is authorized once with `authorize` (the broker wants it every 15-20 minutes), and then every
9//! command is a message with an `opcode` like `create:limit`, a `guid` of ours and the fields of the order. The
10//! answer carries the `guid` back as `requestGuid`.
11
12use std::time::Duration;
13
14use futures_util::{SinkExt, StreamExt};
15use serde_json::{Value, json};
16use tokio::net::TcpStream;
17use tokio::sync::Mutex;
18use tokio::time::{Instant, timeout};
19use tokio_tungstenite::tungstenite::Message;
20use tokio_tungstenite::{MaybeTlsStream, WebSocketStream, connect_async};
21
22use crate::error::AlorApiError;
23use crate::http::client::AlorHttpClient;
24use crate::ws::models::CWS_PATH;
25
26type Stream = WebSocketStream<MaybeTlsStream<TcpStream>>;
27
28/// The opcodes of the order commands of the trading channel.
29pub mod opcodes {
30 /// Create a market order.
31 pub const CREATE_MARKET: &str = "create:market";
32 /// Create a limit order.
33 pub const CREATE_LIMIT: &str = "create:limit";
34 /// Create a stop order (a market order on trigger).
35 pub const CREATE_STOP: &str = "create:stop";
36 /// Create a stop-limit order.
37 pub const CREATE_STOP_LIMIT: &str = "create:stopLimit";
38 /// Change a market order.
39 pub const UPDATE_MARKET: &str = "update:market";
40 /// Change a limit order.
41 pub const UPDATE_LIMIT: &str = "update:limit";
42 /// Change a stop order.
43 pub const UPDATE_STOP: &str = "update:stop";
44 /// Change a stop-limit order.
45 pub const UPDATE_STOP_LIMIT: &str = "update:stopLimit";
46 /// Cancel a market order.
47 pub const DELETE_MARKET: &str = "delete:market";
48 /// Cancel a limit order.
49 pub const DELETE_LIMIT: &str = "delete:limit";
50 /// Cancel a stop order.
51 pub const DELETE_STOP: &str = "delete:stop";
52 /// Cancel a stop-limit order.
53 pub const DELETE_STOP_LIMIT: &str = "delete:stopLimit";
54}
55
56/// The answer to a command of the trading channel.
57#[derive(Debug, Clone, PartialEq)]
58pub struct CwsReply {
59 /// The `guid` of the command (`requestGuid` of the answer).
60 pub guid: String,
61 /// The HTTP-like code: 200 done, 400 refused, 401 not authorized.
62 pub code: u16,
63 /// The text.
64 pub message: String,
65 /// The number of the order, for an answer to a create or change.
66 pub order_number: Option<String>,
67 /// The whole answer.
68 pub raw: Value,
69}
70
71/// An authorized connection to the trading channel.
72///
73/// # Examples
74///
75/// ```
76/// # #[tokio::main(flavor = "current_thread")]
77/// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
78/// # let server = brokapi_mocks::alor::MockServer::start(Default::default()).await;
79/// # let cfg = brokapi_alor_protocol::config::AlorConfig::for_mock("mock-refresh-token", &server.http_base);
80/// use std::time::Duration;
81/// use serde_json::json;
82/// use brokapi_alor_protocol::http::client::AlorHttpClient;
83/// use brokapi_alor_protocol::ws::cws::{CwsSession, opcodes};
84///
85/// let http = AlorHttpClient::new(cfg)?;
86/// let cws = CwsSession::connect(http, Duration::from_secs(5)).await?;
87/// let reply = cws
88/// .command(opcodes::CREATE_LIMIT, json!({
89/// "side": "buy", "quantity": 1, "price": 100.0,
90/// "instrument": {"symbol": "SBER", "exchange": "MOEX", "instrumentGroup": "TQBR"},
91/// "user": {"portfolio": "D00001"},
92/// }))
93/// .await?;
94/// assert_eq!(reply.code, 200);
95/// assert!(reply.order_number.is_some());
96/// # Ok(())
97/// # }
98/// ```
99pub struct CwsSession {
100 http: AlorHttpClient,
101 stream: Mutex<Stream>,
102 answer_timeout: Duration,
103 authorized_at: Mutex<Instant>,
104}
105
106/// How long an authorization of the connection is trusted before the session authorizes again.
107const REAUTHORIZE_AFTER: Duration = Duration::from_secs(15 * 60);
108
109impl CwsSession {
110 /// Opens the connection and authorizes it. `answer_timeout` is how long a command waits for its answer.
111 ///
112 /// # Errors
113 ///
114 /// [`AlorApiError::Ws`] when the connection or the authorization fails; the errors of
115 /// [`AlorHttpClient::get_access_token`] when there is no token.
116 pub async fn connect(
117 http: AlorHttpClient,
118 answer_timeout: Duration,
119 ) -> Result<Self, AlorApiError> {
120 let url = format!("{}{}", http.ws_base_url(), CWS_PATH);
121 let (stream, _) = connect_async(url.as_str())
122 .await
123 .map_err(|e| AlorApiError::Ws {
124 code: None,
125 message: format!("connect to {url}: {e}"),
126 })?;
127 let session = Self {
128 http,
129 stream: Mutex::new(stream),
130 answer_timeout,
131 authorized_at: Mutex::new(Instant::now()),
132 };
133 session.authorize().await?;
134 Ok(session)
135 }
136
137 /// Authorizes the connection with a fresh access token.
138 ///
139 /// # Errors
140 ///
141 /// [`AlorApiError::Ws`] when the broker refuses (code 401) or does not answer in time.
142 pub async fn authorize(&self) -> Result<(), AlorApiError> {
143 let token = self.http.get_access_token().await?;
144 let reply = self
145 .exchange(json!({"opcode": "authorize", "token": token}), false)
146 .await?;
147 if reply.code != 200 {
148 return Err(AlorApiError::Ws {
149 code: Some(reply.code),
150 message: reply.message,
151 });
152 }
153 *self.authorized_at.lock().await = Instant::now();
154 Ok(())
155 }
156
157 /// Sends a command and waits for its answer. `fields` is the object of the order (`side`, `quantity`,
158 /// `price`, `instrument`, `user`, ...); the session adds `opcode` and a new `guid`.
159 ///
160 /// A refusal of the broker (code 400) is returned as a reply, not as an error: the caller reads
161 /// [`CwsReply::code`] and [`CwsReply::message`].
162 ///
163 /// # Errors
164 ///
165 /// [`AlorApiError::Ws`] when the connection breaks or no answer comes in time. Both are an unknown
166 /// outcome for the command.
167 pub async fn command(&self, opcode: &str, fields: Value) -> Result<CwsReply, AlorApiError> {
168 if self.authorized_at.lock().await.elapsed() >= REAUTHORIZE_AFTER {
169 self.authorize().await?;
170 }
171 let mut body = fields;
172 let obj = body
173 .as_object_mut()
174 .ok_or_else(|| AlorApiError::Parse("the fields must be a JSON object".into()))?;
175 obj.insert("opcode".into(), json!(opcode));
176 self.exchange(body, true).await
177 }
178
179 async fn exchange(&self, mut body: Value, _is_command: bool) -> Result<CwsReply, AlorApiError> {
180 let guid = uuid_like();
181 body["guid"] = json!(guid);
182 let mut stream = self.stream.lock().await;
183 stream
184 .send(Message::Text(body.to_string().into()))
185 .await
186 .map_err(|e| AlorApiError::Ws {
187 code: None,
188 message: format!("write: {e}"),
189 })?;
190 let wait = async {
191 while let Some(frame) = stream.next().await {
192 match frame {
193 Ok(Message::Text(text)) => {
194 let Ok(v) = serde_json::from_str::<Value>(&text) else {
195 continue;
196 };
197 if v.get("requestGuid").and_then(Value::as_str) != Some(guid.as_str()) {
198 continue;
199 }
200 return Ok(CwsReply {
201 guid: guid.clone(),
202 code: v.get("httpCode").and_then(Value::as_u64).unwrap_or(0) as u16,
203 message: v
204 .get("message")
205 .and_then(Value::as_str)
206 .unwrap_or_default()
207 .to_string(),
208 order_number: v
209 .get("orderNumber")
210 .and_then(Value::as_str)
211 .map(str::to_string),
212 raw: v,
213 });
214 }
215 Ok(Message::Close(_)) | Err(_) => break,
216 Ok(_) => {}
217 }
218 }
219 Err(AlorApiError::Ws {
220 code: None,
221 message: "the connection closed before the answer".into(),
222 })
223 };
224 timeout(self.answer_timeout, wait)
225 .await
226 .map_err(|_| AlorApiError::Ws {
227 code: None,
228 message: "no answer in time".into(),
229 })?
230 }
231}
232
233/// A unique id for a request: the time in nanoseconds and a counter, as text.
234fn uuid_like() -> String {
235 use std::sync::atomic::{AtomicU64, Ordering};
236 static N: AtomicU64 = AtomicU64::new(0);
237 let t = std::time::SystemTime::now()
238 .duration_since(std::time::UNIX_EPOCH)
239 .map(|d| d.as_nanos())
240 .unwrap_or(0);
241 format!("cws-{t:x}-{}", N.fetch_add(1, Ordering::Relaxed))
242}