Add the rabbitmq module

rabbitmq/src/main/java/com/ankurm/rabbit/ConverterConfiguration.java

@@ -0,0 +1,36 @@
+package com.ankurm.rabbit;
+
+import org.springframework.amqp.support.converter.JacksonJsonMessageConverter;
+import org.springframework.amqp.support.converter.MessageConverter;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * Spring Boot does <b>not</b> auto-configure a JSON message converter for RabbitMQ. The default
+ * is {@code SimpleMessageConverter}, which handles {@code String}, {@code byte[]} and
+ * {@code Serializable} and nothing else:
+ *
+ * <pre>
+ * IllegalArgumentException: SimpleMessageConverter only supports String, byte[] and
+ * Serializable payloads, received: com.ankurm.rabbit.OrderMessage
+ * </pre>
+ *
+ * <p>Declaring one {@code MessageConverter} bean fixes both directions &mdash; the auto-configured
+ * {@code RabbitTemplate} and the listener container factory both pick it up.
+ *
+ * <p>Note the class name. Spring AMQP 4.1 ships {@code Jackson2JsonMessageConverter} (Jackson 2)
+ * and {@code JacksonJsonMessageConverter} (Jackson 3) side by side, exactly as Spring Kafka
+ * ships {@code JsonSerializer} and {@code JacksonJsonSerializer}. Boot 4 is a Jackson 3
+ * application; pick the one without the 2.
+ *
+ * @see <a href="../../../../../docs/01-the-on-ramp.md">docs/01-the-on-ramp.md</a>
+ */
+@Configuration(proxyBeanMethods = false)
+public class ConverterConfiguration {
+
+	@Bean
+	MessageConverter messageConverter() {
+		return new JacksonJsonMessageConverter();
+	}
+
+}

rabbitmq/src/main/java/com/ankurm/rabbit/OrderMessage.java

@@ -0,0 +1,13 @@
+package com.ankurm.rabbit;
+
+/**
+ * The payload. Kept to a String field so the tests are about routing and acknowledgement rather
+ * than about converters.
+ */
+public record OrderMessage(String orderId, String detail) {
+
+	public static OrderMessage of(String orderId) {
+		return new OrderMessage(orderId, "detail for " + orderId);
+	}
+
+}

rabbitmq/src/main/java/com/ankurm/rabbit/RabbitDemoApplication.java

@@ -0,0 +1,20 @@
+package com.ankurm.rabbit;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+/**
+ * All four AMQP 0-9-1 exchange types, a working dead-letter path, and the traps that make
+ * RabbitMQ feel unpredictable until you know them.
+ *
+ * <p>Everything here runs against a real broker. See the module README for how the scripts start
+ * one without Docker.
+ */
+@SpringBootApplication
+public class RabbitDemoApplication {
+
+	public static void main(String[] args) {
+		SpringApplication.run(RabbitDemoApplication.class, args);
+	}
+
+}

rabbitmq/src/main/java/com/ankurm/rabbit/Topology.java

