Page Comparison

...

For a cache named myEntity with the following properties:|

...

...

...

...

...

...

...

...

...

...

...

...

...

...

...

...

...

...

...

...

...

...

...

...

...

Each cache must have a unique name, which will be automatically normalized to kebab-case (i.e. myEntity becomes my-entity)

...

Configuration

Basic Properties

...

...

...

...

...

...

...

...

...

...

...

...

...

...

...

...

...

...

...

Properties Template

For a collection of type my-collection, a set of default properties can be defined as a bean or in the application.yml:

...

Configuration

Basic Configuration

...

...

...

...

Using the MessageQueueProvider

The MessageQueueProvider instance can be injected as follows:

public class MyClass {

  @Inject
  MessageQueueProvider mqProvider;
  ...

Messages

Message Properties

| Property | Type | Default | Description | |---------------|:-------------------:|:-------:|-------------------------------------------------------------------------------------------| | body | byte[] | | The body of the message. | | queue | String | | The queue this message was sent to or delivered from. | | type | Message.MessageType | | The type of the message. Can be DIRECT or BROADCAST. See Message Types. | | properties | Map<String,String> | | Map of message properties. |

Message Types

Direct Messages

Direct messages are sent to a given queue and are meant to be consumed by a single consumer. The message is persisted in the queue until a consumer is available.

Broadcast Messages

Broadcast messages are sent to a given queue and are meant to be consumed by all consumers available at the moment of sending. The message is not persisted if no consumers are available.

Creating a new Message

The Message class provides a builder to simplify the message creation process:

Message newMessage = Message.builder()
        .body("My message")
        .type(Message.MessageType.DIRECT)
        .queue("queue")
        .property("key", "value")
        .build();

Some considerations with the Message builder:

• There are multiple ways to set the message body. If more than one is set, only the last one will be used.
• Every time the property method is called, the given property is appended to the map.
• Calling properties method will overwrite any previously set properties.

Consumers

Configuration

| Property | Type | Default | Description | |---------------------------------------|:-------:|:-------:|-------------------------------------------------------------------------------| | consumer.{name}.maxMessages | Integer | 5 | Maximum number of messages to consume in parallel. | | consumer.{name}.maxMessageRequeues | Integer | 5 | Maximum number of times a message should be sent back to the queue for retry. |

Example

...

Using the MessageQueueProvider

The MessageQueueProvider instance can be injected as follows:

public class MyClass {

  @Inject
  MessageQueueProvider mqProvider;
  ...

Messages

Message Properties

Message Types

Direct Messages

Direct messages are sent to a given queue and are meant to be consumed by a single consumer. The message is persisted in the queue until a consumer is available.

Broadcast Messages

Broadcast messages are sent to a given queue and are meant to be consumed by all consumers available at the moment of sending. The message is not persisted if no consumers are available.

Creating a new Message

The Message class provides a builder to simplify the message creation process:

Message newMessage = Message.builder()
        .body("My message")
        .type(Message.MessageType.DIRECT)
        .queue("queue")
        .property("key", "value")
        .build();

Some considerations with the Message builder:

• There are multiple ways to set the message body. If more than one is set, only the last one will be used.
• Every time the property method is called, the given property is appended to the map.
• Calling properties method will overwrite any previously set properties.

Consumers

Configuration

Example

messageQueue:
  consumer:
      directConsumer:
        maxMessages: 4
        maxMessageRequeues: 6

This configuration can be loaded as follows

protected void myMethod(@Named("direct-consumer") ConsumerProperties consumerProperties) {
  try {
    var maxRequeues = consumerProperties.getMaxMessageRequeues();
    ...
  }
}

Registering Consumers

To register a consumer do the following:

public class MyClass {

  @Inject
  MessageQueueProvider mqProvider;
  ...
  
  protected void myMethod(@Named("direct-consumer") ConsumerProperties consumerProperties) {
    try {
      mqProvider.registerConsumer(queue, consumerProperties, Message.MessageType.DIRECT, this::consumeMessage);    
    ...
    }
  }

The registerConsumer method takes the following parameters:

Producers

Configuration

Example

messageQueue:
  producer:
      sendMessageTimeout: 10000
      threads: 10
      retry:
        retries: 10
        maxMessageRequeuesdelay: 610000

This configuration can be loaded as follows

...

by injecting the ProducerProperties instance:

public class MyClass {

 try {@Inject
    var maxRequeues = consumerProperties.getMaxMessageRequeues();
 ProducerProperties producerProperties;
  ...
  }
}

...

Getting a Producer instance

To register get a consumer new producer instance do the following:

public class MyClass {

  @Inject
  MessageQueueProvider mqProvider;
  ...
  
