Apache Camel Integration Patterns

Setting Up Apache Camel with FHIR

Apache Camel provides enterprise integration patterns that simplify building healthcare data pipelines. The camel-fhir component wraps HAPI FHIR client operations, enabling declarative routes for CRUD operations, transformations, and error handling. Use Camel when you need to integrate FHIR with other systems, process files, implement message queuing, or build complex healthcare workflows.

To begin using Apache Camel with FHIR, you’ll need to add several dependencies to your project. The camel-spring-boot-starter provides the core Camel framework when using Spring Boot, while camel-fhir gives you the FHIR-specific integration components. Additional utilities like camel-jackson for JSON processing and camel-http for HTTP communications complete the required dependencies.

<!-- Additional Maven dependencies for Camel -->
<dependencies>
    <!-- Apache Camel Spring Boot -->
    <dependency>
        <groupId>org.apache.camel.springboot</groupId>
        <artifactId>camel-spring-boot-starter</artifactId>
        <version>${apache.camel.version}</version>
    </dependency>

    <!-- Camel FHIR Component -->
    <dependency>
        <groupId>org.apache.camel</groupId>
        <artifactId>camel-fhir</artifactId>
        <version>${apache.camel.version}</version>
    </dependency>

    <!-- Camel Jackson for JSON -->
    <dependency>
        <groupId>org.apache.camel</groupId>
        <artifactId>camel-jackson</artifactId>
        <version>${apache.camel.version}</version>
    </dependency>

    <!-- Camel HTTP -->
    <dependency>
        <groupId>org.apache.camel</groupId>
        <artifactId>camel-http</artifactId>
        <version>${apache.camel.version}</version>
    </dependency>
</dependencies>

Basic Camel FHIR Routes

Create routes for CRUD operations on FHIR resources. Camel routes define the flow of messages through processing steps. The FHIR component uses URIs like fhir://create/resource to specify operations. Routes can be triggered by direct endpoints for synchronous calls, timers for polling, or file watchers for batch processing. The fluent DSL makes complex integrations readable and maintainable.

import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.fhir.FhirConfiguration;
import org.apache.camel.component.fhir.api.ExtraParameters;
import org.hl7.fhir.r4.model.Patient;
import org.hl7.fhir.r4.model.Bundle;

public class BasicFhirRoutes extends RouteBuilder {

    @Override
    public void configure() throws Exception {

        // Configure FHIR component
        FhirConfiguration fhirConfig = new FhirConfiguration();
        fhirConfig.setServerUrl("http://hapi.fhir.org/baseR4");
        fhirConfig.setFhirVersion("R4");

        // Route 1: Create Patient
        from("direct:createPatient")
            .log("Creating patient")
            .to("fhir://create/resource?" +
                "inBody=resourceAsString&" +
                "serverUrl=http://hapi.fhir.org/baseR4&" +
                "fhirVersion=R4")
            .log("Patient created: ${body}");

        // Route 2: Read Patient by ID
        from("direct:readPatient")
            .log("Reading patient: ${header.patientId}")
            .toD("fhir://read/resourceById?" +
                "inBody=id&" +
                "resourceClass=org.hl7.fhir.r4.model.Patient&" +
                "serverUrl=http://hapi.fhir.org/baseR4&" +
                "fhirVersion=R4")
            .log("Patient retrieved: ${body}");

        // Route 3: Search Patients
        from("direct:searchPatients")
            .log("Searching for patients")
            .to("fhir://search/searchByUrl?" +
                "inBody=url&" +
                "serverUrl=http://hapi.fhir.org/baseR4&" +
                "fhirVersion=R4")
            .log("Search results: ${body}");

        // Route 4: Update Patient
        from("direct:updatePatient")
            .log("Updating patient")
            .to("fhir://update/resource?" +
                "inBody=resourceAsString&" +
                "serverUrl=http://hapi.fhir.org/baseR4&" +
                "fhirVersion=R4")
            .log("Patient updated: ${body}");

        // Route 5: Delete Patient
        from("direct:deletePatient")
            .log("Deleting patient: ${header.patientId}")
            .toD("fhir://delete/resourceById?" +
                "inBody=id&" +
                "resourceClass=org.hl7.fhir.r4.model.Patient&" +
                "serverUrl=http://hapi.fhir.org/baseR4&" +
                "fhirVersion=R4")
            .log("Patient deleted");
    }
}