@@ -0,0 +1,217 @@
+package com.ankurm.rabbit;
+
+import org.springframework.amqp.core.Binding;
+import org.springframework.amqp.core.BindingBuilder;
+import org.springframework.amqp.core.DirectExchange;
+import org.springframework.amqp.core.FanoutExchange;
+import org.springframework.amqp.core.HeadersExchange;
+import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.core.QueueBuilder;
+import org.springframework.amqp.core.TopicExchange;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import java.util.Map;
+
+/**
+ * The whole topology as beans. {@code RabbitAdmin} declares every {@code Exchange},
+ * {@code Queue} and {@code Binding} bean when the connection is first opened.
+ *
+ * <p>Two things about that are worth knowing before you rely on it:
+ * <ul>
+ *   <li>Declaration is <b>idempotent only if the arguments match</b>. Redeclaring an existing
+ *       queue with different arguments is a channel-level {@code PRECONDITION_FAILED} &mdash; see
+ *       docs/06-changing-your-mind.md.</li>
+ *   <li>Declarations happen on connection, not on context refresh, so a topology error surfaces
+ *       at the first publish rather than at startup.</li>
+ * </ul>
+ *
+ * @see <a href="../../../../../docs/02-exchanges.md">docs/02-exchanges.md</a>
+ */
+@Configuration(proxyBeanMethods = false)
+public class Topology {
+
+	public static final String DIRECT = "orders.direct";
+
+	public static final String FANOUT = "orders.fanout";
+
+	public static final String TOPIC = "orders.topic";
+
+	public static final String HEADERS = "orders.headers";
+
+	public static final String DLX = "orders.dlx";
+
+	public static final String Q_NEW = "orders.new";
+
+	public static final String Q_CANCEL = "orders.cancel";
+
+	public static final String Q_AUDIT = "audit.all";
+
+	public static final String Q_ANALYTICS = "analytics.all";
+
+	public static final String Q_EU = "orders.eu";
+
+	public static final String Q_HIGH = "orders.high";
+
+	public static final String Q_PRIORITY = "orders.priority";
+
+	public static final String Q_ANY = "orders.any";
+
+	public static final String Q_WORK = "orders.work";
+
+	public static final String Q_TTL = "orders.ttl";
+
+	public static final String Q_DLQ = "orders.dlq";
+
+	// --- direct: routing key must match the binding key exactly -----------------------------
+
+	@Bean
+	DirectExchange directExchange() {
+		return new DirectExchange(DIRECT);
+	}
+
+	@Bean
+	Queue newQueue() {
+		return QueueBuilder.durable(Q_NEW).build();
+	}
+
+	@Bean
+	Queue cancelQueue() {
+		return QueueBuilder.durable(Q_CANCEL).build();
+	}
+
+	@Bean
+	Binding bindNew(DirectExchange directExchange, Queue newQueue) {
+		return BindingBuilder.bind(newQueue).to(directExchange).with("new");
+	}
+
+	@Bean
+	Binding bindCancel(DirectExchange directExchange, Queue cancelQueue) {
+		return BindingBuilder.bind(cancelQueue).to(directExchange).with("cancel");
+	}
+
+	// --- fanout: routing key ignored entirely -----------------------------------------------
+
+	@Bean
+	FanoutExchange fanoutExchange() {
+		return new FanoutExchange(FANOUT);
+	}
+
+	@Bean
+	Queue auditQueue() {
+		return QueueBuilder.durable(Q_AUDIT).build();
+	}
+
+	@Bean
+	Queue analyticsQueue() {
+		return QueueBuilder.durable(Q_ANALYTICS).build();
+	}
+
+	@Bean
+	Binding bindAudit(FanoutExchange fanoutExchange, Queue auditQueue) {
+		return BindingBuilder.bind(auditQueue).to(fanoutExchange);
+	}
+
+	@Bean
+	Binding bindAnalytics(FanoutExchange fanoutExchange, Queue analyticsQueue) {
+		return BindingBuilder.bind(analyticsQueue).to(fanoutExchange);
+	}
+
+	// --- topic: '*' is exactly one word, '#' is zero or more -------------------------------
+
+	@Bean
+	TopicExchange topicExchange() {
+		return new TopicExchange(TOPIC);
+	}
+
+	@Bean
+	Queue euQueue() {
+		return QueueBuilder.durable(Q_EU).build();
+	}
+
+	@Bean
+	Queue highQueue() {
+		return QueueBuilder.durable(Q_HIGH).build();
+	}
+
+	@Bean
+	Binding bindEu(TopicExchange topicExchange, Queue euQueue) {
+		return BindingBuilder.bind(euQueue).to(topicExchange).with("order.eu.*");
+	}
+
+	@Bean
+	Binding bindHigh(TopicExchange topicExchange, Queue highQueue) {
+		return BindingBuilder.bind(highQueue).to(topicExchange).with("order.#.high");
+	}
+
+	// --- headers: routing key ignored, header map matched ------------------------------------
+
+	@Bean
+	HeadersExchange headersExchange() {
+		return new HeadersExchange(HEADERS);
+	}
+
+	@Bean
+	Queue priorityQueue() {
+		return QueueBuilder.durable(Q_PRIORITY).build();
+	}
+
+	@Bean
+	Queue anyQueue() {
+		return QueueBuilder.durable(Q_ANY).build();
+	}
+
+	/** x-match=all: every header must match. */
+	@Bean
+	Binding bindPriority(HeadersExchange headersExchange, Queue priorityQueue) {
+		return BindingBuilder.bind(priorityQueue).to(headersExchange)
+				.whereAll(Map.of("priority", "high", "region", "eu")).match();
+	}
+
+	/** x-match=any: one is enough. */
+	@Bean
+	Binding bindAny(HeadersExchange headersExchange, Queue anyQueue) {
+		return BindingBuilder.bind(anyQueue).to(headersExchange)
+				.whereAny(Map.of("priority", "high", "region", "eu")).match();
+	}
+
+	// --- the dead-letter path ----------------------------------------------------------------
+
+	@Bean
+	DirectExchange deadLetterExchange() {
+		return new DirectExchange(DLX);
+	}
+
+	@Bean
+	Queue deadLetterQueue() {
+		// Deliberately NO dead-letter-exchange on the DLQ itself. Pointing a DLQ's DLX back at
+		// the exchange that feeds it is the classic infinite loop.
+		return QueueBuilder.durable(Q_DLQ).build();
+	}
+
+	@Bean
+	Binding bindDlq(DirectExchange deadLetterExchange, Queue deadLetterQueue) {
+		// The routing key used when dead-lettering is the message's ORIGINAL routing key unless
+		// deadLetterRoutingKey() overrides it. Both work queues below set it explicitly to
+		// "failed", so this one binding catches everything.
+		return BindingBuilder.bind(deadLetterQueue).to(deadLetterExchange).with("failed");
+	}
+
+	@Bean
+	Queue workQueue() {
+		return QueueBuilder.durable(Q_WORK)
+				.deadLetterExchange(DLX)
+				.deadLetterRoutingKey("failed")
+				.build();
+	}
+
+	@Bean
+	Queue ttlQueue() {
+		return QueueBuilder.durable(Q_TTL)
+				.ttl(1500)
+				.deadLetterExchange(DLX)
+				.deadLetterRoutingKey("failed")
+				.build();
+	}
+
+}