Add headers example

Author: devdevx

src/main/java/com/devdevx/examples/springbootrabbitmq/headers/config/HeadersRmqConfig.java

@@ -0,0 +1,68 @@
+package com.devdevx.examples.springbootrabbitmq.headers.config;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.amqp.core.Binding;
+import org.springframework.amqp.core.BindingBuilder;
+import org.springframework.amqp.core.HeadersExchange;
+import org.springframework.amqp.core.Queue;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import java.util.HashMap;
+import java.util.Map;
+
+@Configuration
+@ConditionalOnProperty(name = "app.rabbitmq.example", havingValue = "headers")
+public class HeadersRmqConfig {
+
+    private static final Logger log = LoggerFactory.getLogger(HeadersRmqConfig.class);
+
+    @Value("${app.rabbitmq.headers.exchange}")
+    private String exchangeName;
+
+    @Value("${app.rabbitmq.headers.queue-fast}")
+    private String queueFast;
+
+    @Value("${app.rabbitmq.headers.queue-slow}")
+    private String queueSlow;
+
+    @Bean
+    HeadersExchange exchange() {
+        log.info("Creating exchange: {}", exchangeName);
+        return new HeadersExchange(exchangeName);
+    }
+
+    @Bean
+    Queue queueFast() {
+        log.info("Creating queue: {}", queueFast);
+        return new Queue(queueFast, false);
+    }
+
+    @Bean
+    Queue queueSlow() {
+        log.info("Creating queue: {}", queueSlow);
+        return new Queue(queueSlow, false);
+    }
+
+    @Bean
+    Binding bindingFast(@Qualifier("queueFast") Queue queue, HeadersExchange exchange) {
+        Map<String, Object> absolutePriority = new HashMap<>();
+        absolutePriority.put("priority", "high");
+        absolutePriority.put("client", "premium");
+        log.info("Binding queue '{}' to the exchange '{}' with all match '{}'", queueFast, exchangeName, absolutePriority);
+        return BindingBuilder.bind(queue).to(exchange).whereAll(absolutePriority).match();
+    }
+
+    @Bean
+    Binding bindingSlow(@Qualifier("queueSlow") Queue queue, HeadersExchange exchange) {
+        Map<String, Object> otherPriority = new HashMap<>();
+        otherPriority.put("priority", "normal");
+        otherPriority.put("client", "basic");
+        log.info("Binding queue '{}' to the exchange '{}' with any match '{}'", queueSlow, exchangeName, otherPriority);
+        return BindingBuilder.bind(queue).to(exchange).whereAny(otherPriority).match();
+    }
+}

src/main/java/com/devdevx/examples/springbootrabbitmq/headers/config/HeadersRmqProduConfig.java

@@ -0,0 +1,28 @@
+package com.devdevx.examples.springbootrabbitmq.headers.config;
+
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Profile;
+
+@Profile("producer")
+@Configuration
+@ConditionalOnProperty(name = "app.rabbitmq.example", havingValue = "headers")
+public class HeadersRmqProduConfig {
+
+    @Value("${app.rabbitmq.headers.exchange}")
+    private String exchangeName;
+
+    @Bean
+    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory, Jackson2JsonMessageConverter converter) {
+        final RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
+        rabbitTemplate.setMessageConverter(converter);
+        rabbitTemplate.setExchange(exchangeName);
+        return rabbitTemplate;
+    }
+
+}

src/main/java/com/devdevx/examples/springbootrabbitmq/headers/messaging/HeadersMessage.java

@@ -0,0 +1,38 @@
+package com.devdevx.examples.springbootrabbitmq.headers.messaging;
+
+public class HeadersMessage {
+    private Integer id;
+    private String message;
+
+    public HeadersMessage() {
+    }
+
+    public HeadersMessage(Integer id, String message) {
+        this.id = id;
+        this.message = message;
+    }
+
+    public Integer getId() {
+        return id;
+    }
+
+    public void setId(Integer id) {
+        this.id = id;
+    }
+
+    public String getMessage() {
+        return message;
+    }
+
+    public void setMessage(String message) {
+        this.message = message;
+    }
+
+    @Override
+    public String toString() {
+        return "HeadersMessage{" +
+                "id=" + id +
+                ", message='" + message + '\'' +
+                '}';
+    }
+}

src/main/java/com/devdevx/examples/springbootrabbitmq/headers/messaging/HeadersMessageReceiver.java

@@ -0,0 +1,33 @@
+package com.devdevx.examples.springbootrabbitmq.headers.messaging;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.amqp.rabbit.annotation.RabbitListener;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.context.annotation.Profile;
+import org.springframework.stereotype.Component;
+
+@Profile("consumer")
+@Component
+@ConditionalOnProperty(name = "app.rabbitmq.example", havingValue = "headers")
+public class HeadersMessageReceiver {
+
+    private static final Logger log = LoggerFactory.getLogger(HeadersMessageReceiver.class);
+
+    @Value("${app.rabbitmq.headers.queue-slow}")
+    private String queueSlow;
+
+    @Value("${app.rabbitmq.headers.queue-fast}")
+    private String queueFast;
+
+    @RabbitListener(queues = "${app.rabbitmq.headers.queue-slow}")
+    public void receiveFast(HeadersMessage message) {
+        log.info("Message received on queue '{}' : {}", queueSlow, message);
+    }
+
+    @RabbitListener(queues = "${app.rabbitmq.headers.queue-fast}")
+    public void receiveSlow(HeadersMessage message) {
+        log.info("Message received on queue '{}' : {}", queueFast, message);
+    }
+}

src/main/java/com/devdevx/examples/springbootrabbitmq/headers/messaging/HeadersMessageSender.java

@@ -0,0 +1,43 @@
+package com.devdevx.examples.springbootrabbitmq.headers.messaging;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.context.annotation.Profile;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+
+import java.util.Random;
+
+@Profile("producer")
+@Component
+@ConditionalOnProperty(name = "app.rabbitmq.example", havingValue = "headers")
+public class HeadersMessageSender {
+
+    private static final Logger log = LoggerFactory.getLogger(HeadersMessageSender.class);
+
+    private RabbitTemplate rabbitTemplate;
+
+    private int id = 0;
+    private Random random = new Random();
+    private String[] priorities = {"high", "normal"};
+    private String[] clients = {"premium", "basic"};
+
+    public HeadersMessageSender(RabbitTemplate rabbitTemplate) {
+        this.rabbitTemplate = rabbitTemplate;
+    }
+
+    @Scheduled(fixedDelayString = "${app.rabbitmq.delay-ms}")
+    public void send() {
+        String priority = priorities[random.nextInt(priorities.length)];
+        String client = clients[random.nextInt(clients.length)];
+        HeadersMessage message = new HeadersMessage(id++, priority + " " + client);
+        log.info("Sending message: {} with headers: priority='{}', client='{}'", message, priority, client);
+        rabbitTemplate.convertAndSend(message, m -> {
+            m.getMessageProperties().getHeaders().put("priority", priority);
+            m.getMessageProperties().getHeaders().put("client", client);
+            return m;
+        });
+    }
+}

src/main/resources/application.properties

@@ -9,4 +9,8 @@ app.rabbitmq.direct.exchange=direct-exchange
 app.rabbitmq.direct.queue=direct-queue
 app.rabbitmq.direct.routing-key=direct-routing-key
 
+app.rabbitmq.headers.exchange=headers-exchange
+app.rabbitmq.headers.queue-fast=queue-fast
+app.rabbitmq.headers.queue-slow=queue-slow
+
 app.rabbitmq.delay-ms=10000