haste_health/commands/
hl7v2.rs

1use crate::cli::state::CliState;
2use clap::Subcommand;
3use haste_fhir_client::FHIRClient;
4use haste_fhir_converter::Input;
5use haste_fhir_model::r4::generated::resources::Resource;
6use haste_fhir_model::r4::generated::terminology::{BundleType, IssueType};
7use haste_fhir_operation_error::OperationOutcomeError;
8use haste_hl7v2::mllp::MllpFormatter;
9use std::collections::HashMap;
10use std::io::Write;
11use std::net::TcpListener;
12use std::sync::Arc;
13use tokio::sync::Mutex;
14
15/// Bridge HL7v2 messages to and from the FHIR server.
16#[derive(Subcommand, Debug)]
17pub(crate) enum HL7v2Commands {
18    /// Listen for MLLP-framed HL7v2 messages, convert them to FHIR, and submit them.
19    Receiver {
20        /// Address to bind the MLLP listener to.
21        #[arg(short, long)]
22        address: String,
23        /// Port to bind the MLLP listener to.
24        #[arg(short, long)]
25        port: u16,
26        /// Entry template file name (resolved within --template-dir) used to convert
27        /// incoming HL7v2 messages to FHIR.
28        #[arg(short, long)]
29        main: String,
30        /// Directory containing the conversion templates.
31        #[arg(short, long)]
32        template_dir: String,
33    },
34    /// Send HL7v2 messages over MLLP. Not yet implemented.
35    Sender {
36        /// Address of the MLLP receiver to send to.
37        #[arg(short, long)]
38        address: String,
39        /// Port of the MLLP receiver to send to.
40        #[arg(short, long)]
41        port: u16,
42    },
43}
44
45/// Runs the `hl7v2` command group.
46pub(crate) async fn run(
47    state: Arc<Mutex<CliState>>,
48    command: &HL7v2Commands,
49) -> Result<(), OperationOutcomeError> {
50    let fhir_client = crate::cli::client::fhir_client(state).await?;
51
52    match command {
53        HL7v2Commands::Receiver {
54            address,
55            port,
56            main,
57            template_dir,
58        } => {
59            let listener = TcpListener::bind(format!("{}:{}", address, port)).unwrap();
60
61            let environment = haste_fhir_converter::create_environment(Some(template_dir));
62
63            let template = environment
64                .get_template(&main)
65                .map_err(|e| OperationOutcomeError::error(IssueType::exception(), e.to_string()))?;
66
67            for stream in listener.incoming() {
68                let mut stream = match stream {
69                    Ok(s) => s,
70                    Err(e) => {
71                        eprintln!("Failed to accept connection: {}", e);
72                        continue;
73                    }
74                };
75
76                loop {
77                    let frame = match MllpFormatter::read_frame(&mut stream) {
78                        Ok(f) => f,
79                        Err(e) => {
80                            eprintln!("Connection ended: {}", e);
81                            break;
82                        }
83                    };
84
85                    let start = std::time::Instant::now();
86
87                    let hl7v2_bytes = match MllpFormatter::decode(frame.as_slice()) {
88                        Ok(b) => b.to_vec(),
89                        Err(e) => {
90                            eprintln!("Failed to decode MLLP frame: {}", e);
91                            let _ = stream.write_all(&MllpFormatter::nak());
92                            continue;
93                        }
94                    };
95
96                    let hl7v2_string = String::from_utf8_lossy(&hl7v2_bytes).to_string();
97
98                    let hl7v2 = haste_fhir_converter::convert_input(Input::HL7V2(hl7v2_string))?;
99
100                    let mut ctx = HashMap::new();
101                    ctx.insert("hl7v2", hl7v2);
102
103                    let haste_fhir_converter::Output::FHIR(resource) =
104                        haste_fhir_converter::transform(
105                            &template,
106                            ctx,
107                            &haste_fhir_converter::OutputFormat::FHIR,
108                        )?
109                    else {
110                        eprintln!("Unexpected output format from template");
111                        let _ = stream.write_all(&MllpFormatter::nak());
112                        continue;
113                    };
114
115                    tracing::info!("total transformation: {:?}", start.elapsed());
116
117                    match *resource {
118                        Resource::Bundle(bundle) => match &bundle.type_ {
119                            b if b == &BundleType::batch() => {
120                                match fhir_client.batch((), bundle).await {
121                                    Ok(_) => {
122                                        if let Err(e) = stream.write_all(&MllpFormatter::ack()) {
123                                            eprintln!("Failed to send ACK: {}", e);
124                                            break;
125                                        }
126                                    }
127                                    Err(e) => {
128                                        eprintln!("Failed to send batch {}", e);
129                                        if let Err(e) = stream.write_all(&MllpFormatter::nak()) {
130                                            eprintln!("Failed to send NAK: {}", e);
131                                            break;
132                                        }
133                                    }
134                                }
135                            }
136                            b if b == &BundleType::transaction() => {
137                                match fhir_client.transaction((), bundle).await {
138                                    Ok(_) => {
139                                        if let Err(e) = stream.write_all(&MllpFormatter::ack()) {
140                                            eprintln!("Failed to send ACK: {}", e);
141                                            break;
142                                        }
143                                    }
144                                    Err(e) => {
145                                        eprintln!("Failed to send transaction {}", e);
146                                        if let Err(e) = stream.write_all(&MllpFormatter::nak()) {
147                                            eprintln!("Failed to send NAK: {}", e);
148                                            break;
149                                        }
150                                    }
151                                }
152                            }
153                            _ => {
154                                eprintln!("Unsupported Bundle type: {:?}", bundle.type_);
155                                let _ = stream.write_all(&MllpFormatter::nak());
156                                continue;
157                            }
158                        },
159                        _ => {
160                            let resource_type = resource.resource_type();
161                            match fhir_client.create((), resource_type, *resource).await {
162                                Ok(_) => {
163                                    if let Err(e) = stream.write_all(&MllpFormatter::ack()) {
164                                        eprintln!("Failed to send ACK: {}", e);
165                                        break;
166                                    }
167                                }
168                                Err(e) => {
169                                    eprintln!("Failed to send resource {}", e);
170                                    if let Err(e) = stream.write_all(&MllpFormatter::nak()) {
171                                        eprintln!("Failed to send NAK: {}", e);
172                                        break;
173                                    }
174                                }
175                            }
176                        }
177                    }
178                }
179            }
180
181            Ok(())
182        }
183        HL7v2Commands::Sender {
184            address: _,
185            port: _,
186        } => {
187            todo!("HL7v2 sender not implemented yet");
188        }
189    }
190}
haste_health/commands/hl7v2.rs

haste_health/commands/
hl7v2.rs
