ActiveMQ Configuration

ActiveMQ is an open-source message broker software that implements the Java Message Service (JMS) API, as well as other protocols such as Advanced Message Queuing Protocol (AMQP), Streaming Text Oriented Messaging Protocol (STOMP), and MQ Telemetry Transport (MQTT). It is written in Java and is built on the Apache ActiveMQ project, which provides a high-performance, reliable, and scalable messaging infrastructure.

ActiveMQ can be used to decouple and distribute systems by sending messages between them. It supports a variety of messaging patterns, including point-to-point, publish/subscribe, and request/response. It also provides features such as message persistence, load balancing, and clustering, which allows for high availability and scalability.

ActiveMQ also has a built-in management console, which allows for monitoring and management of the broker, queues, and connections. It also supports various plugins, such as the Apache Camel plugin, that provide additional functionality.

ActiveMQ Artemis

ActiveMQ Artemis is a high-performance, open-source message broker that is the next generation of the ActiveMQ project. It is designed to handle large numbers of concurrent clients and large volumes of messages, while providing low latency and high throughput. It is written in Java and is built on the HornetQ project, which provides a high-performance, reliable, and scalable messaging infrastructure.

Configure ActiveMQ

To configure the ActiveMQ transport, use the UsingActiveMq method.

using MassTransit;

services.AddMassTransit(x =>
{
    x.AddDelayedMessageScheduler();

    x.AddConsumer<AuditOrderCreatedConsumer>();

    x.UsingActiveMq((context, cfg) =>
    {
        cfg.Host("localhost", h =>
        {
            h.UseSsl();

            h.Username("admin");
            h.Password("admin");
        });

        cfg.UseDelayedMessageScheduler();

        cfg.ConfigureEndpoints(context));
    });
});

The configuration includes:

The ActiveMQ host
- Host name: localhost
- Username and password used to connect to the host

The port can also be specified as an additional parameter on the Host method. If port 61617 is specified, SSL is automatically enabled.

MassTransit includes several Receive Endpoint level configuration options that control Receive Endpoint behavior.

Property	Description
PrefetchCount	The number of unacknowledged messages that can be processed concurrently (default based on CPU count)
AutoDelete	If true, the queue will be automatically deleted when the bus is stopped (default: false)
Durable	If true, messages are persisted to disk before being acknowledged (default: true)

When using ActiveMQ, Receive Endpoint queue names must not include any . characters. Using a dotted queue name will break pub/sub message routing. If using a dotted queue name is required, such as when interacting with an existing queue, disable topic binding on the Receive Endpoint by specifying:

endpoint.ConfigureConsumeTopology = false;

When the Consume Topology is not configured, the virtual consumer queues are not created.

Using Amazon MQ

Amazon MQ uses ActiveMQ, so the same transport is used. Amazon MQ requires SSL, so if MassTransit detects the host name ends with amazonaws.com, SSL is automatically configured.

Using Artemis

Artemis also supports the OpenWire protocol. However, some differences exist that cause the MassTransit ActiveMQ transport provider to malfunction. One of those causes is that Artemis works differently with queues compared to ActiveMq. See Artemis:Virtual Topics

To use Artemis with MassTransit, use the EnableArtemisCompatibility() method on the ActiveMQ configuration.

services.AddMassTransit(x =>
{
    x.UsingActiveMq((context, cfg) =>
    {
        cfg.Host("localhost", 61618, cfgHost =>
        {
            cfgHost.Username("admin");
            cfgHost.Password("admin");
        });

        cfg.EnableArtemisCompatibility();
    });
});

Calling cfg.EnableArtemisCompatibility() will initialize the minimum necessary features so that the Masstransit ActiveMQ transport provider will work with the Artemis broker

Currently, the only thing cfg.EnableArtemisCompatibility() does is setting a predefined formatter ArtemisConsumerEndpointQueueNameFormatter (which implements the IActiveMqConsumerEndpointQueueNameFormatter interface) on the ConsumeTopology (accessible via cfg.ConsumeTopology on the bus configuration).

Example of setting your own ConsumerEndpointQueueNameFormatter:

cfg.SetConsumerEndpointQueueNameFormatter(new MyCustomConsumerEndpointQueueNameFormatter());

So it is still possible to create your own IActiveMqConsumerEndpointQueueNameFormatter if you want to tweak the queue name. The responsibility of the formatter is to return the queue name for the following cases:

- a Receive Endpoint name
- a topic.

TemporaryQueueNameFormatter

On the Consume Topology a TemporaryQueueNameFormatter can be configured. The responsibility of the formatter is to transform the ‘system’ generated name for a temporary queue.

This could be used to e.g., add a prefix to the generated temporary queue names. This helps to support namespaces in queue names. Artemis can use this to enforce security policies.

For adding a prefix, a handy helper is already provided. During bus configuration, specify the prefix as shown:

cfg.SetTemporaryQueueNamePrefix("mycustomnamespace.");

Which is equivalent to the following:

cfg.SetTemporaryQueueNameFormatter( new PrefixTemporaryQueueNameFormatter("mycustomnamespace."));