49.10.2.1.1Fixed Interval Polling Trigger

The sample below provides one variation for this route configured to poll an S3 Bucket named "sample-bucket-123" at 60 second intervals and download any files found.

<route id="download_from_s3">
	<!-- Check for incoming messages from S3 Bucket -->
	<from uri="aws2-s3://sample-bucket-123?region=us-east-2&amp;useDefaultCredentialsProvider=true&amp;delay=60000"/>
	<to uri="direct:parseFileAndSendToKafka" />
</route>

49.10.2.1.2External Event Trigger

The sample below provides an alternate implementation for this route configured to instead wait for a message received from an IBM MQ queue before downloading files. In this example, the message from IBM MQ contains the name of a sub-folder in the S3 Bucket and all files contained in the sub-folder will be downloaded and processed.

49.10.2.2.1Splitting HL7 v2 File into Message Objects

The following Camel processor class parses text files containing HL7 v2 messages into HL7 v2 message objects using the same library as the Smile CDR CLI hl7v2-transmit-flatfile command.

_/*-
 * #%L
 * Smile CDR - CDR
 * %%
 * Copyright (C) 2016 - 2025 Smile CDR, Inc.
 * %%
 * All rights reserved.
 * #L%
 */
package com.example.camel;

import ca.uhn.hl7v2.HapiContext;
import ca.uhn.hl7v2.util.Hl7InputStreamMessageIterator;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;

import java.io.StringReader;

/**
 * Processor used to convert String text containing HL7v2 messages (typically loaded from a file)
 * into an {@link Hl7InputStreamMessageIterator} class instance. This effectively parses out the HL7v2
 * String into a collection of HL7v2 message objects which can be iterated through for subsequent
 * processing.
 */
public class HL7v2TextParsingProcessor implements Processor {

   HapiContext myHapiContext;

   public HL7v2TextParsingProcessor(HapiContext theHapiContext) {
      myHapiContext = theHapiContext;
   }

   @Override
   public void process(Exchange exchange) throws Exception {

      // Parse the incoming HL7 v2 file into messages using the HAPI HL7 libraries
      org.apache.camel.Message message = exchange.getIn();
      StringReader hl7v2FileContentsReader = new StringReader(message.getBody(String.class));
      Hl7InputStreamMessageIterator messageIterator =
            new Hl7InputStreamMessageIterator(hl7v2FileContentsReader, myHapiContext);

      // Replace the message body with the Iterator object created above.
      message.setBody(messageIterator);
   }
}

49.10.2.2.2Optional: Calculating a Kafka Partition Based on Patient ID Field

The following Camel processor class parses HL7 v2 message objects, extracting the PID segment from the message and then the first patient identifier from the PID segment. It then calculates a hash of the patient identifier and uses the hash to generate a Kafka partition ID between 0 and 4 (this assumes that the hl7v2-in-topic Kafka topic has at least 5 partitions). The Kafka partition ID is then passed along as a header in the Camel exchange and later used by the Kafka component to determine which partition to publish the message to.

_/*-
 * #%L
 * Smile CDR - CDR
 * %%
 * Copyright (C) 2016 - 2025 Smile CDR, Inc.
 * %%
 * All rights reserved.
 * #L%
 */
package com.example.camel;

import ca.uhn.hl7v2.model.v25.segment.PID;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.component.kafka.KafkaConstants;

/**
 * Sample processor used to calculate and set a Kafka partition ID in Camel Message header
 * for an HL7 v2 message based on Patient identifier. When the message is later published to a
 * Kafka topic using a Kafka Camel component, the PARTITION_KEY header will direct the component
 * to publish to the specified Kafka partition ID in the topic.
 */
public class HL7v2PartitionCalculateProcessor implements Processor {

   @Override
   public void process(Exchange exchange) throws Exception {

      // Parse the incoming message and extract PID segment
      org.apache.camel.Message message = exchange.getIn();
      ca.uhn.hl7v2.model.Message hapiMsg = message.getBody(ca.uhn.hl7v2.model.Message.class);

      PID pid = (PID) hapiMsg.get("PID");

      // Retrieve the first Patient Identifier and generate a partition ID between 0-4
      // based on a hash of the Identifier value.
      String patientId = pid.getPatientIdentifierList(0).getIDNumber().getValue();
      Integer partition = Math.abs(patientId.hashCode() % 5);

      // Add partition key to Kafka message header
      message.getHeaders().put(KafkaConstants.PARTITION_KEY, partition);
   }
}

49.10.2.2.3Spring Context Config Class

The following class is used to instantiate the two processor classes above and make them accessible to Camel. This class would be configured in the Camel module Spring Context Config Class(es) property as value com.example.camel.HL7ToKafkaCamelAppCtx.

_/*-
 * #%L
 * Smile CDR - CDR
 * %%
 * Copyright (C) 2016 - 2025 Smile CDR, Inc.
 * %%
 * All rights reserved.
 * #L%
 */
package com.example.camel;

import ca.uhn.hl7v2.DefaultHapiContext;
import ca.uhn.hl7v2.HapiContext;
import ca.uhn.hl7v2.parser.CanonicalModelClassFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.io.IOException;

/**
 * Sample Spring Config Context class that can be used to create processor beans useful for
 * bulk ingestion of HL7v2 messages from a file into a FHIR repository.
 */
@Configuration
public class HL7ToKafkaCamelAppCtx {

   /**
    * This bean is used to calculate a Kafka partition ID for an HL7v2 message and set an
    * Exchange header that would be referenced by a downstream Kafka Camel component when
    * the message is published to a Kafka topic.
    */
   @Bean
   public HL7v2PartitionCalculateProcessor hl7v2PartitionCalculateProcessor() {
      return new HL7v2PartitionCalculateProcessor();
   }

   /**
    * This bean is used to convert HL7v2 text (e.g. loaded from a file) into a collection
    * of HL7v2 message objects.
    */
   @Bean
   public HL7v2TextParsingProcessor hl7v2TextParsingProcessor() {
      try (HapiContext myHapiContext = new DefaultHapiContext()) {
         CanonicalModelClassFactory mcf = new CanonicalModelClassFactory("2.5");
         myHapiContext.setModelClassFactory(mcf);

         return new HL7v2TextParsingProcessor(myHapiContext);
      } catch (IOException e) {
         throw new RuntimeException(e);
      }
   }
}