  protected void myMethod(@Named("direct-consumer") ConsumerProperties consumerProperties) {
    try {
      var myProducer = mqProvider.registerConsumergetProducer(queue,); consumerProperties, Message.MessageType.DIRECT, this::consumeMessage);    
    ...
    }
  }

The registerConsumer method takes the following parameters:

| Parameter | Type | Description | |--------------------|:--------------------:|----------------------------------------------------------------------------------------------------------------------------------------------------------------------| | queue | String | Name of the queue to register the Consumer to. | | consumerProperties | ConsumerProperties | A ConsumerProperties instance for the consumer. | | messageType | Message.MessageType | The type of message this consumer is supposed to listen to. Can be Message.MessageType.DIRECT or Message.MessageType.BROADCAST. See Message Types. | | onMessageDelivery | Predicate<Message> | The function to execute when a message is received. Returns true if the message is successfully consumed, false otherwise. |

Producers

Configuration

| Property | Type | Default | Description | |-------------------------------|:--------:|:-------:|--------------------------------------------------------------------------------------------------| | producer.threads | Integer | 5 | Number of threads available for producers to send asyncMessages. | | producer.sendMessageTimeout | Duration | 30s | Maximum time to wait to get a confirmation that a message has been successfully sent. | | producer.retry.retries | Integer | 5 | Maximum number of times to try to send a message. | | producer.retry.delay | Duration | 1s | Time to wait before retrying. The time is multiplied by the number of retries to backoff between executions. |

Example

messageQueue:
  producer:
      sendMessageTimeout: 10000
      threads: 10
      retry:
        retries: 10
        delay: 10000

This configuration can be loaded by injecting the ProducerProperties instance:

public class MyClass {

  @Inject
  ProducerProperties producerProperties;
  ...

Getting a Producer instance

To get a new producer instance do the following:

public class MyClass {

  @Inject
  MessageQueueProvider mqProvider;
  ...
  
  protected void myMethod() {
    try {
      var myProducer = mqProvider.getProducer();    
    ...
    }
  }

Sending a message

Async Messages

producer.asyncSend(newMessage,
        (message) -> messageSuccessHandler(message),
        (message, exception) -> messageFailureHandler(message, exception));

The asyncSend method takes the following parameters:

| Parameter | Type | Description | |----------------|:------------------------------:|------------------------------------------------------------------------------------------------------------------| | message | Message | The Message to send. | | successHandler | Consumer<Message> | A consumer function to execute when the message is succesfully sent. | | failureHandler | BiConsumer<Message, Exception> | A biconsumer function to execute when there is an error sending the message and the retries have been exhausted. |

Sync Messages

producer.syncSend(newMessage);

The syncSend method takes the following parameters:

| Parameter | Type | Description | |----------------|:------------------------------:|-----------------------------------| | message | Message | The Message to send. |

Providers

RabbitMQ

To enable the RabbitMQ provider, set the messageQueue.provider property with rabbitmq as value.

Configuration

...

Sending a message
Async Messages
producer.asyncSend(newMessage,
        (message) -> messageSuccessHandler(message),
        (message, exception) -> messageFailureHandler(message, exception));
The asyncSend method takes the following parameters:
Parameter
Type
Description
message
Message
The Message to send.
successHandler
Consumer<Message>
A consumer function to execute when the message is succesfully sent.
failureHandler
BiConsumer<Message, Exception>
A biconsumer function to execute when there is an error sending the message and the retries have been exhausted.
Sync Messages
producer.syncSend(newMessage);
The syncSend method takes the following parameters:
Parameter
Type
Description
message
Message
The Message to send.
Providers
RabbitMQ
To enable the RabbitMQ provider, set the messageQueue.provider property with rabbitmq as value.
Configuration
Property
Type
Default
Description
host
String
localhost
The RabbitMQ host to connect to.
port
Integer
5672
The port to connect to.
user
String
The user in case authentication is required
password
String
The password for the user
virtualHost
String
/
The RabbitMQ virtual host to connect to.
uri
String
The complete RabbitMQ url. For example [amqps://user:pass@localhost:5671/%2f]. Setting this property will overwrite host, port, user, password and virtualHost.
...
exchangeName
...
String
...
pdp.exchange
...
The name of the exchange to send the messages to.
...
tls.enabled
...
Boolean
...
false
...
Whether TLS is enabled or not on RabbitMQ. This assumes the required certificates come from a known CA or have already been registered into the JVM cacerts keystore.
...
tls.tlsVersion
...
String
...
TLSv1.2
...
TLS version being 
...
RabbitMQ Producer Configuration
The following are the RabbitMQ specific configuration properties for a producer.
...
used.
tls.p12KeyPath
String
Path to the P12 key.
tls.p12KeyPassphrase
String
Passphrase for the P12 key.
RabbitMQ Producer Configuration
The following are the RabbitMQ specific configuration properties for a producer.
Property
Type
Default
Description
minIdleChannels
Integer
5
The minimum number of idle channels to leave on the channel pool.
...
maxIdleChannels
...
Integer
...
10
...
The maximum number of idle channels to leave on the channel pool.
...
maxOpenChannels
...
Integer
...
20
...
The maximum number of open channels to have at the same time. This includes idle and in use channels.
...
Configuration Example
messageQueue:
  provider: rabbitmq
  uri: amqps://user:pass@localhost:5671/%2f
  exchangeName: testExchange
  tls:
    enabled: true
    p12KeyPath: C:/certs/rmq.p12
    p12KeyPassphrase: rmqpass
  producer:
    minIdleChannels: 2
    maxIdleChannels: 4
    maxOpenChannels: 6
    sendMessageTimeout: 10000
    threads: 10
    retry:
      retries: 10
      delay: 10000
  consumer:
    directConsumer:
      maxMessages: 4
      mmaxMessageRequeues: 6
    broadcastConsumer:
      maxMessages: 3
      maxMessageRequeues: 6