hl7v2 1.3.0

HL7 v2 message parser and processor for Rust
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288

//! MLLP TCP client for sending HL7 messages.
//!
//! This module provides an async TCP client that:
//! - Connects to MLLP servers
//! - Encodes and sends HL7 messages with MLLP framing
//! - Receives and decodes ACK responses

#![expect(
    clippy::map_err_ignore,
    clippy::missing_errors_doc,
    reason = "pre-existing network client implementation debt moved from staged microcrate into hl7v2; cleanup is split from topology collapse"
)]

use super::codec::MllpCodec;
use crate::{Message, parse, write};
use bytes::BytesMut;
use futures::prelude::*;
use std::net::SocketAddr;
use std::time::Duration;
use tokio::net::TcpStream;
use tokio::time::timeout;
use tokio_util::codec::Framed;

/// Configuration for MLLP client
#[derive(Debug, Clone)]
pub struct MllpClientConfig {
    /// Connection timeout
    pub connect_timeout: Duration,
    /// Read timeout for responses
    pub read_timeout: Duration,
    /// Write timeout for sending messages
    pub write_timeout: Duration,
    /// Maximum frame size
    pub max_frame_size: usize,
}

impl Default for MllpClientConfig {
    fn default() -> Self {
        Self {
            connect_timeout: Duration::from_secs(10),
            read_timeout: Duration::from_secs(30),
            write_timeout: Duration::from_secs(30),
            max_frame_size: 10 * 1024 * 1024, // 10MB
        }
    }
}

/// MLLP TCP client
pub struct MllpClient {
    config: MllpClientConfig,
    framed: Option<Framed<TcpStream, MllpCodec>>,
    peer_addr: Option<SocketAddr>,
}

impl MllpClient {
    /// Create a new MLLP client with the given configuration
    pub fn new(config: MllpClientConfig) -> Self {
        Self {
            config,
            framed: None,
            peer_addr: None,
        }
    }

    /// Create a new MLLP client with default configuration
    pub fn with_default_config() -> Self {
        Self::new(MllpClientConfig::default())
    }

    /// Connect to a remote MLLP server
    pub async fn connect(&mut self, addr: impl Into<SocketAddr>) -> Result<(), std::io::Error> {
        let addr = addr.into();

        let stream = timeout(self.config.connect_timeout, TcpStream::connect(addr))
            .await
            .map_err(|_| {
                std::io::Error::new(std::io::ErrorKind::TimedOut, "Connection timeout")
            })??;

        let codec = MllpCodec::with_max_frame_size(self.config.max_frame_size);
        self.framed = Some(Framed::new(stream, codec));
        self.peer_addr = Some(addr);

        Ok(())
    }

    /// Check if the client is connected
    pub fn is_connected(&self) -> bool {
        self.framed.is_some()
    }

    /// Get the peer address if connected
    pub fn peer_addr(&self) -> Option<SocketAddr> {
        self.peer_addr
    }