File-to-FHIR Integration

Process FHIR resources from files and upload to a server. File-based integration is common for batch imports from legacy systems or partner organizations. Camel’s file component monitors directories for new files, processes them through your route, and handles completion (moving or deleting files). Add validation before creating resources to catch errors early and route invalid files to error directories.

import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
import ca.uhn.fhir.context.FhirContext;
import ca.uhn.fhir.parser.IParser;

public class FileToFhirRoute extends RouteBuilder {

    private final FhirContext fhirContext = FhirContext.forR4();

    @Override
    public void configure() throws Exception {

        // Monitor directory for JSON patient files
        from("file:input/patients?noop=true&include=.*\\.json")
            .log("Processing file: ${header.CamelFileName}")
            .process(new Processor() {
                @Override
                public void process(Exchange exchange) throws Exception {
                    String jsonContent = exchange.getIn().getBody(String.class);

                    // Parse JSON to Patient
                    IParser parser = fhirContext.newJsonParser();
                    Patient patient = parser.parseResource(Patient.class, jsonContent);

                    // Validate patient has required fields
                    if (!patient.hasName()) {
                        throw new IllegalArgumentException("Patient must have a name");
                    }

                    exchange.getIn().setBody(jsonContent);
                }
            })
            .to("fhir://create/resource?" +
                "inBody=resourceAsString&" +
                "serverUrl=http://hapi.fhir.org/baseR4&" +
                "fhirVersion=R4")
            .log("Patient created with ID: ${body.id}")
            .to("file:output/success")
            .onException(Exception.class)
                .log("Error processing file: ${exception.message}")
                .to("file:output/error")
            .end();
    }
}

HL7 V2 to FHIR Transformation

Transform HL7 V2 messages to FHIR resources. Many healthcare systems still use HL7 V2 for real-time messaging. This integration pattern receives V2 messages via MLLP (Minimal Lower Layer Protocol), transforms them to FHIR resources, and posts them to a FHIR server. The transformation requires mapping V2 segments (PID, PV1, OBX) to corresponding FHIR resources. Consider using established mapping libraries for production systems.

import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.Exchange;
import ca.uhn.hl7v2.model.v24.message.ADT_A01;
import ca.uhn.hl7v2.model.v24.segment.PID;
import org.hl7.fhir.r4.model.*;

public class Hl7ToFhirRoute extends RouteBuilder {

    private final FhirContext fhirContext = FhirContext.forR4();

    @Override
    public void configure() throws Exception {

        from("mllp://0.0.0.0:8888")
            .log("Received HL7 message")
            .unmarshal().hl7()
            .process(exchange -> {
                ADT_A01 adt = (ADT_A01) exchange.getIn().getBody();
                PID pid = adt.getPID();

                // Transform to FHIR Patient
                Patient patient = new Patient();

                // Extract patient ID
                String patientId = pid.getPatientID().getIDNumber().getValue();
                patient.addIdentifier()
                    .setSystem("http://hospital.org/mrn")
                    .setValue(patientId);

                // Extract name
                if (pid.getPatientName().length > 0) {
                    patient.addName()
                        .setFamily(pid.getPatientName(0).getFamilyName().getSurname().getValue())
                        .addGiven(pid.getPatientName(0).getGivenName().getValue());
                }

                // Extract birth date
                String birthDate = pid.getDateTimeOfBirth().getTimeOfAnEvent().getValue();
                if (birthDate != null && !birthDate.isEmpty()) {
                    // Parse HL7 date format (YYYYMMDD)
                    patient.setBirthDateElement(new DateType(
                        birthDate.substring(0, 4) + "-" +
                        birthDate.substring(4, 6) + "-" +
                        birthDate.substring(6, 8)));
                }

                // Extract gender
                String gender = pid.getAdministrativeSex().getValue();
                if ("M".equals(gender)) {
                    patient.setGender(Enumerations.AdministrativeGender.MALE);
                } else if ("F".equals(gender)) {
                    patient.setGender(Enumerations.AdministrativeGender.FEMALE);
                }

                // Convert to JSON
                IParser parser = fhirContext.newJsonParser();
                String jsonPatient = parser.encodeResourceToString(patient);

                exchange.getIn().setBody(jsonPatient);
            })
            .to("fhir://create/resource?" +
                "inBody=resourceAsString&" +
                "serverUrl=http://hapi.fhir.org/baseR4&" +
                "fhirVersion=R4")
            .log("Created FHIR Patient from HL7 message: ${body.id}")
            .marshal().hl7()
            .transform(constant("MSA|AA|${id}"));
    }
}

