Why Apache Camel for HL7?
Apache Camel provides:
- Built-in HL7 support through the
camel-hl7component - MLLP (Minimal Lower Layer Protocol) transport
- Message routing and transformation
- Integration patterns for healthcare workflows
- Error handling and retry mechanisms
MLLP Protocol
MLLP is the standard transport protocol for HL7 V2 messages. Unlike HTTP or other modern protocols, MLLP uses a simple framing mechanism with special control characters to mark message boundaries over raw TCP connections. Understanding these framing bytes is essential for debugging connection issues. Most healthcare facilities still rely on MLLP for real-time message exchange, so proper implementation is critical for interoperability.
Format: <VT>message<FS><CR>
- <VT>: Vertical Tab (0x0B) - Start byte
- message: HL7 V2 message content
- <FS>: File Separator (0x1C) - End byte
- <CR>: Carriage Return (0x0D) - Terminating byteBasic Camel HL7 Route
Apache Camel routes define the flow of messages through your integration. The simplest HL7 route listens on a TCP port for incoming MLLP messages and processes them. The HL7MLLPCodec handles framing and unframing automatically. Routes are defined using Camel’s fluent DSL, making the integration logic readable and maintainable. This basic example logs incoming messages, but real implementations would add parsing, validation, and business logic.
import org.apache.camel.CamelContext;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.impl.DefaultCamelContext;
public class BasicHL7Route {
public static void main(String[] args) throws Exception {
CamelContext context = new DefaultCamelContext();
context.addRoutes(new RouteBuilder() {
@Override
public void configure() throws Exception {
// Listen for HL7 messages on port 8888
from("mina:tcp://localhost:8888?sync=true&codec=#hl7codec")
.log("Received HL7 message: ${body}")
.to("log:hl7-messages");
}
});
// Register HL7 codec
ca.uhn.hl7v2.parser.PipeParser parser = new ca.uhn.hl7v2.parser.PipeParser();
org.apache.camel.component.hl7.HL7MLLPCodec codec =
new org.apache.camel.component.hl7.HL7MLLPCodec();
codec.setCharset("UTF-8");
context.getRegistry().bind("hl7codec", codec);
context.start();
System.out.println("HL7 server started on port 8888");
System.out.println("Press Enter to stop...");
System.in.read();
context.stop();
}
}Complete HL7 Server with Camel
A production HL7 server must handle errors gracefully, generate appropriate acknowledgments, and process different message types. This example demonstrates a complete server implementation with exception handling that sends negative acknowledgments (NACK) on errors. The server parses incoming messages, processes them based on type, and returns positive acknowledgments (ACK) with the AA code. Note the importance of copying the original message control ID into the acknowledgment for proper correlation.
import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.impl.DefaultCamelContext;
import org.apache.camel.component.hl7.HL7MLLPCodec;
import ca.uhn.hl7v2.model.Message;
import ca.uhn.hl7v2.model.v25.message.ADT_A01;
import ca.uhn.hl7v2.model.v25.message.ACK;
import ca.uhn.hl7v2.model.v25.segment.MSA;
import ca.uhn.hl7v2.parser.Parser;
import ca.uhn.hl7v2.DefaultHapiContext;
import ca.uhn.hl7v2.HapiContext;
public class HL7Server {
public static void main(String[] args) throws Exception {
CamelContext context = new DefaultCamelContext();
// Register HL7 codec
HL7MLLPCodec codec = new HL7MLLPCodec();
codec.setCharset("UTF-8");
context.getRegistry().bind("hl7codec", codec);
context.addRoutes(new RouteBuilder() {
@Override
public void configure() throws Exception {
// Error handler
onException(Exception.class)
.log("Error processing HL7 message: ${exception.message}")
.handled(true)
.process(new Processor() {
@Override
public void process(Exchange exchange) throws Exception {
// Send NACK (negative acknowledgment)
exchange.getIn().setBody(createNACK(exchange));
}
});
// Main HL7 processing route
from("mina:tcp://localhost:8888?sync=true&codec=#hl7codec")
.log("Received message: ${body}")
.process(new HL7MessageProcessor())
.log("Processed successfully")
.process(new Processor() {
@Override
public void process(Exchange exchange) throws Exception {
// Create ACK
Message message = exchange.getIn().getBody(Message.class);
ACK ack = createACK(message);
HapiContext hapiContext = new DefaultHapiContext();
Parser parser = hapiContext.getPipeParser();
String ackString = parser.encode(ack);
hapiContext.close();
exchange.getMessage().setBody(ackString);
}
})
.log("Sending ACK: ${body}");
}
});
context.start();
System.out.println("HL7 Server started on port 8888");
System.out.println("Waiting for HL7 messages...");
System.out.println("Press Enter to stop.");
System.in.read();
context.stop();
}
private static ACK createACK(Message originalMessage) throws Exception {
HapiContext context = new DefaultHapiContext();
ACK ack = new ACK();
ack.initQuickstart("ACK", "A01", "P");
// Copy message control ID from original message
String messageControlId = originalMessage.get("MSH-10").encode();
ack.getMSH().getMessageControlID().setValue(messageControlId);
// Set acknowledgment code
MSA msa = ack.getMSA();
msa.getAcknowledgmentCode().setValue("AA"); // Application Accept
msa.getMessageControlID().setValue(messageControlId);
context.close();
return ack;
}
private static String createNACK(Exchange exchange) throws Exception {
HapiContext context = new DefaultHapiContext();
ACK nack = new ACK();
nack.initQuickstart("ACK", "A01", "P");
MSA msa = nack.getMSA();
msa.getAcknowledgmentCode().setValue("AE"); // Application Error
Parser parser = context.getPipeParser();
String nackString = parser.encode(nack);
context.close();
return nackString;
}
// Inner class for processing
static class HL7MessageProcessor implements Processor {
@Override
public void process(Exchange exchange) throws Exception {
String hl7String = exchange.getIn().getBody(String.class);
HapiContext context = new DefaultHapiContext();
Parser parser = context.getGenericParser();
Message message = parser.parse(hl7String);
System.out.println("\n=== Processing Message ===");
System.out.println("Message Type: " + message.getName());
// Process based on message type
if (message instanceof ADT_A01) {
processADT((ADT_A01) message);
}
exchange.getIn().setBody(message);
context.close();
}
private void processADT(ADT_A01 adtMessage) throws Exception {
String patientId = adtMessage.getPID()
.getPatientIdentifierList(0).getIDNumber().getValue();
String patientName = adtMessage.getPID()
.getPatientName(0).getFamilyName().getSurname().getValue() + ", " +
adtMessage.getPID().getPatientName(0).getGivenName().getValue();
System.out.println("Patient: " + patientName + " (ID: " + patientId + ")");
System.out.println("Event: Patient admission");
}
}
}HL7 Client with Camel
HL7 clients initiate connections to remote servers and send messages. Camel simplifies client implementation with its producer template pattern. The client creates a message, sends it through a route, and receives the acknowledgment response synchronously. This example shows how to build an ADT message and transmit it to a server. The sync=true parameter ensures the client waits for and receives the ACK response.
import org.apache.camel.CamelContext;
import org.apache.camel.ProducerTemplate;
import org.apache.camel.impl.DefaultCamelContext;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.hl7.HL7MLLPCodec;
import ca.uhn.hl7v2.DefaultHapiContext;
import ca.uhn.hl7v2.HapiContext;
import ca.uhn.hl7v2.model.v25.message.ADT_A01;
import ca.uhn.hl7v2.model.v25.segment.MSH;
import ca.uhn.hl7v2.model.v25.segment.PID;
import ca.uhn.hl7v2.parser.Parser;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
public class HL7Client {
public static void main(String[] args) throws Exception {
CamelContext context = new DefaultCamelContext();
// Register HL7 codec
HL7MLLPCodec codec = new HL7MLLPCodec();
codec.setCharset("UTF-8");
context.getRegistry().bind("hl7codec", codec);
context.addRoutes(new RouteBuilder() {
@Override
public void configure() throws Exception {
// Route for sending HL7 messages
from("direct:sendHL7")
.log("Sending HL7 message")
.to("mina:tcp://localhost:8888?sync=true&codec=#hl7codec")
.log("Received response: ${body}");
}
});
context.start();
// Create and send HL7 message
String hl7Message = createADTMessage();
ProducerTemplate template = context.createProducerTemplate();
String response = template.requestBody("direct:sendHL7", hl7Message, String.class);
System.out.println("\nResponse received:");
System.out.println(response);
template.stop();
context.stop();
}
private static String createADTMessage() throws Exception {
HapiContext context = new DefaultHapiContext();
ADT_A01 adtMessage = new ADT_A01();
adtMessage.initQuickstart("ADT", "A01", "P");
// MSH
MSH msh = adtMessage.getMSH();
msh.getSendingApplication().getNamespaceID().setValue("CLIENT_APP");
msh.getSendingFacility().getNamespaceID().setValue("CLIENT_FACILITY");
msh.getReceivingApplication().getNamespaceID().setValue("SERVER_APP");
msh.getReceivingFacility().getNamespaceID().setValue("SERVER_FACILITY");
String timestamp = LocalDateTime.now()
.format(DateTimeFormatter.ofPattern("yyyyMMddHHmmss"));
msh.getDateTimeOfMessage().getTime().setValue(timestamp);
msh.getMessageControlID().setValue("MSG" + System.currentTimeMillis());
// PID
PID pid = adtMessage.getPID();
pid.getPatientIdentifierList(0).getIDNumber().setValue("123456");
pid.getPatientIdentifierList(0).getAssigningAuthority()
.getNamespaceID().setValue("HOSPITAL");
pid.getPatientIdentifierList(0).getIdentifierTypeCode().setValue("MR");
pid.getPatientName(0).getFamilyName().getSurname().setValue("DOE");
pid.getPatientName(0).getGivenName().setValue("JOHN");
pid.getDateTimeOfBirth().getTime().setValue("19800115");
pid.getAdministrativeSex().setValue("M");
Parser parser = context.getPipeParser();
String encodedMessage = parser.encode(adtMessage);
context.close();
return encodedMessage;
}
}Message Routing and Transformation
Real healthcare integrations often need to route messages to different destinations based on their type or content. Camel’s content-based router pattern enables this using the choice() DSL. Messages can be parsed, enriched with headers indicating their type, and then directed to appropriate processing routes. This pattern separates routing logic from processing logic, making the integration easier to maintain and extend.
import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.impl.DefaultCamelContext;
import org.apache.camel.component.hl7.HL7MLLPCodec;
import ca.uhn.hl7v2.model.Message;
import ca.uhn.hl7v2.model.v25.message.ADT_A01;
import ca.uhn.hl7v2.model.v25.message.ORU_R01;
import ca.uhn.hl7v2.parser.Parser;
import ca.uhn.hl7v2.DefaultHapiContext;
import ca.uhn.hl7v2.HapiContext;
import ca.uhn.hl7v2.util.Terser;
public class HL7RoutingExample {
public static void main(String[] args) throws Exception {
CamelContext context = new DefaultCamelContext();
HL7MLLPCodec codec = new HL7MLLPCodec();
codec.setCharset("UTF-8");
context.getRegistry().bind("hl7codec", codec);
context.addRoutes(new RouteBuilder() {
@Override
public void configure() throws Exception {
// Main receiving endpoint
from("mina:tcp://localhost:8888?sync=true&codec=#hl7codec")
.log("Received message on main port")
.process(new Processor() {
@Override
public void process(Exchange exchange) throws Exception {
String hl7String = exchange.getIn().getBody(String.class);
HapiContext hapiContext = new DefaultHapiContext();
Parser parser = hapiContext.getGenericParser();
Message message = parser.parse(hl7String);
Terser terser = new Terser(message);
String messageType = terser.get("/.MSH-9-1");
String triggerEvent = terser.get("/.MSH-9-2");
exchange.getIn().setHeader("messageType", messageType);
exchange.getIn().setHeader("triggerEvent", triggerEvent);
exchange.getIn().setBody(message);
hapiContext.close();
}
})
.choice()
// Route ADT messages
.when(header("messageType").isEqualTo("ADT"))
.log("Routing ADT message: ${header.triggerEvent}")
.to("direct:processADT")
// Route ORU (lab results)
.when(header("messageType").isEqualTo("ORU"))
.log("Routing ORU message: ${header.triggerEvent}")
.to("direct:processORU")
// Route ORM (orders)
.when(header("messageType").isEqualTo("ORM"))
.log("Routing ORM message: ${header.triggerEvent}")
.to("direct:processORM")
// Unknown message types
.otherwise()
.log("Unknown message type: ${header.messageType}")
.to("direct:processUnknown")
.end();
// ADT processing route
from("direct:processADT")
.log("Processing ADT message")
.process(new Processor() {
@Override
public void process(Exchange exchange) throws Exception {
Message message = exchange.getIn().getBody(Message.class);
if (message instanceof ADT_A01) {
ADT_A01 adt = (ADT_A01) message;
String patientId = adt.getPID()
.getPatientIdentifierList(0).getIDNumber().getValue();
System.out.println("ADT - Patient ID: " + patientId);
}
}
})
.to("file:output/adt?fileName=${date:now:yyyyMMdd-HHmmss}.hl7");
// ORU processing route
from("direct:processORU")
.log("Processing ORU lab results")
.process(new Processor() {
@Override
public void process(Exchange exchange) throws Exception {
Message message = exchange.getIn().getBody(Message.class);
if (message instanceof ORU_R01) {
ORU_R01 oru = (ORU_R01) message;
System.out.println("Lab results received");
}
}
})
.to("file:output/oru?fileName=${date:now:yyyyMMdd-HHmmss}.hl7");
// ORM processing route
from("direct:processORM")
.log("Processing ORM order")
.to("file:output/orm?fileName=${date:now:yyyyMMdd-HHmmss}.hl7");
// Unknown message handling
from("direct:processUnknown")
.log("Handling unknown message type")
.to("file:output/unknown?fileName=${date:now:yyyyMMdd-HHmmss}.hl7");
}
});
context.start();
System.out.println("HL7 Router started on port 8888");
System.out.println("Messages will be routed based on type");
System.out.println("Press Enter to stop.");
System.in.read();
context.stop();
}
}