    /// Send a message and wait for an ACK response
    pub async fn send_message(&mut self, message: &Message) -> Result<Message, std::io::Error> {
        let framed = self.framed.as_mut().ok_or_else(|| {
            std::io::Error::new(std::io::ErrorKind::NotConnected, "Not connected")
        })?;

        // Serialize the message
        let bytes = write(message);

        // Send the message with timeout
        timeout(
            self.config.write_timeout,
            framed.send(BytesMut::from(&bytes[..])),
        )
        .await
        .map_err(|_| std::io::Error::new(std::io::ErrorKind::TimedOut, "Write timeout"))??;

        // Wait for ACK response with timeout
        let response = timeout(self.config.read_timeout, framed.next())
            .await
            .map_err(|_| std::io::Error::new(std::io::ErrorKind::TimedOut, "Read timeout"))?
            .ok_or_else(|| {
                std::io::Error::new(std::io::ErrorKind::UnexpectedEof, "Connection closed")
            })??;

        // Parse the ACK
        let ack = parse(&response)
            .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e.to_string()))?;

        Ok(ack)
    }

    /// Send a message without waiting for a response (fire-and-forget)
    pub async fn send_message_no_ack(&mut self, message: &Message) -> Result<(), std::io::Error> {
        let framed = self.framed.as_mut().ok_or_else(|| {
            std::io::Error::new(std::io::ErrorKind::NotConnected, "Not connected")
        })?;

        // Serialize the message
        let bytes = write(message);

        // Send the message with timeout
        timeout(
            self.config.write_timeout,
            framed.send(BytesMut::from(&bytes[..])),
        )
        .await
        .map_err(|_| std::io::Error::new(std::io::ErrorKind::TimedOut, "Write timeout"))??;

        Ok(())
    }

    /// Receive a message from the server
    pub async fn receive_message(&mut self) -> Result<Option<Message>, std::io::Error> {
        let framed = self.framed.as_mut().ok_or_else(|| {
            std::io::Error::new(std::io::ErrorKind::NotConnected, "Not connected")
        })?;

        match timeout(self.config.read_timeout, framed.next()).await {
            Ok(Some(Ok(frame))) => {
                let message = parse(&frame).map_err(|e| {
                    std::io::Error::new(std::io::ErrorKind::InvalidData, e.to_string())
                })?;
                Ok(Some(message))
            }
            Ok(Some(Err(e))) => Err(e),
            Ok(None) => Ok(None), // Connection closed
            Err(_) => Err(std::io::Error::new(
                std::io::ErrorKind::TimedOut,
                "Read timeout",
            )),
        }
    }

    /// Close the connection
    pub async fn close(mut self) -> Result<(), std::io::Error> {
        if let Some(framed) = self.framed.take() {
            // Get the underlying stream and shut it down
            let stream = framed.into_inner();
            // Just dropping the stream will close it
            drop(stream);
        }
        Ok(())
    }

    /// Disconnect without consuming the client (allows reconnection)
    pub async fn disconnect(&mut self) -> Result<(), std::io::Error> {
        if let Some(framed) = self.framed.take() {
            // Get the underlying stream and shut it down
            let stream = framed.into_inner();
            // Just dropping the stream will close it
            drop(stream);
        }
        self.peer_addr = None;
        Ok(())
    }
}

/// Builder for creating MLLP clients with custom configuration
pub struct MllpClientBuilder {
    config: MllpClientConfig,
}

impl MllpClientBuilder {
    /// Create a new client builder with default configuration
    pub fn new() -> Self {
        Self {
            config: MllpClientConfig::default(),
        }
    }

    /// Set the connection timeout
    pub fn connect_timeout(mut self, timeout: Duration) -> Self {
        self.config.connect_timeout = timeout;
        self
    }

    /// Set the read timeout
    pub fn read_timeout(mut self, timeout: Duration) -> Self {
        self.config.read_timeout = timeout;
        self
    }

    /// Set the write timeout
    pub fn write_timeout(mut self, timeout: Duration) -> Self {
        self.config.write_timeout = timeout;
        self
    }

    /// Set the maximum frame size
    pub fn max_frame_size(mut self, size: usize) -> Self {
        self.config.max_frame_size = size;
        self
    }

    /// Build the client
    pub fn build(self) -> MllpClient {
        MllpClient::new(self.config)
    }
}

impl Default for MllpClientBuilder {
    fn default() -> Self {
        Self::new()
    }
}

#[cfg(test)]
mod tests {
    #![expect(
        clippy::unwrap_used,
        reason = "pre-existing network client test debt moved into hl7v2; cleanup is split from topology collapse"
    )]

    use super::*;

    #[test]
    fn test_client_builder() {
        let client = MllpClientBuilder::new()
            .connect_timeout(Duration::from_secs(5))
            .read_timeout(Duration::from_secs(15))
            .write_timeout(Duration::from_secs(15))
            .max_frame_size(1024 * 1024)
            .build();

        assert_eq!(client.config.connect_timeout, Duration::from_secs(5));
        assert_eq!(client.config.read_timeout, Duration::from_secs(15));
        assert_eq!(client.config.write_timeout, Duration::from_secs(15));
        assert_eq!(client.config.max_frame_size, 1024 * 1024);
    }

    #[test]
    fn test_client_not_connected() {
        let client = MllpClient::with_default_config();
        assert!(!client.is_connected());
        assert!(client.peer_addr().is_none());
    }

    #[tokio::test]
    async fn test_client_connect_timeout() {
        use std::net::SocketAddr;

        let mut client = MllpClientBuilder::new()
            .connect_timeout(Duration::from_millis(1))
            .build();

        // Try to connect to a non-routable address (should timeout)
        let addr: SocketAddr = "192.0.2.1:2575".parse().unwrap();
        let result = client.connect(addr).await;
        assert!(result.is_err());
    }
}