quickfix_tokio/
parser.rs

1//! Stream framing: carve complete FIX messages out of a raw byte stream.
2//!
3//! Mirrors the reference engines' resync behavior: discard garbage before
4//! `8=`, use BodyLength(9) to jump the body, then locate `<SOH>10=...<SOH>`.
5//! A frame that violates this structure is skipped without dropping the
6//! connection.
7
8use bytes::{Buf, Bytes, BytesMut};
9
10use crate::message::SOH;
11
12#[derive(Debug)]
13pub enum Frame {
14    /// A complete message: bytes from `8=` through the SOH after CheckSum.
15    Message(Bytes),
16    /// More bytes are needed.
17    Incomplete,
18}
19
20/// Extract the next complete FIX message from `buf`, consuming it (and any
21/// preceding garbage). Returns `Incomplete` when the buffer holds no full
22/// message yet. Garbled frames (bad BodyLength structure) are discarded and
23/// scanning resumes at the next `8=`.
24pub fn extract_frame(buf: &mut BytesMut) -> Frame {
25    loop {
26        // Resync: drop everything before "8=".
27        match find(buf, b"8=") {
28            Some(start) => {
29                if start > 0 {
30                    buf.advance(start);
31                }
32            }
33            None => {
34                // Keep a trailing '8' in case '=' arrives next.
35                let keep = if buf.last() == Some(&b'8') { 1 } else { 0 };
36                let drop = buf.len() - keep;
37                buf.advance(drop);
38                return Frame::Incomplete;
39            }
40        }
41
42        // Need "...<SOH>9=<len><SOH>" next.
43        let Some(soh1) = find(buf, &[SOH]) else { return Frame::Incomplete };
44        let after_begin = soh1 + 1;
45        if buf.len() < after_begin + 2 {
46            return Frame::Incomplete;
47        }
48        if &buf[after_begin..after_begin + 2] != b"9=" {
49            // Garbled: skip this "8=" and resync.
50            buf.advance(2);
51            continue;
52        }
53        let len_start = after_begin + 2;
54        let Some(rel_soh2) = buf[len_start..].iter().position(|&b| b == SOH) else {
55            return Frame::Incomplete;
56        };
57        let body_len: usize = match std::str::from_utf8(&buf[len_start..len_start + rel_soh2])
58            .ok()
59            .and_then(|s| s.parse().ok())
60        {
61            Some(n) => n,
62            None => {
63                buf.advance(2);
64                continue;
65            }
66        };
67        let body_start = len_start + rel_soh2 + 1;
68
69        // The body nominally ends where "10=" begins. Like the C++ parser,
70        // search *forward* from there for "<SOH>10=" — a lying BodyLength
71        // still yields a frame, and Message::parse then fails its
72        // length/checksum validation so the session ignores it as garbled.
73        let search_from = (body_start + body_len).saturating_sub(1);
74        if search_from >= buf.len() {
75            return Frame::Incomplete;
76        }
77        let Some(rel_cs) = find(&buf[search_from..], b"\x0110=") else {
78            return Frame::Incomplete;
79        };
80        let after_cs_tag = search_from + rel_cs + 4;
81        let Some(rel_end) = buf[after_cs_tag..].iter().position(|&b| b == SOH) else {
82            return Frame::Incomplete;
83        };
84        let end = after_cs_tag + rel_end + 1;
85        return Frame::Message(buf.split_to(end).freeze());
86    }
87}
88
89fn find(haystack: &[u8], needle: &[u8]) -> Option<usize> {
90    haystack.windows(needle.len()).position(|w| w == needle)
91}
92
93#[cfg(test)]
94mod tests {
95    use super::*;
96    use crate::message::build_raw;
97
98    fn msg() -> Vec<u8> {
99        build_raw(&[(8, "FIX.4.2"), (35, "0"), (34, "2"), (49, "A"), (56, "B")])
100    }
101
102    #[test]
103    fn extracts_single_message() {
104        let raw = msg();
105        let mut buf = BytesMut::from(&raw[..]);
106        match extract_frame(&mut buf) {
107            Frame::Message(m) => assert_eq!(&m[..], &raw[..]),
108            _ => panic!("expected message"),
109        }
110        assert!(buf.is_empty());
111    }
112
113    #[test]
114    fn incomplete_returns_incomplete() {
115        let raw = msg();
116        let mut buf = BytesMut::from(&raw[..raw.len() - 5]);
117        assert!(matches!(extract_frame(&mut buf), Frame::Incomplete));
118        buf.extend_from_slice(&raw[raw.len() - 5..]);
119        assert!(matches!(extract_frame(&mut buf), Frame::Message(_)));
120    }
121
122    #[test]
123    fn skips_leading_garbage() {
124        let raw = msg();
125        let mut buf = BytesMut::from(&b"garbage\x01noise"[..]);
126        buf.extend_from_slice(&raw);
127        match extract_frame(&mut buf) {
128            Frame::Message(m) => assert_eq!(&m[..], &raw[..]),
129            _ => panic!("expected message"),
130        }
131    }
132
133    #[test]
134    fn frames_message_with_lying_body_length() {
135        // A wrong BodyLength still frames (forward search for 10=); the
136        // session detects the mismatch during parse and ignores the message.
137        // This mirrors the C++ Parser and is required by acceptance test 2m.
138        let garbled = b"8=FIX.4.2\x019=3\x0135=D\x0158=hi\x0110=000\x01";
139        let raw = msg();
140        let mut buf = BytesMut::new();
141        buf.extend_from_slice(garbled);
142        buf.extend_from_slice(&raw);
143        match extract_frame(&mut buf) {
144            Frame::Message(m) => {
145                assert_eq!(&m[..], &garbled[..]);
146                assert!(crate::message::Message::parse(&m, true).is_err());
147            }
148            _ => panic!("expected the garbled frame"),
149        }
150        match extract_frame(&mut buf) {
151            Frame::Message(m) => assert_eq!(&m[..], &raw[..]),
152            _ => panic!("expected the valid frame after the garbled one"),
153        }
154    }
155
156    #[test]
157    fn two_messages_back_to_back() {
158        let raw = msg();
159        let mut buf = BytesMut::new();
160        buf.extend_from_slice(&raw);
161        buf.extend_from_slice(&raw);
162        assert!(matches!(extract_frame(&mut buf), Frame::Message(_)));
163        assert!(matches!(extract_frame(&mut buf), Frame::Message(_)));
164        assert!(matches!(extract_frame(&mut buf), Frame::Incomplete));
165    }
166}
quickfix_tokio/parser.rs

quickfix_tokio/
parser.rs
