haste_hl7v2/
mllp_async.rs

1//! Async MLLP framing, for a listener that serves many senders at once.
2//!
3//! [`crate::mllp::MllpFormatter::read_frame`] reads a byte at a time from a
4//! blocking [`std::io::Read`], which is fine for a client sending one message
5//! and wrong for a listener: it holds a thread per connection and cannot be
6//! cancelled, so one sender that opens a socket and stops writing occupies that
7//! thread until the process restarts.
8//!
9//! This reader is cancel-safe (a timeout around it drops cleanly), buffers
10//! rather than syscalling per byte, and refuses a frame past
11//! [`MAX_FRAME_BYTES`] instead of growing until the host runs out of memory.
12
13use haste_fhir_model::r4::generated::terminology::IssueType;
14use haste_fhir_operation_error::OperationOutcomeError;
15use tokio::io::{AsyncRead, AsyncReadExt};
16
17const START_BLOCK: u8 = 0x0B;
18const END_BLOCK: u8 = 0x1C;
19const CARRIAGE_RETURN: u8 = 0x0D;
20
21/// Largest message this listener accepts, in bytes.
22///
23/// A sender that never writes `<EB>` would otherwise grow the buffer forever.
24/// 16 MiB is far above any real HL7 v2 message a large one with embedded
25/// documents runs to a few hundred KiB so this rejects only a broken or
26/// hostile sender.
27pub const MAX_FRAME_BYTES: usize = 16 * 1024 * 1024;
28
29/// Why a frame read ended.
30#[derive(Debug)]
31pub enum FrameEnd {
32    /// A complete frame; the payload with framing bytes removed.
33    Frame(Vec<u8>),
34    /// The sender closed cleanly between frames. Not an error: a sender that
35    /// sends one message and hangs up is behaving correctly.
36    Closed,
37}
38
39/// Reads one MLLP frame from `reader`.
40///
41/// Returns [`FrameEnd::Closed`] when the stream ends *between* frames, and an
42/// error when it ends *within* one, since a truncated message must not be
43/// mistaken for a complete one.
44///
45/// # Errors
46///
47/// Returns an [`OperationOutcomeError`] when the stream does not begin with
48/// `<SB>`, the frame exceeds [`MAX_FRAME_BYTES`], the stream ends mid-frame, or
49/// the underlying read fails.
50pub async fn read_frame<R>(reader: &mut R) -> Result<FrameEnd, OperationOutcomeError>
51where
52    R: AsyncRead + Unpin,
53{
54    // Leading bytes before `<SB>` are a framing error, not something to skip:
55    // silently resynchronising would accept a corrupted stream as valid.
56    let first = match reader.read_u8().await {
57        Ok(byte) => byte,
58        Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => return Ok(FrameEnd::Closed),
59        Err(e) => return Err(read_error(&e)),
60    };
61
62    if first != START_BLOCK {
63        return Err(OperationOutcomeError::error(
64            IssueType::invalid(),
65            format!("Expected an MLLP frame to start with <SB> (0x0B), got 0x{first:02X}"),
66        ));
67    }
68
69    let mut payload = Vec::with_capacity(4096);
70    // `<EB>` is only a terminator when `<CR>` follows it, and a payload may
71    // legitimately contain 0x1C, so the previous byte has to be remembered.
72    let mut pending_end_block = false;
73
74    loop {
75        let byte = match reader.read_u8().await {
76            Ok(byte) => byte,
77            Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => {
78                return Err(OperationOutcomeError::error(
79                    IssueType::incomplete(),
80                    format!(
81                        "Stream ended after {} byte(s) without <EB><CR>; \
82                         a truncated message is not acknowledged",
83                        payload.len()
84                    ),
85                ));
86            }
87            Err(e) => return Err(read_error(&e)),
88        };
89
90        if pending_end_block {
91            if byte == CARRIAGE_RETURN {
92                return Ok(FrameEnd::Frame(payload));
93            }
94            // The 0x1C was payload after all.
95            payload.push(END_BLOCK);
96            pending_end_block = false;
97        }
98
99        if byte == END_BLOCK {
100            pending_end_block = true;
101            continue;
102        }
103
104        payload.push(byte);
105
106        if payload.len() > MAX_FRAME_BYTES {
107            return Err(OperationOutcomeError::error(
108                IssueType::too_costly(),
109                format!("MLLP frame exceeded the {MAX_FRAME_BYTES} byte limit"),
110            ));
111        }
112    }
113}
114
115fn read_error(error: &std::io::Error) -> OperationOutcomeError {
116    OperationOutcomeError::error(
117        IssueType::exception(),
118        format!("Failed to read from the MLLP stream: {error}"),
119    )
120}
121
122/// Wraps `payload` in `<SB>`/`<EB><CR>`.
123#[must_use]
124pub fn encode_frame(payload: &[u8]) -> Vec<u8> {
125    let mut framed = Vec::with_capacity(payload.len() + 3);
126    framed.push(START_BLOCK);
127    framed.extend_from_slice(payload);
128    framed.push(END_BLOCK);
129    framed.push(CARRIAGE_RETURN);
130    framed
131}
haste_hl7v2/mllp_async.rs

haste_hl7v2/
mllp_async.rs
