TP14.2: Messaging with RabbitMQ

Objectives of the Lab

• Configure RabbitMQ with Spring Boot
• Create an asynchronous notification system
• Implement producers and consumers
• Manage errors and Dead Letter Queue

Estimated duration: 1h30

---

Architecture

Order Service                    Notification Service
     |                                  |
     +---> [Order Exchange] ---> [Order Queue] ---> Consumer
                                        |
                                        +--> Email
                                        +--> SMS
                                        +--> Push

---

Part 1: Configuration (15 min)

1.1 Start RabbitMQ

docker run -d --name rabbitmq \
  -p 5672:5672 \
  -p 15672:15672 \
  rabbitmq:3-management

Console: http://localhost:15672 (guest/guest)

1.2 Dependencies

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
</dependency>

1.3 Configuration application.yml

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    listener:
      simple:
        acknowledge-mode: auto
        retry:
          enabled: true
          initial-interval: 1000
          max-attempts: 3

1.4 RabbitMQ Configuration

package com.example.messaging.config;

import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitMQConfig {
    
    public static final String ORDER_QUEUE = "order.queue";
    public static final String ORDER_EXCHANGE = "order.exchange";
    public static final String ORDER_ROUTING_KEY = "order.created";
    
    public static final String DLQ_QUEUE = "order.dlq";
    public static final String DLX_EXCHANGE = "order.dlx";
    
    @Bean
    public Queue orderQueue() {
        return QueueBuilder.durable(ORDER_QUEUE)
            .withArgument("x-dead-letter-exchange", DLX_EXCHANGE)
            .withArgument("x-dead-letter-routing-key", "dlq")
            .build();
    }
    
    @Bean
    public DirectExchange orderExchange() {
        return new DirectExchange(ORDER_EXCHANGE);
    }
    
    @Bean
    public Binding orderBinding() {
        return BindingBuilder
            .bind(orderQueue())
            .to(orderExchange())
            .with(ORDER_ROUTING_KEY);
    }
    
    // Dead Letter Queue
    @Bean
    public Queue dlqQueue() {
        return QueueBuilder.durable(DLQ_QUEUE).build();
    }
    
    @Bean
    public DirectExchange dlxExchange() {
        return new DirectExchange(DLX_EXCHANGE);
    }
    
    @Bean
    public Binding dlqBinding() {
        return BindingBuilder
            .bind(dlqQueue())
            .to(dlxExchange())
            .with("dlq");
    }
    
    @Bean
    public Jackson2JsonMessageConverter messageConverter() {
        return new Jackson2JsonMessageConverter();
    }
    
    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate template = new RabbitTemplate(connectionFactory);
        template.setMessageConverter(messageConverter());
        return template;
    }
}

---

Part 2: DTOs (10 min)

package com.example.messaging.dto;

import java.io.Serializable;
import java.time.LocalDateTime;

public class OrderEvent implements Serializable {
    
    private Long orderId;
    private Long userId;
    private String userEmail;
    private String productName;
    private Double totalAmount;
    private String status;
    private LocalDateTime createdAt;
    
    public OrderEvent() {
        this.createdAt = LocalDateTime.now();
    }
    
    public OrderEvent(Long orderId, Long userId, String userEmail, 
            String productName, Double totalAmount) {
        this();
        this.orderId = orderId;
        this.userId = userId;
        this.userEmail = userEmail;
        this.productName = productName;
        this.totalAmount = totalAmount;
        this.status = "CREATED";
    }
    
    // Getters and Setters
    public Long getOrderId() { return orderId; }
    public void setOrderId(Long orderId) { this.orderId = orderId; }
    
    public Long getUserId() { return userId; }
    public void setUserId(Long userId) { this.userId = userId; }
    
    public String getUserEmail() { return userEmail; }
    public void setUserEmail(String userEmail) { this.userEmail = userEmail; }
    
    public String getProductName() { return productName; }
    public void setProductName(String productName) { this.productName = productName; }
    
    public Double getTotalAmount() { return totalAmount; }
    public void setTotalAmount(Double totalAmount) { this.totalAmount = totalAmount; }
    
    public String getStatus() { return status; }
    public void setStatus(String status) { this.status = status; }
    
    public LocalDateTime getCreatedAt() { return createdAt; }
    public void setCreatedAt(LocalDateTime createdAt) { this.createdAt = createdAt; }
    
    @Override
    public String toString() {
        return "OrderEvent{orderId=" + orderId + ", userId=" + userId + 
               ", status=" + status + "}";
    }
}

---

Part 3: Producer (15 min)

package com.example.messaging.service;

import com.example.messaging.config.RabbitMQConfig;
import com.example.messaging.dto.OrderEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Service;

