Messaging & Async
Systems

// Queue, Exchange, and Binding declarations
@Configuration
public class RabbitMQConfig {

    public static final String ORDER_EXCHANGE = "order.exchange";
    public static final String NOTIFICATION_QUEUE = "order.notification.queue";
    public static final String NOTIFICATION_DLQ = "order.notification.dlq";

    @Bean
    public TopicExchange orderExchange() {
        return new TopicExchange(ORDER_EXCHANGE, true, false);
    }

    @Bean
    public Queue notificationQueue() {
        return QueueBuilder.durable(NOTIFICATION_QUEUE)
            .withArgument("x-dead-letter-exchange", "")   // default exchange
            .withArgument("x-dead-letter-routing-key", NOTIFICATION_DLQ)
            .withArgument("x-message-ttl", 60000)          // 60s TTL
            .build();
    }

    @Bean
    public Queue notificationDlq() {
        return QueueBuilder.durable(NOTIFICATION_DLQ).build();
    }

    @Bean
    public Binding notificationBinding(Queue notificationQueue, TopicExchange orderExchange) {
        // order.# matches order.created, order.confirmed, order.cancelled
        return BindingBuilder.bind(notificationQueue)
            .to(orderExchange)
            .with("order.#");
    }

    @Bean
    public Jackson2JsonMessageConverter messageConverter(ObjectMapper mapper) {
        return new Jackson2JsonMessageConverter(mapper);
    }
}

// Producer
@Service
@RequiredArgsConstructor
public class OrderEventPublisher {
    private final RabbitTemplate rabbitTemplate;

    public void publishOrderCreated(OrderCreatedEvent event) {
        rabbitTemplate.convertAndSend(
            RabbitMQConfig.ORDER_EXCHANGE,
            "order.created",   // routing key
            event
        );
    }
}

// Consumer
@Component
@RequiredArgsConstructor
public class OrderNotificationConsumer {

    @RabbitListener(queues = RabbitMQConfig.NOTIFICATION_QUEUE)
    public void handleOrderEvent(OrderCreatedEvent event,
                                  Channel channel,
                                  @Header(AmqpHeaders.DELIVERY_TAG) long tag)
            throws IOException {
        try {
            notificationService.send(event.customerId(), event.orderId());
            channel.basicAck(tag, false);   // acknowledge success
        } catch (Exception e) {
            // false = don't requeue, send to DLQ
            channel.basicNack(tag, false, false);
            log.error("Failed to process {}, sent to DLQ", event.orderId(), e);
        }
    }
}

// Enable async processing
@Configuration
@EnableAsync
public class AsyncConfig {

    @Bean(name = "emailTaskExecutor")
    public Executor emailTaskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(5);
        executor.setMaxPoolSize(20);
        executor.setQueueCapacity(100);
        executor.setThreadNamePrefix("email-async-");
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.initialize();
        return executor;
    }

    // Separate pool for heavy processing tasks
    @Bean(name = "processingTaskExecutor")
    public Executor processingTaskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(2);
        executor.setMaxPoolSize(5);
        executor.setQueueCapacity(50);
        executor.setThreadNamePrefix("processing-async-");
        executor.initialize();
        return executor;
    }
}

@Service
public class NotificationService {

    // Runs in emailTaskExecutor pool — caller doesn't wait
    @Async("emailTaskExecutor")
    public CompletableFuture<Void> sendWelcomeEmail(String email) {
        try {
            emailClient.send(email, "Welcome!", buildWelcomeTemplate(email));
            return CompletableFuture.completedFuture(null);
        } catch (Exception e) {
            log.error("Failed to send welcome email to {}", email, e);
            return CompletableFuture.failedFuture(e);
        }
    }

    // Caller can wait for the result if needed
    @Async("processingTaskExecutor")
    public CompletableFuture<ReportDto> generateReport(ReportRequest req) {
        ReportDto report = heavyReportService.generate(req);
        return CompletableFuture.completedFuture(report);
    }
}

