# Message Brokers

Smile CDR uses a message broker for [Subscription](/content/docs/subscription/index.html) processing.

The message broker is associated with the overall Smile CDR cluster, meaning that one single logical broker is shared by all modules (although this broker may itself be clustered).

There are three broker technologies supported by Smile CDR:

- Apache ActiveMQ / JMS ( [documentation](/content/docs/guide_installation/activemq.html)). This is the default.
- Apache Kafka ( [documentation](/content/docs/guide_installation/kafka.html)).
- Apache Pulsar ( [documentation](/content/docs/guide_installation/pulsar.html)).

The number of threads allocated to subscription processing is configured in the storage module using these properties:

```properties
module.persistence.config.subscription.consumers_per_matching_queue =2
module.persistence.config.subscription.consumers_per_delivery_queue =5
```

Copy

These determine the number of consumers (threads) that listen to the two different types of subscription queues used by Smile CDR. See [Smile CDR Subscription Architecture](/content/docs/subscription/architecture.html).

- [`consumers_per_matching_queue`](/content/docs/configuration_categories/fhir_subscription_persistence.html#subscription.consumers_per_matching_queue/index.html) is the number of listener threads that match new resources against active subscriptions

- [`consumers_per_delivery_queue`](/content/docs/configuration_categories/fhir_subscription_persistence.html#subscription.consumers_per_delivery_queue/index.html) is the number of listener threads created per subscription. These threads deliver subscription matches to subscription endpoints.

# Subscription Channels

Smile CDR performs subscription matching by registering an interceptor with a [FHIR Storage](/content/docs/fhir_storage_relational/fhir_storage_relational_module.html) module and submits all incoming resources to the Subscription Matching Queue on the configured Message Broker. (In the case of Kafka, Kafka topics are used in place of JMS queues.) A Smile CDR [Subscription Matching](/content/docs/subscription/index.html) module subscribes to this Subscription Matching Queue and matches all incoming resources against all active subscriptions. Each active subscription has its own dedicated Delivery Queue. When a resource matches a subscription, it is placed on the Delivery Queue for that subscription. A separate consumer within the Subscription Matching module then picks up the matched resource from the Delivery Queue and attempts to deliver it to the endpoint specified by that subscription. If a single endpoint is down, only queues for subscriptions delivering to that endpoint are affected; deliveries to other endpoints continue unimpeded.

# Subscription Channel Names

All Subscription Queue names end with `-[NODE_ID]-[STORAGE_MODULE_ID]` where `[NODE_ID]` and `[STORAGE_MODULE_ID]` are the Node ID and FHIR Storage module ID for the module that is submitting the events for processing. In the default Smile CDR configuration these will be `Master` and `persistence` respectively.

The name of the Subscription Matching Queue is `subscription-matching-[NODE_ID]-[STORAGE_MODULE_ID]`.

The name of each rest-hook delivery queue is `subscription-delivery-rest-hook-[NAME]-[NODE_ID]-[STORAGE_MODULE_ID]` where `[NAME]` depends on the value of [`module.subscription.config.delivery_queue_naming`](/content/docs/configuration_categories/fhir_subscription_delivery.html#property-subscription-delivery-queue-naming/index.html). The value of `delivery_queue_naming` can be one of the following options:

- `SUBSCRIPTION_ID`: Create one delivery queue per subscription with the NAME equal to the subscription id. This is the default for backwards compatibility.

- `SHARED` Create a single delivery queue shared by all subscriptions

- `DELIVERY_GROUP`: Look for the extension `https://smilecdr.com/fhir/ns/StructureDefinition/subscription-channel-delivery-group` on each subscription and use the value of that extension for the name of the delivery queue. When the extension is not present, the system treats the subscription as shared and sets the `[NAME]` to `SHARED`.

The following three options derive the delivery queue name from the REST endpoint url of the subscription. E.g. if the subscription endpoint url is http://some.hostname.org:1234/app/path/subpath?foo=bar&baz=some/path then the queue name for that subscription delivery queue will be as follows.

- `ENDPOINT_HOST`: "some-hostname-org"

- `ENDPOINT_HOST_PORT`: "some-hostname-org-1234". This is the recommended value for enterprise installations. The advantage of this option is that if there are issues delivering to a particular endpoint, those issues will be constrained to the single threadpool managing deliveries for that queue.

- `ENDPOINT_HOST_PORT_PATH`: "some-hostname-org-1234-app-path-subpath"

When naming option is set to one of the ENDPOINT\* options and the Subscription is not a rest-hook subscription, Smile CDR will still attempt to parse out a host and port number from the Subscription.channel.endpoint and set the `[NAME]` accordingly. If the endpoint contains a value that does not meet the minimum requirements for a URL as defined by https://www.ietf.org/rfc/rfc2732.txt, it will instead set the `[NAME]` based on host name malformed.endpoint.url and port number 0.

Similarly, the name of each email delivery queue is `subscription-delivery-email-[NAME]-[NODE_ID]-[STORAGE_MODULE_ID]`.

## Prefixes

If the [Channel Naming Prefix](/content/docs/configuration_categories/clustermgr_message_broker.html#property-channel-naming-prefix/index.html) property is set, the given value will be prepended to all non-customizable channel names used by the broker. This can be useful if you are deploying to an environment with enterprise rules around channel naming.

Non-customizable channels used by Smile CDR, to which this configuration is applied:

- `empi` topic used by [MDM Module](/content/docs/modules/mdm.html).
- `subscription.delivery`, `subscription.matching` topics used by [Subscriptions](/content/docs/subscription/introduction.html).
- `batch2.work.notification` topic used by [Batch jobs](/content/docs/fhir_repository/batch_and_scheduled_jobs.html).
- `KAFKA.DLQ` topic used as kafka [Dead Letter Queue](/content/docs/subscription/introduction.html#dlq/index.html).

If it is required to add prefixes for custom Smile CDR channel names, prefixes could be added to each configuration property individually:

- For channels created by the [Channel Import Module](/content/docs/modules/channel_import.html), prefix could be added to properties listed below:

- [Property: Channel Name](/content/docs/configuration_categories/channel_import.html#property-channel-name/index.html)
  - [Property: Failure Channel Name](/content/docs/configuration_categories/channel_retry.html#property-failure-channel-name/index.html)
  - [Property: Retry Channel Name](/content/docs/configuration_categories/channel_retry.html#property-retry-channel-name/index.html)
- For channels created by the [Realtime Export Module](/content/docs/modules/realtime_export.html), prefix could be added to the [Channel Name Prefix](/content/docs/configuration_categories/realtime_export.html#property-channel-name-prefix/index.html) property.
- For Audit Log channel created if [Audit Log Broker](/content/docs/configuration_categories/clustermgr_maintenance.html#property-audit-log-broker-enabled/index.html) is enabled in Cluster Manager configuration prefix could be added to the [Audit Log Broker Channel Name](/content/docs/configuration_categories/clustermgr_maintenance.html#property-audit-log-broker-channel-name/index.html) property.
- For Transaction Log channel prefix could be added to the [Transaction Log Broker Channel Name](/content/docs/configuration_categories/transactionlog.html#property-transaction-log-broker-channel-name/index.html) property.
- For ActiveMQ [Dead Letter Queue](/content/docs/subscription/introduction.html#dlq/index.html) prefix could be added by modifying the destination policy in the `activemq.xml` configuration file. See [Message Redelivery and DLQ Handling](https://activemq.apache.org/components/classic/documentation/message-redelivery-and-dlq-handling) for the details. Example of modified `activemq.xml` configuration:

```xml
<broker>
   <destinationPolicy>
      <policyMap>
         <policyEntries>
         <!-- Set the following policy on all queues using the '>' wildcard -->
             <policyEntry queue=">">
               <deadLetterStrategy>
                  <!-- Use the prefix 'PREFIX.DLQ.' for the destination name -->
                  <individualDeadLetterStrategy queuePrefix="PREFIX.DLQ." useQueueForQueueMessages="true"/>
               </deadLetterStrategy>
            </policyEntry>
         </policyEntries>
      </policyMap>
   </destinationPolicy>
</broker>
```

Copy

# Modules Dependent On Message Brokers

The following list shows the modules that require the message broker when enabled:

- Subscriptions
- Batch Job Management
- Audit Log
- Transaction Log
- MDM
- Channel Import
- Realtime Export
- Camel

If you do not want to use message brokers, ensure that none of these modules are enabled.

# Message Ordering and Partition Keys

When processing ordered message streams (such as HL7v2 feeds or sequences of cda, csv, or FHIR data updates), it is important to ensure that messages about the same entity are processed in order. Both Kafka and Pulsar support this through **partition keys** (called "message keys" in Pulsar).

Messages with the same partition key are guaranteed to be delivered to the same partition, and within a partition, messages are processed in order. This ensures that all messages related to the same patient or entity are processed sequentially.

**Choosing a Partition Key**:

- **Patient-centric data**: Use the patient identifier as the partition key. This ensures all messages about the same patient (ADT events, lab results, clinical documents) are processed in order.
- **Non-patient-centric data**: Determine some other partition key. If the volume of non-patient data is low, you can use a shared constant key to route all such messages to a single partition.

**Broker Support**:

- **Kafka**: Full partition key support. See [Message Ordering with Partition Keys](/content/docs/guide_installation/kafka.html#message-ordering/index.html) for details.
- **Pulsar**: Full message key support. See [Message Ordering with Message Keys](/content/docs/guide_installation/pulsar.html#message-ordering/index.html) for details.
- **ActiveMQ**: Does not support partitioning for in-order delivery. Use Kafka or Pulsar when message ordering is required.

For Camel routes, see [Message Keys for Ordering](/content/docs/camel/smile_camel_processors_cluster_manager.html#message-keys/index.html) for guidance on setting partition keys.

For Channel Import, see [Message Ordering for Channel Import](/content/docs/channel_import/channel_import_overview.html#message-ordering/index.html) for guidance on ensuring ordered processing.

# Sending your own messages to Message Brokers

It is possible to send your own messages to custom topics or queues on your configured message broker. This can be done for example from these places:

- Custom Resource Providers
- Interceptors

To do this, you must include the `IPublicBrokerSender` as an autowired bean in your custom class, and make use of its public API methods. The interface provides both type-safe and legacy methods for sending messages:

## Type-Safe Message Sending (Recommended)

```java
import ca.cdr.api.broker.IPublicBrokerSender;
import ca.cdr.api.broker.PublicBrokerSendRequest;
import ca.uhn.fhir.broker.api.RawStringMessage;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.HashMap;
import java.util.Map;

@Autowired
private IPublicBrokerSender myPublicBrokerSender;

private void sendTypedMessageToBroker() throws Exception {
    Map<String, Object> headers = new HashMap<>();
    headers.put("some-header", "some-value");
    String topic = "my-custom-topic";
    String payload = "This is the custom message payload.";

// Send with explicit message and payload types
    myPublicBrokerSender.send(new PublicBrokerSendRequest<>(topic, payload, RawStringMessage.class, String.class, headers));
}
```

Copy

## Legacy Message Sending (String-based)

For backward compatibility, a simplified method is also available:

```java
private void sendLegacyMessageToBroker() throws Exception {
    Map<String, Object> headers = new HashMap<>();
    headers.put("some-header", "some-value");
    String topic = "my-custom-topic";
    String payload = "This is the custom message payload.";

// Send using legacy method (deprecated)
    myPublicBrokerSender.send(topic, payload, headers);
}
```

Copy

[Designing a Cluster](/content/docs/guide_installation/designing_a_cluster.html)

Message Brokers

Installing

[Platform Requirements](/content/docs/guide_installation/platform_requirements.html)
[Preparing a Linux Host](/content/docs/guide_installation/preparing_a_linux_host.html)
[Installing Smile CDR, NGINX and PostgreSQL in a Docker Stack](/content/docs/guide_installation/installing_smilecdr_nginx_and_postgresql_in_a_docker_stack.html)
[Deploying a Kubernetes Managed Cluster](/content/docs/guide_installation/deploying_a_kubernetes_managed_cluster.html)
[Docker Container Installation](/content/docs/guide_installation/docker_container_installation.html)
[Configuring Smile CDR](/content/docs/guide_installation/configuring_smile_cdr.html)
[Unix Service Installation](/content/docs/guide_installation/unix_service_installation.html)
[Tuning your Installation](/content/docs/guide_installation/tuning_your_installation.html)
[Designing a Cluster](/content/docs/guide_installation/designing_a_cluster.html)
[Message Brokers](/content/docs/guide_installation/message_broker.html)
[Message Broker Failure Management](/content/docs/guide_installation/message_broker_failure_management.html)
[Message Broker: Kafka](/content/docs/guide_installation/kafka.html)
[Message Broker: ActiveMQ](/content/docs/guide_installation/activemq.html)
[Message Broker: Pulsar](/content/docs/guide_installation/pulsar.html)
[Pre-Seeding Configuration and Data](/content/docs/guide_installation/pre_seeding.html)
[Production Checklist](/content/docs/guide_installation/production_checklist.html)
[Module Licensing](/content/docs/guide_installation/licensing.html)

[Message Broker Failure Management](/content/docs/guide_installation/message_broker_failure_management.html)