Introduce spring-rabbitmq-client for RabbitMQ AMQP 1.0

Fixes: #2991
Fixes: #2992
Fixes: #2993

* Add `QueueInformation.type` since RabbitMQ AMQP Client exposes such an info

Author: artembilan

build.gradle

@@ -59,6 +59,7 @@ ext {
 	micrometerVersion = '1.15.0-SNAPSHOT'
 	micrometerTracingVersion = '1.5.0-SNAPSHOT'
 	mockitoVersion = '5.15.2'
+	rabbitmqAmqpClientVersion = '0.4.0'
 	rabbitmqStreamVersion = '0.22.0'
 	rabbitmqVersion = '5.24.0'
 	reactorVersion = '2024.0.3'
@@ -472,6 +473,28 @@ project('spring-rabbit-stream') {
 	}
 }
 
+project('spring-rabbitmq-client') {
+	description = 'Spring RabbitMQ Client for AMQP 1.0'
+
+	dependencies {
+		api project(':spring-rabbit')
+		api "com.rabbitmq.client:amqp-client:$rabbitmqAmqpClientVersion"
+		api 'io.micrometer:micrometer-observation'
+
+		testApi project(':spring-rabbit-junit')
+
+		testRuntimeOnly 'com.fasterxml.jackson.core:jackson-databind'
+
+		testImplementation 'org.testcontainers:rabbitmq'
+		testImplementation 'org.testcontainers:junit-jupiter'
+		testImplementation 'org.apache.logging.log4j:log4j-slf4j-impl'
+		testImplementation 'io.micrometer:micrometer-observation-test'
+		testImplementation 'io.micrometer:micrometer-tracing-bridge-brave'
+		testImplementation 'io.micrometer:micrometer-tracing-test'
+		testImplementation 'io.micrometer:micrometer-tracing-integration-test'
+	}
+}
+
 project('spring-rabbit-junit') {
 	description = 'Spring Rabbit JUnit Support'
 

spring-amqp/src/main/java/org/springframework/amqp/core/QueueInformation.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019-2024 the original author or authors.
+ * Copyright 2019-2025 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -21,18 +21,22 @@
  *
  * @author Gary Russell
  * @author Ngoc Nhan
+ * @author Artem Bilan
+ *
  * @since 2.2
  *
  */
 public class QueueInformation {
 
 	private final String name;
 
-	private final int messageCount;
+	private final long messageCount;
 
 	private final int consumerCount;
 
-	public QueueInformation(String name, int messageCount, int consumerCount) {
+	private String type = "classic";
+
+	public QueueInformation(String name, long messageCount, int consumerCount) {
 		this.name = name;
 		this.messageCount = messageCount;
 		this.consumerCount = consumerCount;
@@ -42,19 +46,38 @@ public String getName() {
 		return this.name;
 	}
 
-	public int getMessageCount() {
+	public long getMessageCount() {
 		return this.messageCount;
 	}
 
 	public int getConsumerCount() {
 		return this.consumerCount;
 	}
 
+	/**
+	 * Return a queue type.
+	 * {@code classic} by default since AMQP 0.9.1 protocol does not return this info in {@code DeclareOk} reply.
+	 * @return a queue type
+	 * @since 4.0
+	 */
+	public String getType() {
+		return this.type;
+	}
+
+	/**
+	 * Set a queue type.
+	 * @param type the queue type: {@code quorum}, {@code classic} or {@code stream}
+	 * @since 4.0
+	 */
+	public void setType(String type) {
+		this.type = type;
+	}
+
 	@Override
 	public int hashCode() {
 		final int prime = 31;
 		int result = 1;
-		result = prime * result + ((this.name == null) ? 0 : this.name.hashCode());
+		result = prime * result + this.name.hashCode();
 		return result;
 	}
 
@@ -70,9 +93,6 @@ public boolean equals(Object obj) {
 			return false;
 		}
 		QueueInformation other = (QueueInformation) obj;
-		if (this.name == null) {
-			return other.name == null;
-		}
 		return this.name.equals(other.name);
 	}
 

spring-amqp/src/main/java/org/springframework/amqp/utils/JavaUtils.java

@@ -44,9 +44,9 @@ private JavaUtils() {
 	}
 
 	/**
-	 * Invoke {@link Consumer#accept(Object)} with the value if the condition is true.
+	 * Invoke {@link Consumer#accept(Object)} with the value if it is not null and the condition is true.
 	 * @param condition the condition.
-	 * @param value the value.
+	 * @param value the value. Skipped if null.
 	 * @param consumer the consumer.
 	 * @param <T> the value type.
 	 * @return this.

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java

@@ -119,7 +119,7 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Applicat
 	 */
 	public static final Object QUEUE_CONSUMER_COUNT = "QUEUE_CONSUMER_COUNT";
 
-	private static final String DELAYED_MESSAGE_EXCHANGE = "x-delayed-message";
+	public static final String DELAYED_MESSAGE_EXCHANGE = "x-delayed-message";
 
 	/** Logger available to subclasses. */
 	protected final Log logger = LogFactory.getLog(getClass()); // NOSONAR

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminTests.java

@@ -163,7 +163,7 @@ public void testProperties() throws Exception {
 		}
 	}
 
-	private int messageCount(RabbitAdmin rabbitAdmin, String queueName) {
+	private long messageCount(RabbitAdmin rabbitAdmin, String queueName) {
 		QueueInformation info = rabbitAdmin.getQueueInfo(queueName);
 		assertThat(info).isNotNull();
 		return info.getMessageCount();