// Calling code
@PostMapping("/api/users/register")
public ResponseEntity<UserDto> register(@RequestBody RegisterRequest req) {
    User user = userService.create(req);
    // Fire and forget — email sent in background, response is immediate
    notificationService.sendWelcomeEmail(user.getEmail());
    return ResponseEntity.status(201).body(toDto(user));
}

@Configuration
@EnableScheduling
public class SchedulingConfig {
    @Bean
    public TaskScheduler taskScheduler() {
        ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
        scheduler.setPoolSize(5);  // pool handles all @Scheduled methods
        scheduler.setThreadNamePrefix("scheduled-");
        return scheduler;
    }
}

@Component
@Slf4j
public class MaintenanceJobs {

    // Fixed rate: runs every 30s regardless of how long the previous run took
    @Scheduled(fixedRate = 30000)
    public void pollExternalApi() {
        externalApiSyncService.syncPendingOrders();
    }

    // Fixed delay: 10s AFTER the previous execution finishes
    @Scheduled(fixedDelay = 10000)
    public void cleanExpiredSessions() {
        int deleted = sessionRepo.deleteExpiredBefore(Instant.now());
        log.debug("Cleaned {} expired sessions", deleted);
    }

    // Cron: every day at 2:30 AM
    @Scheduled(cron = "0 30 2 * * *")
    public void generateDailyReport() {
        log.info("Starting daily report generation");
        reportService.generateAndEmail();
    }

    // Cron with timezone
    @Scheduled(cron = "0 0 9 * * MON-FRI", zone = "America/New_York")
    public void sendBusinessSummary() {
        emailService.sendDailySummary();
    }

    // First run delayed by 5 seconds after startup
    @Scheduled(initialDelay = 5000, fixedDelay = 60000)
    public void warmupCache() {
        cacheWarmupService.warmProductCache();
    }
}

// pom.xml: net.javacrumbs.shedlock:shedlock-spring
// + net.javacrumbs.shedlock:shedlock-provider-jdbc-template

// DB table (create once):
// CREATE TABLE shedlock(name VARCHAR(64) NOT NULL, lock_until TIMESTAMP NOT NULL,
//   locked_at TIMESTAMP NOT NULL, locked_by VARCHAR(255) NOT NULL, PRIMARY KEY (name));

@Configuration
@EnableSchedulerLock(defaultLockAtMostFor = "PT10M")
public class ShedLockConfig {
    @Bean
    public LockProvider lockProvider(DataSource ds) {
        return new JdbcTemplateLockProvider(
            JdbcTemplateLockProvider.Configuration.builder()
                .withJdbcTemplate(new JdbcTemplate(ds))
                .usingDbTime()  // use DB time to avoid clock skew between instances
                .build());
    }
}

@Component
public class ClusteredJobs {

    // Only ONE instance acquires the lock and runs the job
    // Other instances skip. Lock released after completion or after 10 min.
    @Scheduled(cron = "0 0 8 * * *")
    @SchedulerLock(
        name = "sendMorningDigest",
        lockAtLeastFor = "PT5M",    // keep lock for at least 5 min (prevent double-run on fast execution)
        lockAtMostFor = "PT30M"     // force release after 30 min (prevent stuck locks)
    )
    public void sendMorningDigest() {
        emailService.sendDailyDigest();
    }
}

// pom.xml: spring-boot-starter-websocket

@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {

    @Override
    public void configureMessageBroker(MessageBrokerRegistry config) {
        // In-memory broker for pub/sub
        config.enableSimpleBroker("/topic", "/queue");
        // Prefix for messages from client to server (@MessageMapping)
        config.setApplicationDestinationPrefixes("/app");
        // Prefix for user-specific queues (/user/{userId}/queue/...)
        config.setUserDestinationPrefix("/user");
    }

    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        registry.addEndpoint("/ws")
            .setAllowedOriginPatterns("https://*.yourdomain.com")
            .withSockJS();  // fallback for browsers without WebSocket support
    }
}

// Controller — handles messages from clients
@Controller
@RequiredArgsConstructor
public class ChatController {
    private final SimpMessagingTemplate messaging;

