netget/server/fix/
session.rs

1//! The FIX session layer both roles share: sequence numbers in both directions, gap detection
2//! with ResendRequest, answering a peer's ResendRequest with the stored application messages
3//! (PossDupFlag) and SequenceReset-GapFill for everything else, TestRequest / Heartbeat, Logout,
4//! and the timers that drive them.
5use super::codec::{self, Message};
6use crate::utils::clock::Instant;
7use anyhow::Result;
8use std::collections::VecDeque;
9use std::time::Duration;
10
11/// Sent application messages kept for resending.
12const RESEND_STORE: usize = 2048;
13/// Header and trailer tags the session owns; nothing else may set them.
14pub const SESSION_TAGS: &[u32] = &[8, 9, 10, 34, 35, 43, 49, 52, 56, 97, 122];
15
16struct Sent {
17    seq: u32,
18    msg_type: String,
19    sending_time: String,
20    body: Vec<(u32, String)>,
21}
22
23pub struct Session {
24    pub begin: String,
25    pub sender: String,
26    pub target: String,
27    pub heartbeat: Duration,
28    next_out: u32,
29    next_in: u32,
30    sent: VecDeque<Sent>,
31    /// Highest sequence number seen while a ResendRequest is outstanding.
32    resend_until: Option<u32>,
33    last_in: Instant,
34    last_out: Instant,
35    test_request: Option<(String, Instant)>,
36    test_counter: u32,
37    pub logout_sent: bool,
38}
39
40/// What an inbound message asks of the connection.
41pub struct Step {
42    /// Bytes to write now, in order.
43    pub send: Vec<Vec<u8>>,
44    /// An application message for the handler.
45    pub app: Option<Message>,
46    /// Close after writing, with the reason.
47    pub close: Option<String>,
48}
49
50impl Step {
51    fn send(bytes: Vec<Vec<u8>>) -> Self {
52        Self {
53            send: bytes,
54            app: None,
55            close: None,
56        }
57    }
58}
59
60/// Body fields of a message: everything but the header and trailer the session owns.
61pub fn body(m: &Message) -> Vec<(u32, String)> {
62    m.fields
63        .iter()
64        .filter(|(t, _)| !SESSION_TAGS.contains(t))
65        .cloned()
66        .collect()
67}
68
69impl Session {
70    pub fn new(begin: &str, sender: &str, target: &str, heartbeat: Duration) -> Self {
71        let now = Instant::now();
72        Self {
73            begin: begin.to_owned(),
74            sender: sender.to_owned(),
75            target: target.to_owned(),
76            heartbeat,
77            next_out: 1,
78            next_in: 1,
79            sent: VecDeque::new(),
80            resend_until: None,
81            last_in: now,
82            last_out: now,
83            test_request: None,
84            test_counter: 0,
85            logout_sent: false,
86        }
87    }
88
89    pub fn reset(&mut self) {
90        self.next_out = 1;
91        self.next_in = 1;
92        self.sent.clear();
93        self.resend_until = None;
94    }
95
96    pub fn next_in(&self) -> u32 {
97        self.next_in
98    }
99
100    pub fn next_out(&self) -> u32 {
101        self.next_out
102    }
103
104    /// Accept the peer's Logon as message `seq` (expected 1 after a reset).
105    pub fn accept_logon(&mut self, seq: u32) {
106        self.next_in = seq + 1;
107        self.last_in = Instant::now();
108    }
109
110    fn header(&self, msg_type: &str, seq: u32, poss_dup: Option<&str>) -> Vec<(u32, String)> {
111        let mut h = vec![
112            (35, msg_type.to_owned()),
113            (49, self.sender.clone()),
114            (56, self.target.clone()),
115            (34, seq.to_string()),
116        ];
117        if poss_dup.is_some() {
118            h.push((43, "Y".into()));
119        }
120        h.push((52, codec::timestamp()));
121        if let Some(orig) = poss_dup {
122            h.push((122, orig.to_owned()));
123        }
124        h
125    }
126
127    /// Encode the next outbound message; application messages are kept for resending.
128    pub fn encode(&mut self, msg_type: &str, body: &[(u32, String)]) -> Result<Vec<u8>> {
129        let seq = self.next_out;
130        let mut fields = self.header(msg_type, seq, None);
131        let sending_time = fields
132            .iter()
133            .find(|(t, _)| *t == 52)
134            .map(|(_, v)| v.clone())
135            .unwrap_or_default();
136        fields.extend_from_slice(body);
137        let bytes = codec::encode(&self.begin, &fields)?;
138        self.next_out += 1;
139        self.last_out = Instant::now();
140        if !super::dict::MESSAGES
141            .iter()
142            .any(|(t, _, admin)| *t == msg_type && *admin)
143        {
144            self.sent.push_back(Sent {
145                seq,
146                msg_type: msg_type.to_owned(),
147                sending_time,
148                body: body.to_vec(),
149            });
150            while self.sent.len() > RESEND_STORE {
151                self.sent.pop_front();
152            }
153        }
154        Ok(bytes)
155    }
156
157    pub fn logout(&mut self, text: Option<&str>) -> Vec<u8> {
158        self.logout_sent = true;
159        let body: Vec<(u32, String)> = text.map(|t| vec![(58, t.to_owned())]).unwrap_or_default();
160        self.encode("5", &body).unwrap_or_default()
161    }
162
163    fn reject(&mut self, ref_seq: u32, reason: u32, ref_tag: Option<u32>, text: &str) -> Vec<u8> {
164        let mut body = vec![(45, ref_seq.to_string())];
165        if let Some(t) = ref_tag {
166            body.push((371, t.to_string()));
167        }
168        body.push((373, reason.to_string()));
169        body.push((58, text.to_owned()));
170        self.encode("3", &body).unwrap_or_default()
171    }
172
173    /// Answer a ResendRequest for `[begin, end]` (end 0 = everything sent).
174    fn resend(&mut self, begin: u32, end: u32) -> Vec<Vec<u8>> {
175        let last = self.next_out - 1;
176        let end = if end == 0 || end > last { last } else { end };
177        let mut out = Vec::new();
178        let mut gap_from: Option<u32> = None;
179        let mut seq = begin.max(1);
180        while seq <= end {
181            match self.sent.iter().find(|s| s.seq == seq) {
182                Some(s) => {
183                    if let Some(g) = gap_from.take() {
184                        out.push(self.gap_fill(g, seq));
185                    }
186                    let mut fields = self.header(&s.msg_type, s.seq, Some(&s.sending_time));
187                    fields.extend_from_slice(&s.body);
188                    if let Ok(bytes) = codec::encode(&self.begin, &fields) {
189                        out.push(bytes);
190                    }
191                }
192                None => {
193                    gap_from.get_or_insert(seq);
194                }
195            }
196            seq += 1;
197        }
198        if let Some(g) = gap_from {
199            out.push(self.gap_fill(g, end + 1));
200        }
201        self.last_out = Instant::now();
202        out
203    }
204
205    fn gap_fill(&self, seq: u32, new_seq: u32) -> Vec<u8> {
206        let mut fields = self.header("4", seq, Some(&codec::timestamp()));
207        fields.push((123, "Y".into()));
208        fields.push((36, new_seq.to_string()));
209        codec::encode(&self.begin, &fields).unwrap_or_default()
210    }
211
212    /// Process one inbound message after logon.
213    pub fn on_message(&mut self, m: Message) -> Step {
214        self.last_in = Instant::now();
215        let msg_type = m.msg_type().to_owned();
216        let Some(seq) = m.seq() else {
217            return Step {
218                send: vec![self.logout(Some("MsgSeqNum missing"))],
219                app: None,
220                close: Some("MsgSeqNum missing".into()),
221            };
222        };
223        if m.get(49) != Some(self.target.as_str()) || m.get(56) != Some(self.sender.as_str()) {
224            let r = self.reject(
225                seq,
226                9,
227                Some(if m.get(49) != Some(self.target.as_str()) {
228                    49
229                } else {
230                    56
231                }),
232                "CompID problem",
233            );
234            return Step {
235                send: vec![r, self.logout(Some("CompID problem"))],
236                app: None,
237                close: Some("CompID problem".into()),
238            };
239        }
240        let poss_dup = m.get(43) == Some("Y");
241        // A SequenceReset in reset mode applies whatever its sequence number (FIX 4.4 vol 2).
242        if msg_type == "4" && m.get(123) != Some("Y") {
243            return match m.get(36).and_then(|v| v.parse::<u32>().ok()) {
244                Some(n) if n >= self.next_in => {
245                    self.next_in = n;
246                    self.resend_until = None;
247                    Step::send(vec![])
248                }
249                _ => {
250                    let r = self.reject(seq, 5, Some(36), "NewSeqNo is lower than expected");
251                    Step::send(vec![r])
252                }
253            };
254        }
255        if seq < self.next_in {
256            if poss_dup {
257                return Step::send(vec![]);
258            }
259            let text = format!(
260                "MsgSeqNum too low, expecting {} but received {seq}",
261                self.next_in
262            );
263            return Step {
264                send: vec![self.logout(Some(&text))],
265                app: None,
266                close: Some(text),
267            };
268        }
269        if seq > self.next_in {
270            let mut send = Vec::new();
271            match msg_type.as_str() {
272                "2" => send.extend(self.on_resend_request(&m)),
273                "5" => {
274                    let reply = if self.logout_sent {
275                        vec![]
276                    } else {
277                        vec![self.logout(None)]
278                    };
279                    return Step {
280                        send: reply,
281                        app: None,
282                        close: Some("peer logged out".into()),
283                    };
284                }
285                _ => {}
286            }
287            if self.resend_until.is_none() {
288                let from = self.next_in.to_string();
289                if let Ok(b) = self.encode("2", &[(7, from), (16, "0".into())]) {
290                    send.push(b);
291                }
292            }
293            self.resend_until = Some(self.resend_until.unwrap_or(0).max(seq));
294            return Step::send(send);
295        }
296        // In sequence.
297        self.next_in += 1;
298        if self.resend_until.is_some_and(|u| self.next_in > u) {
299            self.resend_until = None;
300        }
301        match msg_type.as_str() {
302            "0" => {
303                if let (Some(id), Some((want, _))) = (m.get(112), &self.test_request) {
304                    if id == want {
305                        self.test_request = None;
306                    }
307                }
308                Step::send(vec![])
309            }
310            "1" => {
311                let id = m.get(112).unwrap_or_default().to_owned();
312                let body: Vec<(u32, String)> = if id.is_empty() {
313                    vec![]
314                } else {
315                    vec![(112, id)]
316                };
317                Step::send(self.encode("0", &body).into_iter().collect())
318            }
319            "2" => Step::send(self.on_resend_request(&m)),
320            "3" => Step::send(vec![]),
321            "4" => match m.get(36).and_then(|v| v.parse::<u32>().ok()) {
322                Some(n) if n > seq => {
323                    self.next_in = n;
324                    if self.resend_until.is_some_and(|u| self.next_in > u) {
325                        self.resend_until = None;
326                    }
327                    Step::send(vec![])
328                }
329                _ => {
330                    let r = self.reject(seq, 5, Some(36), "GapFill NewSeqNo must exceed MsgSeqNum");
331                    Step::send(vec![r])
332                }
333            },
334            "5" => {
335                let reply = if self.logout_sent {
336                    vec![]
337                } else {
338                    vec![self.logout(None)]
339                };
340                Step {
341                    send: reply,
342                    app: None,
343                    close: Some(
344                        m.get(58)
345                            .map(|t| format!("peer logged out: {t}"))
346                            .unwrap_or_else(|| "peer logged out".into()),
347                    ),
348                }
349            }
350            "A" => {
351                let r = self.reject(seq, 11, Some(35), "already logged on");
352                Step::send(vec![r])
353            }
354            _ => {
355                if m.get(52).is_none() {
356                    let r = self.reject(seq, 1, Some(52), "Required tag missing");
357                    return Step::send(vec![r]);
358                }
359                Step {
360                    send: vec![],
361                    app: Some(m),
362                    close: None,
363                }
364            }
365        }
366    }
367
368    fn on_resend_request(&mut self, m: &Message) -> Vec<Vec<u8>> {
369        let begin = m.get(7).and_then(|v| v.parse().ok()).unwrap_or(1);
370        let end = m.get(16).and_then(|v| v.parse().ok()).unwrap_or(0);
371        self.resend(begin, end)
372    }
373
374    /// Timer duty: a Heartbeat when idle, a TestRequest when the peer is quiet, and a reason
375    /// to close when it never answers.
376    pub fn tick(&mut self) -> (Vec<Vec<u8>>, Option<String>) {
377        let now = Instant::now();
378        let mut out = Vec::new();
379        if let Some((_, sent)) = &self.test_request {
380            if now.duration_since(*sent) >= self.heartbeat {
381                let text = "no answer to TestRequest";
382                return (vec![self.logout(Some(text))], Some(text.into()));
383            }
384        } else if now.duration_since(self.last_in) >= self.heartbeat + self.heartbeat / 5 {
385            self.test_counter += 1;
386            let id = format!("TEST{}", self.test_counter);
387            if let Ok(b) = self.encode("1", &[(112, id.clone())]) {
388                out.push(b);
389            }
390            self.test_request = Some((id, now));
391        }
392        if now.duration_since(self.last_out) >= self.heartbeat {
393            if let Ok(b) = self.encode("0", &[]) {
394                out.push(b);
395            }
396        }
397        (out, None)
398    }
399}
netget/server/fix/session.rs

netget/server/fix/
session.rs