Polling FHIR Server for Changes

Monitor a FHIR server for new or updated resources. When servers do not support subscriptions, polling provides an alternative for detecting changes. Use the _lastUpdated parameter to fetch only resources modified since the last poll. Maintain state (last poll time) to avoid reprocessing. Consider the trade-off between polling frequency and server load. For real-time needs, prefer FHIR subscriptions when available.

import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.fhir.api.ExtraParameters;
import java.util.HashMap;
import java.util.Map;

public class FhirPollingRoute extends RouteBuilder {

    @Override
    public void configure() throws Exception {

        // Poll for new observations every 30 seconds
        from("timer://pollObservations?period=30000")
            .setBody(constant("Observation?_lastUpdated=gt{{lastPollTime}}&_sort=-_lastUpdated"))
            .to("fhir://search/searchByUrl?" +
                "inBody=url&" +
                "serverUrl=http://hapi.fhir.org/baseR4&" +
                "fhirVersion=R4")
            .process(exchange -> {
                Bundle bundle = exchange.getIn().getBody(Bundle.class);

                if (bundle.hasEntry()) {
                    log.info("Found " + bundle.getEntry().size() + " new observations");

                    for (Bundle.BundleEntryComponent entry : bundle.getEntry()) {
                        Observation obs = (Observation) entry.getResource();
                        log.info("Processing observation: " + obs.getId());

                        // Process the observation
                        // ...
                    }

                    // Update last poll time
                    String lastUpdated = bundle.getMeta().getLastUpdated().toString();
                    exchange.setProperty("lastPollTime", lastUpdated);
                }
            })
            .choice()
                .when(simple("${body.entry.size} > 0"))
                    .to("direct:processNewObservations")
            .end();
    }
}

Content-Based Routing

Route messages based on their content. Content-based routing examines incoming messages and directs them to different processors based on their characteristics. For FHIR resources, route by resource type, observation category, or clinical values. This pattern enables specialized processing: vital signs to alerting systems, lab results to analytics, imaging studies to PACS integration. Use JSONPath or processor-based routing for complex conditions.

public class ContentBasedFhirRoute extends RouteBuilder {

