netget/server/hl7/
wire.rs

1//! MLLP framing (`<VT> message <FS><CR>`) and HL7 v2 ER7 segments, shared by both roles.
2//!
3//! Messages are exposed as structured segments: `{id, fields}` where `fields[0]` is the first
4//! field after the segment id (for MSH, `fields[0]` is MSH-2, the encoding characters, so
5//! `fields[n]` is always field n+1 — MSH-9 is `fields[7]`). Field text keeps the message's own
6//! component (`^`), repetition (`~`), sub-component (`&`) and escape (`\`) characters; the
7//! field separator and segment terminator can never appear inside a field.
8use anyhow::{bail, ensure, Context, Result};
9use serde::{Deserialize, Serialize};
10use serde_json::{json, Value};
11use std::time::Duration;
12use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
13
14pub const START: u8 = 0x0b;
15pub const END: [u8; 2] = [0x1c, 0x0d];
16pub const MAX_MESSAGE_BYTES: usize = 1024 * 1024;
17pub const MAX_SEGMENTS: usize = 4096;
18pub const MAX_FIELDS: usize = 512;
19pub const MAX_FIELD_BYTES: usize = 64 * 1024;
20pub const IO_TIMEOUT: Duration = Duration::from_secs(30);
21
22#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
23#[serde(deny_unknown_fields)]
24pub struct Segment {
25    pub id: String,
26    #[serde(default)]
27    pub fields: Vec<String>,
28}
29
30fn segment_id(id: &str) -> Result<()> {
31    ensure!(
32        id.len() == 3
33            && id
34                .bytes()
35                .all(|b| b.is_ascii_uppercase() || b.is_ascii_digit())
36            && id.as_bytes()[0].is_ascii_uppercase(),
37        "segment id '{id}' must be three upper-case letters or digits"
38    );
39    Ok(())
40}
41
42/// A field the model supplies: structure characters stay, the field separator is escaped
43/// (`\F\`), and a CR, LF or other control character — which would end the segment and forge
44/// the next one — is refused.
45pub fn field(text: &str) -> Result<String> {
46    ensure!(
47        text.len() <= MAX_FIELD_BYTES,
48        "HL7 field exceeds {MAX_FIELD_BYTES} bytes"
49    );
50    ensure!(
51        !text.chars().any(|c| c.is_control()),
52        "HL7 field contains a control character (CR would start a forged segment)"
53    );
54    Ok(text.replace('|', "\\F\\"))
55}
56
57/// A parsed message.
58#[derive(Clone, Debug, PartialEq)]
59pub struct Message {
60    pub segments: Vec<Segment>,
61    /// true when the bytes were not UTF-8 and were read as ISO-8859-1 (every byte maps).
62    pub latin1: bool,
63}
64
65impl Message {
66    pub fn msh(&self, n: usize) -> &str {
67        // MSH-n is fields[n-2] (MSH-1 is the separator itself).
68        self.segments
69            .first()
70            .and_then(|s| s.fields.get(n.wrapping_sub(2)))
71            .map(String::as_str)
72            .unwrap_or("")
73    }
74    pub fn control_id(&self) -> &str {
75        self.msh(10)
76    }
77    pub fn message_type(&self) -> &str {
78        self.msh(9)
79    }
80    pub fn to_event(&self) -> Value {
81        json!({
82            "message_type": self.message_type(),
83            "control_id": self.control_id(),
84            "version": self.msh(12),
85            "processing_id": self.msh(11),
86            "sending_application": self.msh(3),
87            "sending_facility": self.msh(4),
88            "receiving_application": self.msh(5),
89            "receiving_facility": self.msh(6),
90            "segments": self.segments,
91            "charset_assumed": if self.latin1 { Value::from("ISO-8859-1") } else { Value::Null },
92        })
93    }
94}
95
96/// Parse one ER7 message (the bytes between the MLLP start and end blocks).
97pub fn parse(bytes: &[u8]) -> Result<Message> {
98    ensure!(
99        bytes.len() <= MAX_MESSAGE_BYTES,
100        "HL7 message exceeds {MAX_MESSAGE_BYTES} bytes"
101    );
102    let (text, latin1) = match std::str::from_utf8(bytes) {
103        Ok(t) => (t.to_owned(), false),
104        Err(_) => (bytes.iter().map(|&b| b as char).collect::<String>(), true),
105    };
106    ensure!(text.starts_with("MSH"), "HL7 message must start with MSH");
107    let separator = text
108        .chars()
109        .nth(3)
110        .context("MSH without a field separator")?;
111    ensure!(
112        separator == '|',
113        "only the standard field separator '|' is supported"
114    );
115    let mut segments = Vec::new();
116    for line in text.split(['\r', '\n']).filter(|l| !l.is_empty()) {
117        ensure!(
118            segments.len() < MAX_SEGMENTS,
119            "HL7 message exceeds {MAX_SEGMENTS} segments"
120        );
121        let mut parts = line.split(separator);
122        let id = parts.next().unwrap_or_default().to_owned();
123        segment_id(&id)?;
124        let fields: Vec<String> = parts.map(str::to_owned).collect();
125        ensure!(
126            fields.len() <= MAX_FIELDS,
127            "segment {id} exceeds {MAX_FIELDS} fields"
128        );
129        ensure!(
130            fields
131                .iter()
132                .all(|f| f.len() <= MAX_FIELD_BYTES && !f.chars().any(|c| c.is_control())),
133            "segment {id} has an over-long field or a control character"
134        );
135        segments.push(Segment { id, fields });
136    }
137    ensure!(segments[0].id == "MSH", "first segment must be MSH");
138    let encoding = segments[0].fields.first().map(String::as_str).unwrap_or("");
139    ensure!(
140        encoding.starts_with("^~\\&"),
141        "MSH-2 must declare the standard encoding characters ^~\\&"
142    );
143    let message = Message { segments, latin1 };
144    ensure!(
145        !message.control_id().is_empty(),
146        "MSH-10 (message control id) is required"
147    );
148    ensure!(
149        !message.message_type().is_empty(),
150        "MSH-9 (message type) is required"
151    );
152    Ok(message)
153}
154
155fn render(segments: &[Segment]) -> Result<Vec<u8>> {
156    let mut out = String::new();
157    for s in segments {
158        segment_id(&s.id)?;
159        out.push_str(&s.id);
160        for (i, f) in s.fields.iter().enumerate() {
161            out.push('|');
162            if s.id == "MSH" && i == 0 {
163                out.push_str(f);
164            } else {
165                out.push_str(&field(f)?);
166            }
167        }
168        out.push('\r');
169        ensure!(
170            out.len() <= MAX_MESSAGE_BYTES,
171            "HL7 message exceeds {MAX_MESSAGE_BYTES} bytes"
172        );
173    }
174    Ok(out.into_bytes())
175}
176
177/// MLLP-frame a message.
178pub fn frame(message: &[u8]) -> Vec<u8> {
179    let mut out = Vec::with_capacity(message.len() + 3);
180    out.push(START);
181    out.extend_from_slice(message);
182    out.extend_from_slice(&END);
183    out
184}
185
186/// HL7 DTM, second precision, UTC.
187pub fn timestamp() -> String {
188    let secs = crate::utils::clock::SystemTime::now()
189        .duration_since(crate::utils::clock::UNIX_EPOCH)
190        .map(|d| d.as_secs())
191        .unwrap_or(0);
192    let days = (secs / 86_400) as i64;
193    let rem = secs % 86_400;
194    // Civil-from-days (Howard Hinnant), proleptic Gregorian.
195    let z = days + 719_468;
196    let era = z.div_euclid(146_097);
197    let doe = z - era * 146_097;
198    let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365;
199    let y = yoe + era * 400;
200    let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
201    let mp = (5 * doy + 2) / 153;
202    let d = doy - (153 * mp + 2) / 5 + 1;
203    let m = if mp < 10 { mp + 3 } else { mp - 9 };
204    let y = if m <= 2 { y + 1 } else { y };
205    format!(
206        "{y:04}{m:02}{d:02}{:02}{:02}{:02}",
207        rem / 3600,
208        rem % 3600 / 60,
209        rem % 60
210    )
211}
212
213/// Header fields a sender chooses.
214pub struct Header<'a> {
215    pub sending_application: &'a str,
216    pub sending_facility: &'a str,
217    pub receiving_application: &'a str,
218    pub receiving_facility: &'a str,
219    pub message_type: &'a str,
220    pub control_id: &'a str,
221    pub processing_id: &'a str,
222    pub version: &'a str,
223}
224
225pub fn msh(h: &Header) -> Result<Segment> {
226    for (name, v) in [
227        ("message_type", h.message_type),
228        ("control_id", h.control_id),
229        ("version", h.version),
230    ] {
231        ensure!(!v.is_empty(), "{name} is required");
232    }
233    ensure!(
234        h.control_id.len() <= 199,
235        "control id exceeds 199 characters"
236    );
237    let fields = [
238        "^~\\&",
239        h.sending_application,
240        h.sending_facility,
241        h.receiving_application,
242        h.receiving_facility,
243        &timestamp(),
244        "",
245        h.message_type,
246        h.control_id,
247        h.processing_id,
248        h.version,
249    ];
250    Ok(Segment {
251        id: "MSH".into(),
252        fields: fields.iter().map(|s| s.to_string()).collect(),
253    })
254}
255
256/// Build a message from a header and body segments (no MSH among them).
257pub fn build(header: &Header, body: &[Segment]) -> Result<Vec<u8>> {
258    ensure!(
259        body.len() < MAX_SEGMENTS,
260        "HL7 message exceeds {MAX_SEGMENTS} segments"
261    );
262    ensure!(
263        body.iter().all(|s| s.id != "MSH"),
264        "MSH is built by Rust; do not supply it"
265    );
266    let mut segments = vec![msh(header)?];
267    segments.extend(body.iter().cloned());
268    render(&segments)
269}
270
271pub const ACK_CODES: &[&str] = &["AA", "AE", "AR", "CA", "CE", "CR"];
272
273/// The acknowledgment for `original`: MSH with sender and receiver swapped, ACK^<trigger>^ACK,
274/// MSA echoing the original control id, optional ERR, then any extra body segments.
275pub fn ack(
276    original: &Message,
277    code: &str,
278    text: &str,
279    control_id: &str,
280    error: Option<&Value>,
281    extra: &[Segment],
282) -> Result<Vec<u8>> {
283    ensure!(
284        ACK_CODES.contains(&code),
285        "acknowledgment code must be one of {ACK_CODES:?}"
286    );
287    let trigger = original.message_type().split('^').nth(1).unwrap_or("");
288    let ack_type = if trigger.is_empty() {
289        "ACK".to_owned()
290    } else {
291        format!("ACK^{trigger}^ACK")
292    };
293    let header = Header {
294        sending_application: original.msh(5),
295        sending_facility: original.msh(6),
296        receiving_application: original.msh(3),
297        receiving_facility: original.msh(4),
298        message_type: &ack_type,
299        control_id,
300        processing_id: original.msh(11),
301        version: if original.msh(12).is_empty() {
302            "2.5"
303        } else {
304            original.msh(12)
305        },
306    };
307    let mut body = vec![Segment {
308        id: "MSA".into(),
309        fields: vec![code.into(), original.control_id().into(), text.into()],
310    }];
311    if let Some(e) = error {
312        let s = |k: &str| e.get(k).and_then(Value::as_str).unwrap_or("").to_owned();
313        // ERR-3 HL7 error code (CWE), ERR-4 severity (E/W/I), ERR-8 user message.
314        body.push(Segment {
315            id: "ERR".into(),
316            fields: vec![
317                String::new(),
318                s("location"),
319                s("code"),
320                s("severity"),
321                String::new(),
322                String::new(),
323                String::new(),
324                s("message"),
325            ],
326        });
327    }
328    body.extend(extra.iter().cloned());
329    build(&header, &body)
330}
331
332/// Read one MLLP frame: bytes before the start block are refused, the length is bounded
333/// before it grows, and the end block must be `<FS><CR>`.
334pub async fn read_frame<R: AsyncRead + Unpin>(
335    reader: &mut R,
336    idle: Duration,
337) -> Result<Option<Vec<u8>>> {
338    let mut byte = [0u8; 1];
339    let n = tokio::time::timeout(idle, reader.read(&mut byte))
340        .await
341        .context("MLLP idle deadline")??;
342    if n == 0 {
343        return Ok(None);
344    }
345    ensure!(byte[0] == START, "MLLP frame must begin with <VT> (0x0B)");
346    tokio::time::timeout(IO_TIMEOUT, async {
347        let mut message = Vec::new();
348        let mut buf = [0u8; 8192];
349        loop {
350            let n = reader.read(&mut buf).await?;
351            if n == 0 {
352                bail!("peer closed inside an MLLP frame");
353            }
354            message.extend_from_slice(&buf[..n]);
355            if let Some(end) = message.windows(2).position(|w| w == END) {
356                ensure!(
357                    end + 2 == message.len(),
358                    "bytes after the MLLP end block (one message per frame, one frame in flight)"
359                );
360                message.truncate(end);
361                return Ok(Some(message));
362            }
363            ensure!(
364                !message.contains(&0x1c),
365                "<FS> without <CR> inside an MLLP frame"
366            );
367            ensure!(!message.contains(&START), "<VT> inside an MLLP frame");
368            ensure!(
369                message.len() <= MAX_MESSAGE_BYTES + 2,
370                "MLLP frame exceeds {MAX_MESSAGE_BYTES} bytes"
371            );
372        }
373    })
374    .await
375    .context("MLLP whole-frame deadline")?
376}
377
378pub async fn write_frame<W: AsyncWrite + Unpin>(writer: &mut W, message: &[u8]) -> Result<usize> {
379    let framed = frame(message);
380    tokio::time::timeout(IO_TIMEOUT, async {
381        writer.write_all(&framed).await?;
382        writer.flush().await
383    })
384    .await
385    .context("MLLP write deadline")??;
386    Ok(framed.len())
387}
388
389pub fn segments_from(v: &Value) -> Result<Vec<Segment>> {
390    match v {
391        Value::Null => Ok(Vec::new()),
392        Value::Array(_) => {
393            let segments: Vec<Segment> = serde_json::from_value(v.clone())
394                .context("segments must be [{id, fields: [text]}]")?;
395            ensure!(segments.len() < MAX_SEGMENTS, "too many segments");
396            for s in &segments {
397                segment_id(&s.id)?;
398                ensure!(
399                    s.fields.len() <= MAX_FIELDS,
400                    "segment {} exceeds {MAX_FIELDS} fields",
401                    s.id
402                );
403                for f in &s.fields {
404                    field(f)?;
405                }
406            }
407            Ok(segments)
408        }
409        _ => bail!("segments must be an array"),
410    }
411}
netget/server/hl7/wire.rs

netget/server/hl7/
wire.rs
