Camel processors only require the Camel (CAMEL) module.
allowedPartitionsBy default, a processor that writes or reads FHIR data uses whatever partition the inbound message asks for. To limit a route to specific partitions, add the allowedPartitions parameter to the processor URI – for example smile:persistence/bundleProcessor?allowedPartitions=2,3. A message that asks for a partition outside the list is rejected rather than processed. The parameter is available on every processor that takes its partition from the message: bundleProcessor, singleResourceProcessor, channelImportProcessor, patientMergeProcessor, replaceReferencesProcessor, and the terminology processors.
allowedPartitions is a comma-separated list of partition identifiers. Each entry is either a partition ID or a partition name, and the DEFAULT token refers to the default partition (the DEFAULT and _ALL tokens are not case-sensitive). Its behaviour is:
_ALL (default) – any partition is permitted, so a route without this parameter is not restricted.2,3) – messages may use only the listed partitions, plus the default partition. A message requesting any other partition is rejected.Only a partition that comes from the message is checked against the list – one supplied in the X-Request-Partition-IDs header or in Pulsar message properties. A partition you set directly on the route with the partitionId parameter is always allowed.
allowedPartitions has an effect only when partitioning is enabled on the persistence module. On a non-partitioned server there is a single partition, so the parameter does nothing.
Use partition IDs, not names, for header-based routes. For every processor except channelImportProcessor, the partition comes from the X-Request-Partition-IDs header (or Pulsar properties), which is always a numeric partition ID (or the DEFAULT/_ALL token) – never a name. A list of names therefore matches nothing on these routes and rejects every non-default message, so list partition IDs instead. Partition names are matched only by channelImportProcessor, whose messages can carry a partition by name (see Channel Import With Partitioning Enabled).
A rejected message raises a ForbiddenOperationException, which the route's error handling can catch to retry it or send it to a failure queue.
smile:camel/channelImportProcessorapplication/fhir+json, application/json, and text/plain; CSV, HL7 v2.x, and CDA are not supported. FHIR payloads are persisted according to the message's operationType and require the Camel Module to declare a FHIR Storage module dependency. JSON and string payloads are passed to the handleChannelImportJsonPayload and handleChannelImportStringPayload JavaScript functions, which must be defined in the Camel Module. Messages must include the headers/payload wrapper of a ResourceOperationJsonMessage. Note that even though this processor is called channelImportProcessor, it does not interact with a Channel Import Module in any way. It is provided to give users an option to migrate from the Channel Import Module to the Camel Module. When migrating, the Channel Import Module must be archived; otherwise the module and the Camel route will both consume the same channel, and messages will be ingested twice.When processing ResourceOperationJsonMessage objects, this processor follows the enhanced partition precedence rules:
partitionId is already set on the inbound message, it takes precedenceX-Request-Partition-IDs header (passed as a Camel Exchange header) is used only if no partitionId is set on the messageSee Restricting Partitions with allowedPartitions, which applies to this and every other Smile Camel processor that resolves a partition from an inbound message.
The following routes fully replicate the functionality in the Channel Import Module when a ResourceOperationJsonMessage message is posted to the kafka topic called channel-import-in-topic.
<routes xmlns="http://camel.apache.org/schema/spring">
<route id="channelImportProcessor">
<from uri="kafka:channel-import-in-topic?brokers=localhost:9092"/>
<to uri="direct:processMessage"/>
</route>
<route id="processResourceOperationMessage">
<from uri="direct:processMessage"/>
<onException>
<exception>ca.uhn.fhir.rest.server.exceptions.ResourceNotFoundException</exception>
<exception>ca.uhn.fhir.rest.server.exceptions.InvalidRequestException</exception>
<exception>ca.uhn.fhir.rest.server.exceptions.InternalErrorException</exception>
<handled>
<constant>true</constant>
</handled>
<to uri="smile:camel/incrementRetryCountProcessor"/>
<log logName="ca.cdr.camel" message="Error processing ${body}. Sending to retry queue." loggingLevel="DEBUG"/>
<to uri="kafka:test.retry?brokers=localhost:9092"/>
</onException>
<onException>
<exception>java.lang.Exception</exception>
<handled>
<constant>true</constant>
</handled>
<to uri="direct:failureQueue"/>
</onException>
<log logName="ca.cdr.camel" message="Processing Resource Operation ${body}" loggingLevel="DEBUG"/>
<to uri="smile:camel/channelImportProcessor"/>
</route>
<route id="retryResourceOperationMessage">
<from uri="kafka:test.retry?brokers=localhost:9092"/>
<choice>
<when>
<jsonpath>$.headers[?(@.retryCount < 3)]</jsonpath>
<delay>
<constant>1000</constant>
</delay>
<log logName="ca.cdr.camel" message="Retrying ${body}" loggingLevel="DEBUG"/>
<to uri="direct:processMessage"/>
</when>
<otherwise>
<to uri="direct:failureQueue"/>
</otherwise>
</choice>
</route>
<route id="failureQueue">
<from uri="direct:failureQueue"/>
<log logName="ca.cdr.camel" message="Error processing ${body}. Sending to Failure queue." loggingLevel="DEBUG"/>
<to uri="kafka:test.failed?brokers=localhost:9092"/>
</route>
</routes>
smile:camel/incrementRetryCountProcessorSee the Example Route in channelImportProcessor above for an example of how it can be used.
Example URI: smile:camel/script?function=myFunction&input=bundle&includeExchange=true
Description: This calls an arbitrary JavaScript function with the input from the route. The function can optionally receive the Camel Exchange as a second parameter.
Parameters:
function: The JavaScript function to call (required)input: Specifies the format to convert the input data before calling the function (optional)
json: converts input message into a JSON documenthl7v2: converts input message into an HL7v2 messagebundle input type but that is now automatically handled without specifying an inputType)includeExchange: When equal to true, passes the Camel Exchange as a second parameter to the function (optional)Input:
exchange.headers: Provides read/write access to exchange message headersexchange.properties: Provides read/write access to exchange propertiesOutput: The return value of the function will be passed to the next step in the route.
Below are example routes using the script processor with each of these input types:
<routes xmlns="http://camel.apache.org/schema/spring">
<route id="string">
<from uri="direct:stringStart"/>
<to uri="smile:camel/script?function=processString"/>
<to uri="mock:stringResult"/>
</route>
<route id="stringWithExchange">
<from uri="direct:stringWithExchangeStart"/>
<setProperty name="myProperty">
<constant>original property value</constant>
</setProperty>
<setHeader name="myHeader">
<constant>original header value</constant>
</setHeader>
<to uri="smile:camel/script?function=processStringWithExchange&includeExchange=true"/>
<to uri="mock:stringWithExchangeResult"/>
</route>
<route id="json">
<from uri="direct:jsonStart"/>
<to uri="smile:camel/script?function=processJson&input=json"/>
<to uri="mock:jsonResult"/>
</route>
<route id="bundle">
<from uri="direct:bundleStart"/>
<to uri="smile:camel/script?function=processBundle&input=bundle"/>
<to uri="mock:bundleResult"/>
</route>
<route id="hl7v2">
<from uri="direct:hl7v2Start"/>
<to uri="smile:camel/script?function=processHl7v2&input=hl7v2"/>
<to uri="mock:hl7v2Result"/>
</route>
</routes>
Here are some example JavaScript functions that would work with the above routes:
function processString(string) {
var parts = string.split(/\s+/)
var patient = ResourceBuilder.build('Patient');
patient.name[0].given[0] = parts[0];
patient.name[0].family = parts[1];
return patient;
}
function processStringWithExchange(message, exchange) {
// Access exchange properties
var properties = exchange.properties;
var originalPropertyValue = properties['myProperty'];
// Change existing exchange property
properties['myProperty'] = 'changed property value';
// Add new exchange property
properties['newProperty'] = 'added property';
// Access exchange headers
var headers = exchange.headers;
var originalHeaderValue = headers['myHeader'];
// Change existing exchange header
headers['myHeader'] = 'changed header value';
// Add new exchange header
headers['newHeader'] = 'added header';
return message + ", " + originalPropertyValue + ", " + originalHeaderValue;
}
function processJson(json) {
// Create a patient
var parts = json.name.split(/\s+/)
var patient = ResourceBuilder.build('Patient');
patient.name[0].given[0] = parts[0];
patient.name[0].family = parts[1];
// Add the patient to a transaction bundle
var transaction = TransactionBuilder.newTransactionBuilder();
transaction.create(patient);
return transaction;
}
function processBundle(bundle) {
var resources = bundle.entryResources();
return resources[0].name[0].family;
}
function processHl7v2(message) {
// Create a patient from an ORU_R01 message
var patient = ResourceBuilder.build('Patient');
patient.name[0].family = message['/PATIENT_RESULT[0]/PATIENT/PID-5[0]-1'];
patient.name[0].given[0] = message['/PATIENT_RESULT[0]/PATIENT/PID-5[0]-2'];
// Add the patient to a transaction bundle
var transaction = TransactionBuilder.newTransactionBuilder();
transaction.create(patient);
return transaction;
}
smile:camel/wrapInDocumentReferenceProcessor?status=current&contentType=application/xml&docStatus=final&typeCode=34133-9&formatCode=urn:hl7-org:sdwg:ccda-structuredBody:2.1&contentUrl=/Binary/123
Description: Builds a DocumentReference resource from the input and specified parameters.
Input:
String - This will be used to populate the DocumentReference.content.attachment.data element or,Parameters:
status (required): The status of the document (i.e. current). This will be used to populate the DocumentReference.status element.contentType (required): The content type of the document (i.e. application/xml). This will be used to populate the DocumentReference.content.attachment.contentType element.docStatus (optional): The doc status of the document (i.e. final). This will be used to populate the DocumentReference.docStatus element.typeCode (optional): The type code of the document (i.e. 34133-9). This will be used to populate the DocumentReference.type.coding.code element.formatCode (optional): The format code of the document (i.e. urn:hl7-org:sdwg:ccda-structuredBody:2.1). This will be used to populate the DocumentReference.content.format.code element.contentUrl (optional): The content url of the Attachment (i.e. Attachment.url).Output:
DocumentReference resource with the elements above populated.NOTE: This processor does not persist the resulting DocumentReference. To persist the DocumentReference,
the singleResourceProcessor should be used.
<route>
<from uri="file:input/cda?noop=true"/>
<to uri="smile:camel/wrapInDocumentReferenceProcessor?status=current&contentType=application/xml&docStatus=final&typeCode=34133-9&formatCode=urn:hl7-org:sdwg:ccda-structuredBody:2.1&contentUrl=/Binary/123" />
<to uri="smile:persistence/singleResourceProcessor?method=CREATE" />
</route>
This processor supports automatic processing of FhirToCdaConversionResultJson objects from CDA Exchange workflows. When this type of input is detected:
DocumentReference.subject is determined from FhirToCdaConversionResultJson.bundleDocumentReference.type is determined from FhirToCdaConversionResultJson.bundleDocumentReference.content.attachment.data is determined from FhirToCdaConversionResultJson.modifiableDocumentNote: The contentUrl parameter should still be specified.
See the Sample CDA Export Route Configuration for usage of the wrapInDocumentReferenceProcessor.
Example URI: <to uri="smile:camel/wrapInBinaryResourceProcessor?contentType=application/xml">
Description: Builds a Binary resource from the input and specified parameters.
Input:
String - This will be used to populate the Binary.data element or,Parameters:
contentType (required): Sets the Binary.contentType value (e.g., application/xml, application/pdf, text/plain)Output:
Binary resource with the elements above populated.The following example shows a route that reads content from a file, wraps it in a Binary resource, and stores it in a FHIR repository:
<route>
<from uri="file:input/documents?noop=true"/>
<to uri="smile:camel/wrapInBinaryResourceProcessor?contentType=application/xml"/>
<to uri="smile:persistence/singleResourceProcessor"/>
</route>
Similar to the wrapInDocumentReferenceProcessor, this processor also supports automatic processing of FhirToCdaConversionResultJson objects from CDA Exchange workflows. When this type of input is detected:
Binary.data is determined from FhirToCdaConversionResultJson.modifiableDocumentBinary.contentType is set to application/xmlSee the Sample CDA Export Route Configuration for usage of the wrapInBinaryResourceProcessor.