    @Override
    public void configure() throws Exception {

        from("direct:routeObservation")
            .choice()
                // Route based on observation category
                .when().jsonpath("$.category[0].coding[0].code == 'vital-signs'")
                    .log("Routing to vital signs processor")
                    .to("direct:processVitalSigns")
                .when().jsonpath("$.category[0].coding[0].code == 'laboratory'")
                    .log("Routing to lab results processor")
                    .to("direct:processLabResults")
                .when().jsonpath("$.category[0].coding[0].code == 'imaging'")
                    .log("Routing to imaging processor")
                    .to("direct:processImaging")
                .otherwise()
                    .log("Unknown observation category")
                    .to("direct:processOther")
            .end();

        // Vital signs processing
        from("direct:processVitalSigns")
            .process(exchange -> {
                Observation obs = exchange.getIn().getBody(Observation.class);
                log.info("Processing vital signs: " +
                    obs.getCode().getCodingFirstRep().getDisplay());

                // Check for critical values
                if (obs.hasValueQuantity()) {
                    String code = obs.getCode().getCodingFirstRep().getCode();

                    // Example: Check blood pressure
                    if ("85354-9".equals(code)) {  // Blood pressure
                        boolean isCritical = checkCriticalBloodPressure(obs);
                        if (isCritical) {
                            exchange.setProperty("alertRequired", true);
                        }
                    }
                }
            })
            .choice()
                .when(exchangeProperty("alertRequired").isEqualTo(true))
                    .to("direct:sendAlert")
            .end();

        // Lab results processing
        from("direct:processLabResults")
            .log("Processing lab result")
            .process(exchange -> {
                Observation obs = exchange.getIn().getBody(Observation.class);

                // Check if result is outside reference range
                if (obs.hasInterpretation()) {
                    for (CodeableConcept interp : obs.getInterpretation()) {
                        String interpCode = interp.getCodingFirstRep().getCode();
                        if ("H".equals(interpCode) || "L".equals(interpCode) ||
                            "HH".equals(interpCode) || "LL".equals(interpCode)) {
                            exchange.setProperty("abnormalResult", true);
                            break;
                        }
                    }
                }
            })
            .choice()
                .when(exchangeProperty("abnormalResult").isEqualTo(true))
                    .to("direct:flagForReview")
            .end();
    }

    private boolean checkCriticalBloodPressure(Observation obs) {
        // Check if systolic > 180 or diastolic > 120
        for (Observation.ObservationComponentComponent component : obs.getComponent()) {
            String code = component.getCode().getCodingFirstRep().getCode();
            if ("8480-6".equals(code)) {  // Systolic
                if (component.getValueQuantity().getValue().doubleValue() > 180) {
                    return true;
                }
            } else if ("8462-4".equals(code)) {  // Diastolic
                if (component.getValueQuantity().getValue().doubleValue() > 120) {
                    return true;
                }
            }
        }
        return false;
    }
}

Error Handling and Retry Logic

Implement robust error handling with retries. Healthcare integrations must handle transient failures gracefully. The dead letter channel pattern moves failed messages to an error queue after exhausting retries. Exponential backoff reduces load on struggling services. Configure different retry policies for different error types: retry server errors, but fail fast on validation errors. Always log enough context for troubleshooting.

public class FhirErrorHandlingRoute extends RouteBuilder {

    @Override
    public void configure() throws Exception {

        // Global error handling
        errorHandler(deadLetterChannel("direct:errorHandler")
            .maximumRedeliveries(3)
            .redeliveryDelay(5000)
            .retryAttemptedLogLevel(LoggingLevel.WARN)
            .useExponentialBackOff()
            .backOffMultiplier(2));

        // Main route with error handling
        from("direct:createPatientWithRetry")
            .onException(Exception.class)
                .maximumRedeliveries(3)
                .redeliveryDelay(2000)
                .retryAttemptedLogLevel(LoggingLevel.WARN)
                .handled(true)
                .log("Error creating patient: ${exception.message}")
                .to("direct:errorHandler")
            .end()
            .log("Attempting to create patient")
            .to("fhir://create/resource?" +
                "inBody=resourceAsString&" +
                "serverUrl=http://hapi.fhir.org/baseR4&" +
                "fhirVersion=R4&" +
                "connectTimeout=10000&" +
                "socketTimeout=10000")
            .log("Patient created successfully: ${body.id}");

        // Error handler route
        from("direct:errorHandler")
            .log("Processing error: ${exception.message}")
            .process(exchange -> {
                Exception exception = exchange.getProperty(
                    Exchange.EXCEPTION_CAUGHT, Exception.class);
                String errorMessage = "Failed to process FHIR resource: " +
                    exception.getMessage();

                // Log to error system
                log.error(errorMessage, exception);

                // Could send alert, store in error queue, etc.
            })
            .to("file:output/errors");
    }
}

Integration Patterns Summary