@Service
public class OrderEventProducer {
    
    private static final Logger log = LoggerFactory.getLogger(OrderEventProducer.class);
    
    private final RabbitTemplate rabbitTemplate;
    
    public OrderEventProducer(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }
    
    public void sendOrderCreatedEvent(OrderEvent event) {
        log.info("Sending order event: {}", event);
        
        rabbitTemplate.convertAndSend(
            RabbitMQConfig.ORDER_EXCHANGE,
            RabbitMQConfig.ORDER_ROUTING_KEY,
            event
        );
        
        log.info("Order event sent successfully");
    }
}

---

Part 4: Consumer (15 min)

package com.example.messaging.listener;

import com.example.messaging.config.RabbitMQConfig;
import com.example.messaging.dto.OrderEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class OrderEventListener {
    
    private static final Logger log = LoggerFactory.getLogger(OrderEventListener.class);
    
    @RabbitListener(queues = RabbitMQConfig.ORDER_QUEUE)
    public void handleOrderEvent(OrderEvent event) {
        log.info("Received order event: {}", event);
        
        try {
            // Simulate processing
            processOrder(event);
            log.info("Order processed successfully: {}", event.getOrderId());
        } catch (Exception e) {
            log.error("Error processing order: {}", e.getMessage());
            throw e; // Re-throw for DLQ
        }
    }
    
    private void processOrder(OrderEvent event) {
        // Simulate sending email
        log.info("Sending confirmation email to: {}", event.getUserEmail());
        
        // Simulate processing
        try {
            Thread.sleep(500);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        
        log.info("Email sent for order: {}", event.getOrderId());
    }
    
    @RabbitListener(queues = RabbitMQConfig.DLQ_QUEUE)
    public void handleDeadLetter(OrderEvent event) {
        log.error("Dead letter received: {}", event);
        // Alert, save for analysis, etc.
    }
}

---

Part 5: Controller (10 min)

package com.example.messaging.controller;

import com.example.messaging.dto.OrderEvent;
import com.example.messaging.service.OrderEventProducer;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;

import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;

@RestController
@RequestMapping("/api/orders")
public class OrderController {
    
    private final OrderEventProducer eventProducer;
    private final AtomicLong orderIdGenerator = new AtomicLong(1);
    
    public OrderController(OrderEventProducer eventProducer) {
        this.eventProducer = eventProducer;
    }
    
    @PostMapping
    public ResponseEntity<Map<String, Object>> createOrder(@RequestBody OrderRequest request) {
        Long orderId = orderIdGenerator.getAndIncrement();
        
        OrderEvent event = new OrderEvent(
            orderId,
            request.getUserId(),
            request.getUserEmail(),
            request.getProductName(),
            request.getTotalAmount()
        );
        
        eventProducer.sendOrderCreatedEvent(event);
        
        return ResponseEntity.ok(Map.of(
            "orderId", orderId,
            "status", "CREATED",
            "message", "Order created and notification queued"
        ));
    }
    
    public static class OrderRequest {
        private Long userId;
        private String userEmail;
        private String productName;
        private Double totalAmount;
        
        // Getters and Setters
        public Long getUserId() { return userId; }
        public void setUserId(Long userId) { this.userId = userId; }
        
        public String getUserEmail() { return userEmail; }
        public void setUserEmail(String userEmail) { this.userEmail = userEmail; }
        
        public String getProductName() { return productName; }
        public void setProductName(String productName) { this.productName = productName; }
        
        public Double getTotalAmount() { return totalAmount; }
        public void setTotalAmount(Double totalAmount) { this.totalAmount = totalAmount; }
    }
}

---

Part 6: Tests (15 min)

6.1 Create an Order

curl -X POST http://localhost:8080/api/orders \
  -H "Content-Type: application/json" \
  -d '{
    "userId": 1,
    "userEmail": "john@example.com",
    "productName": "Laptop",
    "totalAmount": 999.99
  }'

6.2 Send Multiple Orders

for i in {1..10}; do
  curl -X POST http://localhost:8080/api/orders \
    -H "Content-Type: application/json" \
    -d "{
      \"userId\": $i,
      \"userEmail\": \"user$i@example.com\",
      \"productName\": \"Product $i\",
      \"totalAmount\": $((i * 100))
    }"
  echo ""
done

6.3 Check RabbitMQ Console

1. Open http://localhost:15672
2. Log in (guest/guest)
3. Check the "Queues" tab
4. Observe the processed messages

---

Validation Checklist

• RabbitMQ starts on localhost:15672
• Exchanges and queues are created
• Messages are published correctly
• Consumer processes the messages
• Logs display the processing
• DLQ captures errors

---

TP14.3: WebSocket Chat

Back to Module 14