wickra_exchange_core/
transport.rs

1//! The transport abstraction: the seam between exchange logic and real sockets.
2//!
3//! Every exchange implementation issues HTTP requests and reads WebSocket frames
4//! through these traits rather than touching a socket directly. Production wires
5//! a thin real-socket adapter; tests — and users who want to exercise their own
6//! trading code offline — wire [`MockHttpTransport`] / [`MockWsTransport`], which
7//! replay canned responses and frames. Because all exchange logic runs through
8//! this seam, signing, parsing, filter rounding, the WebSocket state machine,
9//! reconnect and error handling are all exercisable without a live network.
10
11use crate::error::{Error, Result};
12use std::collections::VecDeque;
13use std::sync::{Arc, Mutex};
14
15/// An HTTP method.
16#[derive(Debug, Clone, Copy, PartialEq, Eq)]
17pub enum HttpMethod {
18    /// `GET`.
19    Get,
20    /// `POST`.
21    Post,
22    /// `PUT`.
23    Put,
24    /// `PATCH`.
25    Patch,
26    /// `DELETE`.
27    Delete,
28}
29
30impl HttpMethod {
31    /// The uppercase method name.
32    #[must_use]
33    pub fn as_str(self) -> &'static str {
34        match self {
35            HttpMethod::Get => "GET",
36            HttpMethod::Post => "POST",
37            HttpMethod::Put => "PUT",
38            HttpMethod::Patch => "PATCH",
39            HttpMethod::Delete => "DELETE",
40        }
41    }
42}
43
44/// An outbound HTTP request.
45#[derive(Debug, Clone, PartialEq, Eq)]
46pub struct HttpRequest {
47    /// The HTTP method.
48    pub method: HttpMethod,
49    /// The fully-qualified URL, including any query string.
50    pub url: String,
51    /// Request headers.
52    pub headers: Vec<(String, String)>,
53    /// Optional request body.
54    pub body: Option<String>,
55}
56
57impl HttpRequest {
58    /// A `GET` request to `url`.
59    pub fn get(url: impl Into<String>) -> Self {
60        Self {
61            method: HttpMethod::Get,
62            url: url.into(),
63            headers: Vec::new(),
64            body: None,
65        }
66    }
67
68    /// A `POST` request to `url`.
69    pub fn post(url: impl Into<String>) -> Self {
70        Self {
71            method: HttpMethod::Post,
72            url: url.into(),
73            headers: Vec::new(),
74            body: None,
75        }
76    }
77
78    /// A request with an arbitrary method.
79    pub fn new(method: HttpMethod, url: impl Into<String>) -> Self {
80        Self {
81            method,
82            url: url.into(),
83            headers: Vec::new(),
84            body: None,
85        }
86    }
87
88    /// Add a header.
89    #[must_use]
90    pub fn with_header(mut self, name: impl Into<String>, value: impl Into<String>) -> Self {
91        self.headers.push((name.into(), value.into()));
92        self
93    }
94
95    /// Set the body.
96    #[must_use]
97    pub fn with_body(mut self, body: impl Into<String>) -> Self {
98        self.body = Some(body.into());
99        self
100    }
101}
102
103/// An HTTP response.
104#[derive(Debug, Clone, PartialEq, Eq)]
105pub struct HttpResponse {
106    /// The status code.
107    pub status: u16,
108    /// The response headers, in the order the venue sent them.
109    ///
110    /// Carried because the body is not the whole answer: a venue that
111    /// rate-limits says *how long to wait* in `Retry-After`, and a response
112    /// without headers throws that away before anything can read it — which is
113    /// why [`Error::RateLimited`] used to be raised with `retry_after: None`
114    /// at every one of its construction sites.
115    pub headers: Vec<(String, String)>,
116    /// The response body.
117    pub body: String,
118}
119
120impl HttpResponse {
121    /// Build a response with no headers.
122    pub fn new(status: u16, body: impl Into<String>) -> Self {
123        Self {
124            status,
125            headers: Vec::new(),
126            body: body.into(),
127        }
128    }
129
130    /// Add a header.
131    #[must_use]
132    pub fn with_header(mut self, name: impl Into<String>, value: impl Into<String>) -> Self {
133        self.headers.push((name.into(), value.into()));
134        self
135    }
136
137    /// The first value of `name`, matched case-insensitively.
138    ///
139    /// Header names are case-insensitive per RFC 9110, and venues are not
140    /// consistent about it: the same field arrives as `Retry-After` from one
141    /// and `retry-after` from another behind HTTP/2, which lower-cases them.
142    #[must_use]
143    pub fn header(&self, name: &str) -> Option<&str> {
144        self.headers
145            .iter()
146            .find(|(key, _)| key.eq_ignore_ascii_case(name))
147            .map(|(_, value)| value.as_str())
148    }
149
150    /// The venue's advised wait from `Retry-After`, if it sent one.
151    ///
152    /// Only the delta-seconds form is read. The HTTP-date form is legal but no
153    /// venue in this repository's error taxonomy uses it, and parsing dates
154    /// would mean a date library and a clock in a function that must stay
155    /// pure — an unread `None` is the better failure here than a wrong
156    /// duration. Fractional seconds are accepted because some venues send
157    /// them, though the RFC specifies an integer.
158    #[must_use]
159    pub fn retry_after(&self) -> Option<std::time::Duration> {
160        let raw = self.header("Retry-After")?.trim();
161        let seconds: f64 = raw.parse().ok()?;
162        if seconds.is_finite() && seconds >= 0.0 {
163            Some(std::time::Duration::from_secs_f64(seconds))
164        } else {
165            None
166        }
167    }
168
169    /// Whether the status is in the 2xx range.
170    #[must_use]
171    pub fn is_success(&self) -> bool {
172        (200..300).contains(&self.status)
173    }
174}
175
176/// A synchronous HTTP transport. The public surface stays blocking; a real
177/// adapter drives an async client internally, a mock returns canned responses.
178pub trait HttpTransport: Send + Sync {
179    /// Execute a request and return the response.
180    ///
181    /// # Errors
182    ///
183    /// Returns a transport-level [`Error`] (network, timeout) on failure; HTTP
184    /// error *status codes* are returned as `Ok` responses for the caller to
185    /// interpret against the venue's error taxonomy.
186    fn execute(&self, request: &HttpRequest) -> Result<HttpResponse>;
187}
188
189/// A WebSocket transport: opens connections to stream URLs.
190pub trait WsTransport: Send + Sync {
191    /// Open a connection to `url`.
192    ///
193    /// # Errors
194    ///
195    /// Returns a transport-level [`Error`] if the connection cannot be opened.
196    fn connect(&self, url: &str) -> Result<Box<dyn WsConnection>>;
197}
198
199/// A single WebSocket connection. `recv` returns `Ok(None)` when the peer closed
200/// the connection cleanly.
201pub trait WsConnection: Send {
202    /// Send a text frame.
203    ///
204    /// # Errors
205    ///
206    /// Returns [`Error::NotConnected`] if the connection is closed, or a
207    /// transport error on failure.
208    fn send(&mut self, text: &str) -> Result<()>;
209
210    /// Receive the next text frame, or `Ok(None)` if nothing is pending or the
211    /// peer closed the connection. Use [`is_connected`](Self::is_connected) to
212    /// distinguish a live-but-idle stream from a closed one.
213    ///
214    /// # Errors
215    ///
216    /// Returns a transport-level [`Error`] on a read failure.
217    fn recv(&mut self) -> Result<Option<String>>;
218
219    /// Whether the connection is still open. Returns `false` once the peer has
220    /// closed it cleanly, so a caller can reconnect and resubscribe.
221    fn is_connected(&self) -> bool;
222
223    /// Close the connection.
224    ///
225    /// # Errors
226    ///
227    /// Returns a transport-level [`Error`] on failure.
228    fn close(&mut self) -> Result<()>;
229}
230
231// ---------------------------------------------------------------------------
232// Mock / replay transports — a first-class offline-testing tool.
233// ---------------------------------------------------------------------------
234
235/// An [`HttpTransport`] that returns queued responses and records the requests
236/// it was given. Queue responses with [`MockHttpTransport::push_response`] /
237/// [`push_json`](MockHttpTransport::push_json) / [`push_error`](MockHttpTransport::push_error);
238/// they are returned FIFO.
239#[derive(Debug, Default)]
240pub struct MockHttpTransport {
241    responses: Mutex<VecDeque<Result<HttpResponse>>>,
242    requests: Mutex<Vec<HttpRequest>>,
243}
244
245impl MockHttpTransport {
246    /// An empty mock.
247    #[must_use]
248    pub fn new() -> Self {
249        Self::default()
250    }
251
252    /// Queue a successful response.
253    pub fn push_response(&self, response: HttpResponse) {
254        self.responses.lock().unwrap().push_back(Ok(response));
255    }
256
257    /// Queue a response from a status code and JSON body.
258    pub fn push_json(&self, status: u16, body: impl Into<String>) {
259        self.push_response(HttpResponse::new(status, body));
260    }
261
262    /// Queue a transport-level error.
263    pub fn push_error(&self, error: Error) {
264        self.responses.lock().unwrap().push_back(Err(error));
265    }
266
267    /// The requests executed so far, in order.
268    #[must_use]
269    pub fn recorded_requests(&self) -> Vec<HttpRequest> {
270        self.requests.lock().unwrap().clone()
271    }
272}
273
274impl HttpTransport for MockHttpTransport {
275    fn execute(&self, request: &HttpRequest) -> Result<HttpResponse> {
276        self.requests.lock().unwrap().push(request.clone());
277        self.responses
278            .lock()
279            .unwrap()
280            .pop_front()
281            .unwrap_or_else(|| Err(Error::Network("mock: no queued response".to_string())))
282    }
283}
284
285/// What the next [`connect`](WsTransport::connect) does.
286///
287/// A socket has three ways to behave badly and the mock could script none of
288/// them: it always opened, and always accepted whatever was sent. That left the
289/// reconnect helper's failure paths -- refused connection, subscriptions that
290/// cannot be replayed -- with no way to be reached from a test, so the one
291/// outcome that reads as a quiet market (a live socket subscribed to nothing)
292/// was never exercised.
293#[derive(Debug)]
294enum ScriptedConnection {
295    /// Opens, replays `frames` from `recv`, accepts everything sent.
296    Open(Vec<Result<Option<String>>>),
297    /// `connect` fails.
298    Refused,
299    /// Opens, then rejects the first `send`. This is the venue that accepts the
300    /// socket and drops it before the subscribe frames land.
301    OpenButUnsendable,
302}
303
304/// A [`WsTransport`] whose connections replay a scripted sequence of frames.
305///
306/// Each call to [`push_connection`](MockWsTransport::push_connection) queues the
307/// script (a list of `recv` results) for the next [`connect`](WsTransport::connect).
308/// Frames sent by the client are recorded and readable via
309/// [`sent`](MockWsTransport::sent).
310///
311/// [`push_refused_connection`](MockWsTransport::push_refused_connection) and
312/// [`push_unsendable_connection`](MockWsTransport::push_unsendable_connection)
313/// queue the two ways a connection can fail instead.
314#[derive(Debug, Default)]
315pub struct MockWsTransport {
316    scripts: Mutex<VecDeque<ScriptedConnection>>,
317    connected_urls: Mutex<Vec<String>>,
318    sent: Arc<Mutex<Vec<String>>>,
319}
320
321impl MockWsTransport {
322    /// An empty mock.
323    #[must_use]
324    pub fn new() -> Self {
325        Self::default()
326    }
327
328    /// Queue the frame script for the next connection. Each entry is the result
329    /// of one `recv`; `Ok(Some(frame))` delivers a frame, `Ok(None)` closes the
330    /// connection, `Err(..)` surfaces a read error.
331    pub fn push_connection(&self, frames: Vec<Result<Option<String>>>) {
332        self.scripts
333            .lock()
334            .unwrap()
335            .push_back(ScriptedConnection::Open(frames));
336    }
337
338    /// Queue a refused connection: the next `connect` returns an error.
339    pub fn push_refused_connection(&self) {
340        self.scripts
341            .lock()
342            .unwrap()
343            .push_back(ScriptedConnection::Refused);
344    }
345
346    /// Queue a connection that opens and then rejects everything sent on it.
347    ///
348    /// The shape that matters for reconnection: the socket comes up, so the
349    /// client believes it recovered, and the subscribe frames never land. What
350    /// follows is a live connection delivering nothing, which is what a quiet
351    /// market looks like.
352    pub fn push_unsendable_connection(&self) {
353        self.scripts
354            .lock()
355            .unwrap()
356            .push_back(ScriptedConnection::OpenButUnsendable);
357    }
358
359    /// The URLs connected to so far, in order.
360    #[must_use]
361    pub fn connected_urls(&self) -> Vec<String> {
362        self.connected_urls.lock().unwrap().clone()
363    }
364
365    /// The frames the client sent across all connections, in order.
366    #[must_use]
367    pub fn sent(&self) -> Vec<String> {
368        self.sent.lock().unwrap().clone()
369    }
370}
371
372impl WsTransport for MockWsTransport {
373    fn connect(&self, url: &str) -> Result<Box<dyn WsConnection>> {
374        self.connected_urls.lock().unwrap().push(url.to_string());
375        let script = self
376            .scripts
377            .lock()
378            .unwrap()
379            .pop_front()
380            .unwrap_or(ScriptedConnection::Open(Vec::new()));
381        let (frames, unsendable) = match script {
382            ScriptedConnection::Open(frames) => (frames, false),
383            ScriptedConnection::OpenButUnsendable => (Vec::new(), true),
384            ScriptedConnection::Refused => {
385                return Err(Error::Network("mock: connection refused".to_string()))
386            }
387        };
388        Ok(Box::new(MockWsConnection {
389            incoming: frames.into(),
390            sent: Arc::clone(&self.sent),
391            closed: false,
392            unsendable,
393        }))
394    }
395}
396
397/// A scripted [`WsConnection`] produced by [`MockWsTransport`].
398///
399/// Crate-internal: `connect` hands it back as a `Box<dyn WsConnection>`, its
400/// fields are private and it has no constructor, so a caller can neither name
401/// nor build one. It drives the transport, not the public API.
402#[derive(Debug)]
403pub(crate) struct MockWsConnection {
404    incoming: VecDeque<Result<Option<String>>>,
405    sent: Arc<Mutex<Vec<String>>>,
406    closed: bool,
407    /// Open, but every `send` fails. See
408    /// [`push_unsendable_connection`](MockWsTransport::push_unsendable_connection).
409    unsendable: bool,
410}
411
412impl WsConnection for MockWsConnection {
413    fn send(&mut self, text: &str) -> Result<()> {
414        if self.closed || self.unsendable {
415            return Err(Error::NotConnected);
416        }
417        self.sent.lock().unwrap().push(text.to_string());
418        Ok(())
419    }
420
421    fn recv(&mut self) -> Result<Option<String>> {
422        // An explicit `Ok(None)` frame in the script is a peer close; running out
423        // of scripted frames just means nothing more is pending (still open).
424        match self.incoming.pop_front() {
425            Some(Ok(None)) => {
426                self.closed = true;
427                Ok(None)
428            }
429            Some(other) => other,
430            None => Ok(None),
431        }
432    }
433
434    fn is_connected(&self) -> bool {
435        !self.closed
436    }
437
438    fn close(&mut self) -> Result<()> {
439        self.closed = true;
440        Ok(())
441    }
442}
443
444#[cfg(test)]
445mod tests {
446    use super::*;
447    use std::time::Duration;
448
449    #[test]
450    fn http_method_strings() {
451        assert_eq!(HttpMethod::Get.as_str(), "GET");
452        assert_eq!(HttpMethod::Post.as_str(), "POST");
453        assert_eq!(HttpMethod::Put.as_str(), "PUT");
454        assert_eq!(HttpMethod::Patch.as_str(), "PATCH");
455        assert_eq!(HttpMethod::Delete.as_str(), "DELETE");
456    }
457
458    #[test]
459    fn request_builder() {
460        let req = HttpRequest::get("https://api/x")
461            .with_header("X-KEY", "abc")
462            .with_body("payload");
463        assert_eq!(req.method, HttpMethod::Get);
464        assert_eq!(req.url, "https://api/x");
465        assert_eq!(req.headers, vec![("X-KEY".to_string(), "abc".to_string())]);
466        assert_eq!(req.body.as_deref(), Some("payload"));
467
468        let post = HttpRequest::post("https://api/y");
469        assert_eq!(post.method, HttpMethod::Post);
470        let del = HttpRequest::new(HttpMethod::Delete, "https://api/z");
471        assert_eq!(del.method, HttpMethod::Delete);
472    }
473
474    #[test]
475    fn response_success_classification() {
476        assert!(HttpResponse::new(200, "{}").is_success());
477        assert!(HttpResponse::new(204, "").is_success());
478        assert!(!HttpResponse::new(400, "{}").is_success());
479        assert!(!HttpResponse::new(500, "{}").is_success());
480    }
481
482    #[test]
483    fn a_header_is_found_whatever_its_case() {
484        let response = HttpResponse::new(429, "{}").with_header("retry-after", "3");
485        assert_eq!(response.header("Retry-After"), Some("3"));
486        assert_eq!(response.header("RETRY-AFTER"), Some("3"));
487        assert_eq!(response.header("X-Absent"), None);
488    }
489
490    #[test]
491    fn retry_after_reads_whole_and_fractional_seconds() {
492        let whole = HttpResponse::new(429, "{}").with_header("Retry-After", "3");
493        assert_eq!(whole.retry_after(), Some(Duration::from_secs(3)));
494        let fractional = HttpResponse::new(429, "{}").with_header("Retry-After", " 2.5 ");
495        assert_eq!(fractional.retry_after(), Some(Duration::from_millis(2500)));
496    }
497
498    #[test]
499    fn retry_after_declines_what_it_cannot_read() {
500        // No header at all, the HTTP-date form this deliberately does not parse,
501        // an outright non-number, and a negative wait. Each yields None rather
502        // than a wrong duration a caller would then sleep for.
503        assert_eq!(HttpResponse::new(429, "{}").retry_after(), None);
504        let date = HttpResponse::new(429, "{}")
505            .with_header("Retry-After", "Wed, 21 Oct 2026 07:28:00 GMT");
506        assert_eq!(date.retry_after(), None);
507        let junk = HttpResponse::new(429, "{}").with_header("Retry-After", "soon");
508        assert_eq!(junk.retry_after(), None);
509        let negative = HttpResponse::new(429, "{}").with_header("Retry-After", "-1");
510        assert_eq!(negative.retry_after(), None);
511    }
512
513    #[test]
514    fn mock_http_returns_queued_responses_fifo_and_records() {
515        let http = MockHttpTransport::new();
516        http.push_json(200, "{\"ok\":1}");
517        http.push_error(Error::Timeout);
518
519        let first = http.execute(&HttpRequest::get("https://api/a")).unwrap();
520        assert_eq!(first.body, "{\"ok\":1}");
521        let second = http.execute(&HttpRequest::post("https://api/b"));
522        assert_eq!(second.unwrap_err(), Error::Timeout);
523
524        // Queue exhausted -> network error.
525        let third = http.execute(&HttpRequest::get("https://api/c"));
526        assert!(matches!(third.unwrap_err(), Error::Network(_)));
527
528        let recorded = http.recorded_requests();
529        assert_eq!(recorded.len(), 3);
530        assert_eq!(recorded[0].url, "https://api/a");
531        assert_eq!(recorded[1].method, HttpMethod::Post);
532    }
533
534    #[test]
535    fn mock_ws_replays_frames_then_closes() {
536        let ws = MockWsTransport::new();
537        ws.push_connection(vec![
538            Ok(Some("frame-1".to_string())),
539            Ok(Some("frame-2".to_string())),
540        ]);
541
542        let mut conn = ws.connect("wss://stream/x").unwrap();
543        conn.send("subscribe").unwrap();
544        assert_eq!(conn.recv().unwrap(), Some("frame-1".to_string()));
545        assert_eq!(conn.recv().unwrap(), Some("frame-2".to_string()));
546        // Exhausted script reads as a clean close.
547        assert_eq!(conn.recv().unwrap(), None);
548
549        assert_eq!(ws.connected_urls(), vec!["wss://stream/x".to_string()]);
550        assert_eq!(ws.sent(), vec!["subscribe".to_string()]);
551    }
552
553    #[test]
554    fn mock_ws_surfaces_read_errors_and_blocks_send_after_close() {
555        let ws = MockWsTransport::new();
556        ws.push_connection(vec![Err(Error::Network("reset".to_string()))]);
557        let mut conn = ws.connect("wss://stream/y").unwrap();
558        assert!(matches!(conn.recv().unwrap_err(), Error::Network(_)));
559
560        conn.close().unwrap();
561        assert_eq!(conn.send("late").unwrap_err(), Error::NotConnected);
562    }
563
564    #[test]
565    fn mock_ws_connect_without_script_is_immediately_closed() {
566        let ws = MockWsTransport::new();
567        let mut conn = ws.connect("wss://stream/empty").unwrap();
568        assert_eq!(conn.recv().unwrap(), None);
569    }
570
571    /// The mocks are public test scaffolding; their `Debug` is part of what a
572    /// failing assertion prints.
573    #[test]
574    fn mock_transports_render_debug() {
575        assert!(format!("{:?}", MockHttpTransport::new()).starts_with("MockHttpTransport"));
576        assert!(format!("{:?}", MockWsTransport::new()).starts_with("MockWsTransport"));
577
578        // `connect` hands back a `Box<dyn WsConnection>`, and the trait object
579        // has no Debug -- build the concrete mock directly, which the test
580        // module can do because it sits in this file.
581        let conn = MockWsConnection {
582            incoming: VecDeque::new(),
583            sent: Arc::new(Mutex::new(Vec::new())),
584            closed: false,
585            unsendable: false,
586        };
587        assert!(format!("{conn:?}").starts_with("MockWsConnection"));
588    }
589}
wickra_exchange_core/transport.rs

wickra_exchange_core/
transport.rs
