haste_health/commands/
hl7v2.rs

1use crate::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#[derive(Subcommand)]
16pub(crate) enum HL7v2Commands {
17    Receiver {
18        #[arg(short, long)]
19        address: String,
20        #[arg(short, long)]
21        port: u16,
22        #[arg(short, long)]
23        main: String,
24        #[arg(short, long)]
25        template_dir: String,
26    },
27    Sender {
28        #[arg(short, long)]
29        address: String,
30        #[arg(short, long)]
31        port: u16,
32    },
33}
34
35pub(crate) async fn hl7v2(
36    state: Arc<Mutex<CLIState>>,
37    command: &HL7v2Commands,
38) -> Result<(), OperationOutcomeError> {
39    let fhir_client = crate::client::fhir_client(state).await?;
40
41    match command {
42        HL7v2Commands::Receiver {
43            address,
44            port,
45            main,
46            template_dir,
47        } => {
48            let listener = TcpListener::bind(format!("{}:{}", address, port)).unwrap();
49
50            let environment = haste_fhir_converter::create_environment(Some(template_dir));
51
52            let template = environment
53                .get_template(&main)
54                .map_err(|e| OperationOutcomeError::error(IssueType::exception(), e.to_string()))?;
55
56            for stream in listener.incoming() {
57                let mut stream = match stream {
58                    Ok(s) => s,
59                    Err(e) => {
60                        eprintln!("Failed to accept connection: {}", e);
61                        continue;
62                    }
63                };
64
65                loop {
66                    let frame = match MllpFormatter::read_frame(&mut stream) {
67                        Ok(f) => f,
68                        Err(e) => {
69                            eprintln!("Connection ended: {}", e);
70                            break;
71                        }
72                    };
73
74                    let start = std::time::Instant::now();
75
76                    let hl7v2_bytes = match MllpFormatter::decode(frame.as_slice()) {
77                        Ok(b) => b.to_vec(),
78                        Err(e) => {
79                            eprintln!("Failed to decode MLLP frame: {}", e);
80                            let _ = stream.write_all(&MllpFormatter::nak());
81                            continue;
82                        }
83                    };
84
85                    let hl7v2_string = String::from_utf8_lossy(&hl7v2_bytes).to_string();
86
87                    let hl7v2 = haste_fhir_converter::convert_input(Input::HL7V2(hl7v2_string))?;
88
89                    let mut ctx = HashMap::new();
90                    ctx.insert("hl7v2", hl7v2);
91
92                    let haste_fhir_converter::Output::FHIR(resource) =
93                        haste_fhir_converter::transform(
94                            &template,
95                            ctx,
96                            &haste_fhir_converter::OutputFormat::FHIR,
97                        )?
98                    else {
99                        eprintln!("Unexpected output format from template");
100                        let _ = stream.write_all(&MllpFormatter::nak());
101                        continue;
102                    };
103
104                    tracing::info!("total transformation: {:?}", start.elapsed());
105
106                    match resource {
107                        Resource::Bundle(bundle) => match &bundle.type_ {
108                            b if b == &BundleType::batch() => {
109                                match fhir_client.batch((), bundle).await {
110                                    Ok(_) => {
111                                        if let Err(e) = stream.write_all(&MllpFormatter::ack()) {
112                                            eprintln!("Failed to send ACK: {}", e);
113                                            break;
114                                        }
115                                    }
116                                    Err(e) => {
117                                        eprintln!("Failed to send batch {}", e);
118                                        if let Err(e) = stream.write_all(&MllpFormatter::nak()) {
119                                            eprintln!("Failed to send NAK: {}", e);
120                                            break;
121                                        }
122                                    }
123                                }
124                            }
125                            b if b == &BundleType::transaction() => {
126                                match fhir_client.transaction((), bundle).await {
127                                    Ok(_) => {
128                                        if let Err(e) = stream.write_all(&MllpFormatter::ack()) {
129                                            eprintln!("Failed to send ACK: {}", e);
130                                            break;
131                                        }
132                                    }
133                                    Err(e) => {
134                                        eprintln!("Failed to send transaction {}", e);
135                                        if let Err(e) = stream.write_all(&MllpFormatter::nak()) {
136                                            eprintln!("Failed to send NAK: {}", e);
137                                            break;
138                                        }
139                                    }
140                                }
141                            }
142                            _ => {
143                                eprintln!("Unsupported Bundle type: {:?}", bundle.type_);
144                                let _ = stream.write_all(&MllpFormatter::nak());
145                                continue;
146                            }
147                        },
148                        _ => {
149                            let resource_type = resource.resource_type();
150                            match fhir_client.create((), resource_type, resource).await {
151                                Ok(_) => {
152                                    if let Err(e) = stream.write_all(&MllpFormatter::ack()) {
153                                        eprintln!("Failed to send ACK: {}", e);
154                                        break;
155                                    }
156                                }
157                                Err(e) => {
158                                    eprintln!("Failed to send resource {}", e);
159                                    if let Err(e) = stream.write_all(&MllpFormatter::nak()) {
160                                        eprintln!("Failed to send NAK: {}", e);
161                                        break;
162                                    }
163                                }
164                            }
165                        }
166                    }
167                }
168            }
169
170            Ok(())
171        }
172        HL7v2Commands::Sender {
173            address: _,
174            port: _,
175        } => {
176            todo!("HL7v2 sender not implemented yet");
177        }
178    }
179}
haste_health/commands/hl7v2.rs

haste_health/commands/
hl7v2.rs
