WSO2 & ActiveMQ: How to send and receive messages

Hello, today we will see an example of how to interact from WSO2 with an ActiveMQ queue. We will show how to configure Micro Integrator, MI, write and read from the queue. Even obtain how many records it has.

This post is based on another one from Chakray Blog, a company specialized in Integration, API, DevOps & Identity technologies with a great experience in WSO2. You can see it here.

For the example we will use the following versions:

• Micro Integrator 4.0.0
• Active MQ 5.8.14.

1. Docker Compose

Firstly, we must include the ActiveMQ libraries and dependencies in the base docker compose we have:

• activemq-broker, activemq-client and activemq-openwire-legacy 5.14.0.
• geronimo-j2ee-management_1_1_spec-1.0.1 and geronimo-jms_1.1_spec-1.1.1
• hawtbu-1.11

And include the ActiveMQ instance:

activemq:
    image: rmohr/activemq:5.14.5
    mem_limit: 1G
    hostname: activemq
    container_name: activemq
    ports:
      - 8161:8161   #UI
      - 61616:61616 #JMS
    networks:
      wso2-net:
        ipv4_address: 172.16.238.20

If we do not configure anything else, when we try to send a message to ActiveMQ, we will receive the following error:

Unexpected error during sending message out org.apache.axis2.AxisFault: 
The system cannot infer the transport information from the jms:/.....

Therefore, we must modify the deployment.toml configuration file to add:

[[transport.jms.sender]]
name = "myQueueSender"
parameter.initial_naming_factory = "org.apache.activemq.jndi.ActiveMQInitialContextFactory"
parameter.provider_url = "tcp://activemq:61616"
parameter.connection_factory_name = "QueueConnectionFactory"
parameter.connection_factory_type = "queue"
parameter.cache_level = "producer"

And to be able to read from a queue we need to configure a JMS Listener also.

[[transport.jms.listener]]
name = "myQueueConnectionFactory"
parameter.initial_naming_factory = "org.apache.activemq.jndi.ActiveMQInitialContextFactory"
parameter.provider_url = "failover:tcp://activemq:61616"
parameter.connection_factory_name = "QueueConnectionFactory"
parameter.connection_factory_type = "queue"
parameter.cache_level = "consumer"

[[transport.jms.listener]]
name = "default"
parameter.initial_naming_factory = "org.apache.activemq.jndi.ActiveMQInitialContextFactory"
parameter.provider_url = "failover:tcp://activemq:61616"
parameter.connection_factory_name = "QueueConnectionFactory"
parameter.connection_factory_type = "queue"

In this configuration it is very important to indicate the failover configuration. Otherwise, if the ActiveMQ node goes down, it will not be possible to reconnect to it unless the MI is restarted.

2. Sending messages

Now we will proceed to create the first API resource, with which we will send information to the activeMQ. But previously we will transform the message, just to add a little logic to the example.

In this code there are two interesting sections:

• The configuration of the OUT_ONLY property that allows us to tell the MI not to wait for a response from the ActiveMQ.
• The configuration of the ActiveMQ invocation, where apart from the basic values for the configuration, we can include the retry policy through the redelivery properties or increase the resilience with the failover configuration, where we can include several ActiveMQ nodes.

If we had configured with more specific values, we would not need an address endpoint with so much detail. We would only need to indicate the queue and the transport sender to be used.

<address uri="jms:/myqueue?transport.jms.ConnectionFactory=myCustomSender" />

3. Counting messages in the queue

Since version 5.8 ActiveMQ includes an API through which we can send and query messages. In addition, with the help of Jolokia, it provides a management API for the users. This API will be used to see how many messages are available in the queue.

The logic of our resource will be simple:

• We create a basic authorization header for the activeMQ API invocation.
• In the API call we indicate the broker and the queue from which we want the data. Which we get them from our own API resource.
• We read the response and if it is correct based on the value of the status field (note that always returns HTTP Status Code to 200 and Content-Type to text/plain). We read the value field and return it.

4. Reading messages

Finally, we will create a Proxy Service that allows us to read the same queue. In this case it is also important to return a response to the queue, otherwise the message will be forwarded to the Dead Letter Channel queue, ActiveMQ.DLQ.

Now we can perform the following invocation and check how we send the message on one side and read it on the other.

curl --request POST 'https://localhost:8253/mock/activeMQ' \
--header 'Content-Type: application/json' \
--data-raw '{
    "name":"the stars my destination",
    "isbn":"1-85798-814-0",
    "year": 1956
}'

With this, we have seen a complete example of what we can do to read and write to an ActiveMQ queue with WSO2. If we go a little deeper into the configuration we will be able to perform more complex tasks. As always, all the code can be seen here.