RabbitMQ

RabbitMQ is lightweight and easy to deploy messaging platform. It supports multiple messaging protocols. RabbitMQ can be deployed in distributed and federated configurations to meet high-scale, high-availability requirements. Joule provides the ability to publish to RabbitMQ queues, exchanges and topics.

Driver details

com.rabbitmq:amqp-client:5.16.0

Configuration Guide

Example configuration

rabbitPublisher:
  queue: quotes_queue

  serializer:
    formatter:
      jsonFormatter:
        encoding: UTF-8

This will publish StreamEvents as JSON formatted messages onto the quotes_queue.

Core Attributes

Available configuration parameters

Attribute	Description	Data Type	Required
host	RabbitMQ server hostname or Ip address	String Default: localhost
port	RabbitMQ server port, must be greater than 0.	Integer Default: 5672
username	User name	String Default: guest
password	password	String Default: guest
virtualHost	Creates a logical group of connections, exchanges, queues, bindings, user permissions, etc. within an instance. See vhost documentation for further information	String Default: /
autoAck	Flag to consider if server should message acknowledged once messages have been delivered. False will provide explicit message acknowledgment.	Boolean Default: true
global	QoS - true if the settings should be applied to the entire channel rather than each consumer	Boolean Default: true
durable	Set a durable queue (queue will survive a server restart)	Boolean Default: true
autoDelete	Server deletes queue when not in use	Boolean Default: true
exclusive	Exclusive queue to this connection	Boolean Default: false
arguments	Additional queue properties	Map<String, Object> Default: null
exchange	Exchange configuration	See Exchange Attributes section
routing	Routing configuration	See Routing Attributes section
serializer	Serialization configuration	See Serialization Attributes section

Exchange Attributes

Attribute	Description	Data Type	Required
name	Name of exchange	String
type	In RabbitMQ, messages are published to an exchange and, depending on the type of exchange, the message gets routed to one or more queues.	TOPIC, DIRECT, FANOUT, HEADERS Default: TOPIC
arguments	Additional binding properties	Map<String, Object> Default: null

Exchange example

exchange:
    name: marketQuotes
    type: TOPIC

Routing Attributes

Attribute	Description	Data Type	Required
event fields	List of event fields to create dynamic routing keys	String Default: None

Routing example

This example will dynamical create route binding using the market and symbol StreamEvent fields. As an example the following would be created nasdaq.IBM as a routing key.

  routing:
    event fields:
      - market
      - symbol

Serialization Attributes

This topic provides configuration parameters available object serialization process.

Attribute	Description	Data Type	Required
transform	User provided implementation that transforms the StreamEvent in to a domain class. The resulting domain class must implement Serializable interface	Implementation of StreamEventParser
formatter	If a custom transformer has not been provided the for formatter setting will be applied to the processed StreamEvent object.	Default: jsonFormatter
compressed	Not implemented	Boolean Default: false

Serializer example

serializer:
  formatter:
    jsonFormatter:
      encoding: UTF-8

Examples

All examples assume default values when not applied.

Queue based eventing

Sets up a simple publisher which sends to Json StreamEvents directly to a named queue.

rabbitPublisher:
  queue: quote_queue

  serializer:
    formatter:
      jsonFormatter:
        encoding: UTF-8

Work Queues

Publish to a Worker Queue that is used to distribute time-consuming tasks among multiple Joule workers.

rabbitPublisher:
  queue: quote_queue
  messageExpiration: 5000

  serializer:
    formatter:
      jsonFormatter:
        encoding: UTF-8

Publish/Subscribe

This example publishes events to an exchange where consumers receive these events from a connected exchange queue. Because the exchange is configured we do not need to define a queue name.

rabbitPublisher:
  exchange:
    name: quotes_exchange
    type: FANOUT

  serializer:
    formatter:
      jsonFormatter:
        encoding: UTF-8

Routing

This time we're going to make it possible to publish events using routing keys so that consumers can selectively subscribe to required events.

rabbitPublisher:
  host: localhost
  exchange:
    name: quotes_exchange
    type: TOPIC

  routing:
    event fields:
      - market
      - symbol

  serializer:
    formatter:
      jsonFormatter:
        encoding: UTF-8

Topic

This example is very similar to the previous case whereby we publish to events using routing keys. However, since we are using a topic exchange this enables the consuming client to use wildcards to subscribe to a wider set of events, see RabbitMQ source documentation for further information.

rabbitPublisher:
  host: localhost
  exchange:
    name: quotes_exchange
    type: TOPIC

  routing:
    event fields:
      - market
      - symbol

  serializer:
    formatter:
      jsonFormatter:
        encoding: UTF-8

Additional Resources

• Offical RabbitMQ documentation

• CloudAMQP documentation