    // Client sends to /app/chat.send
    @MessageMapping("/chat.send")
    @SendTo("/topic/chat.global")   // broadcast to all subscribers
    public ChatMessage handleMessage(@Payload ChatMessage message,
                                      Principal principal) {
        message.setSender(principal.getName());
        return message;
    }

    // Push to a specific user — client subscribes to /user/queue/notifications
    public void pushNotification(String userId, NotificationDto notification) {
        messaging.convertAndSendToUser(userId, "/queue/notifications", notification);
    }
}

// From a service — push updates to all subscribers of an order
@Service
@RequiredArgsConstructor
public class OrderStatusPushService {
    private final SimpMessagingTemplate messaging;

    public void notifyOrderStatusChange(Order order) {
        messaging.convertAndSend(
            "/topic/orders." + order.getId(),
            new OrderStatusUpdate(order.getId(), order.getStatus())
        );
    }
}

@RestController
@RequiredArgsConstructor
public class NotificationController {

    // Store emitters per user (in production use Redis for multi-instance)
    private final Map<String, SseEmitter> emitters = new ConcurrentHashMap<>();

    // Client connects: GET /api/notifications/stream
    @GetMapping(value = "/api/notifications/stream",
                produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public SseEmitter streamNotifications(Principal principal) {
        String userId = principal.getName();

        SseEmitter emitter = new SseEmitter(Long.MAX_VALUE); // no timeout
        emitters.put(userId, emitter);

        emitter.onCompletion(() -> emitters.remove(userId));
        emitter.onTimeout(() -> emitters.remove(userId));
        emitter.onError(e -> emitters.remove(userId));

        // Send initial connection event
        try {
            emitter.send(SseEmitter.event()
                .name("connected")
                .data("Notifications stream connected for " + userId));
        } catch (IOException e) {
            emitters.remove(userId);
        }

        return emitter;
    }

    // Push notification from anywhere in the application
    public void pushToUser(String userId, NotificationDto notification) {
        SseEmitter emitter = emitters.get(userId);
        if (emitter != null) {
            try {
                emitter.send(SseEmitter.event()
                    .name("notification")
                    .id(notification.getId())
                    .data(notification)
                    .reconnectTime(3000));  // client reconnects after 3s on disconnect
            } catch (IOException e) {
                emitters.remove(userId);
            }
        }
    }
}

// Reactive alternative — Spring WebFlux with Flux
@GetMapping(value = "/api/orders/{id}/progress",
            produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<ServerSentEvent<OrderProgress>> streamOrderProgress(
        @PathVariable String id) {
    return Flux.interval(Duration.ofSeconds(1))
        .map(seq -> ServerSentEvent.<OrderProgress>builder()
            .id(String.valueOf(seq))
            .event("progress")
            .data(orderService.getProgress(id))
            .build())
        .takeUntil(event -> event.data().isComplete());
}

@Entity
@Table(name = "processed_messages")
public class ProcessedMessage {
    @Id private String messageId;
    private String topic;
    private Instant processedAt;
}

@Component
@RequiredArgsConstructor
@Transactional
public class IdempotentOrderConsumer {

    private final ProcessedMessageRepo processedRepo;
    private final OrderService orderService;

    @KafkaListener(topics = "order-commands")
    public void handleOrderCommand(
            @Payload CreateOrderCommand cmd,
            @Header(KafkaHeaders.RECORD_METADATA) RecordMetadata meta,
            Acknowledgment ack) {

        // Build a stable message ID from topic + partition + offset
        String messageId = meta.topic() + "-" + meta.partition() + "-" + meta.offset();

        // Check if already processed
        if (processedRepo.existsById(messageId)) {
            log.info("Duplicate message {} — skipping", messageId);
            ack.acknowledge();
            return;
        }

        // Process
        orderService.createOrder(cmd);

        // Mark as processed (in same transaction as the domain write)
        processedRepo.save(new ProcessedMessage(messageId, meta.topic(), Instant.now()));

        ack.acknowledge();
    }
}