Plugin voor vertraagde berichten

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.util.HashMap;
import java.util.Map;

public class DeclareDelayedExchange {
    private static final String EXCHANGE_NAME = "my_delayed_exchange";

    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost"); // Assuming RabbitMQ is local
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {

            Map<String, Object> args = new HashMap<>();
            args.put("x-delayed-type", "direct"); // Underlying exchange type

            channel.exchangeDeclare(EXCHANGE_NAME, "x-delayed-message", true, false, args);
            System.out.println("Delayed exchange '" + EXCHANGE_NAME + "' declared.");
        }
    }
}

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.util.HashMap;
import java.util.Map;

public class DelayedMessageProducer {
    private static final String EXCHANGE_NAME = "my_delayed_exchange";
    private static final String ROUTING_KEY = "delayed.key";

    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactoryFactory();
        factory.setHost("localhost");
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {

            // Ensure the delayed exchange is declared (from previous scene)
            Map<String, Object> args = new HashMap<>();
            args.put("x-delayed-type", "direct");
            channel.exchangeDeclare(EXCHANGE_NAME, "x-delayed-message", true, false, args);
            System.out.println("Exchange declared (if not exists).");

            String message = "Hello, delayed world!";
            int delayInMs = 5000; // 5 seconds

            Map<String, Object> headers = new HashMap<>();
            headers.put("x-delay", delayInMs); // Set the delay header

            AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
                .headers(headers)
                .build();

            channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, props, message.getBytes("UTF-8"));
            System.out.println(" [x] Sent '" + message + "' with delay " + delayInMs + "ms");
        }
    }
}

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;
import java.nio.charset.StandardCharsets;
import java.util.Date;

public class DelayedMessageConsumer {
    private static final String EXCHANGE_NAME = "my_delayed_exchange";
    private static final String QUEUE_NAME = "delayed_queue";
    private static final String ROUTING_KEY = "delayed.key";

    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        // Declare the queue and bind it to the delayed exchange
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);
        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");

        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
            System.out.println(" [x] Received '" + message + "' at " + new Date());
        };
        channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> {});
    }
}