GH-1419: Remove RabbitMQ http-client Usage

Resolves #1419

Use Spring WebFlux instead, while allowing the user to choose some other technology
in the `LocalizedQueueConnectionFactory`.

* Rename DefaultNodeLocator; add generics.
* Remove unnecessary dependencies.

Author: garyrussell

build.gradle

@@ -44,7 +44,6 @@ ext {
 	assertkVersion = '0.24'
 	awaitilityVersion = '4.2.0'
 	commonsCompressVersion = '1.20'
-	commonsHttpClientVersion = '4.5.13'
 	commonsPoolVersion = '2.11.1'
 	googleJsr305Version = '3.0.2'
 	hamcrestVersion = '2.2'
@@ -62,7 +61,6 @@ ext {
 	mockitoVersion = '4.8.0'
 	rabbitmqStreamVersion = '0.8.0'
 	rabbitmqVersion = project.hasProperty('rabbitmqVersion') ? project.rabbitmqVersion : '5.16.0'
-	rabbitmqHttpClientVersion = '3.12.1'
 	reactorVersion = '2022.0.0-SNAPSHOT'
 	snappyVersion = '1.1.8.4'
 	springDataVersion = '2022.0.0-SNAPSHOT'
@@ -384,11 +382,11 @@ project('spring-rabbit') {
 
 		api project(':spring-amqp')
 		api "com.rabbitmq:amqp-client:$rabbitmqVersion"
-		optionalApi "com.rabbitmq:http-client:$rabbitmqHttpClientVersion"
 		optionalApi 'org.springframework:spring-aop'
 		api 'org.springframework:spring-context'
 		api 'org.springframework:spring-messaging'
 		api 'org.springframework:spring-tx'
+		optionalApi 'org.springframework:spring-webflux'
 		optionalApi 'io.projectreactor:reactor-core'
 		optionalApi "ch.qos.logback:logback-classic:$logbackVersion"
 		optionalApi 'org.apache.logging.log4j:log4j-core'
@@ -409,8 +407,6 @@ project('spring-rabbit') {
 		testImplementation 'io.micrometer:micrometer-tracing-bridge-brave'
 		testImplementation 'io.micrometer:micrometer-tracing-test'
 		testImplementation 'io.micrometer:micrometer-tracing-integration-test'
-		testRuntimeOnly 'org.springframework:spring-web'
-		testRuntimeOnly "org.apache.httpcomponents:httpclient:$commonsHttpClientVersion"
 		testRuntimeOnly 'com.fasterxml.jackson.core:jackson-core'
 		testRuntimeOnly 'com.fasterxml.jackson.core:jackson-databind'
 		testRuntimeOnly 'com.fasterxml.jackson.dataformat:jackson-dataformat-xml'
@@ -465,14 +461,12 @@ project('spring-rabbit-stream') {
 
 		api project(':spring-rabbit')
 		api "com.rabbitmq:stream-client:$rabbitmqStreamVersion"
-		optionalApi "com.rabbitmq:http-client:$rabbitmqHttpClientVersion"
 
 		testApi project(':spring-rabbit-junit')
 		testRuntimeOnly 'com.fasterxml.jackson.core:jackson-core'
 		testRuntimeOnly 'com.fasterxml.jackson.core:jackson-databind'
 		testRuntimeOnly 'com.fasterxml.jackson.dataformat:jackson-dataformat-xml'
 		testRuntimeOnly 'com.fasterxml.jackson.module:jackson-module-kotlin'
-		testRuntimeOnly "org.apache.httpcomponents:httpclient:$commonsHttpClientVersion"
 		testRuntimeOnly "org.apache.commons:commons-compress:$commonsCompressVersion"
 		testRuntimeOnly "org.xerial.snappy:snappy-java:$snappyVersion"
 		testRuntimeOnly "org.lz4:lz4-java:$lz4Version"
@@ -494,16 +488,14 @@ project('spring-rabbit-junit') {
 			exclude group: 'org.hamcrest', module: 'hamcrest-core'
 		}
 		api "com.rabbitmq:amqp-client:$rabbitmqVersion"
-		api ("com.rabbitmq:http-client:$rabbitmqHttpClientVersion") {
-			exclude group: 'org.springframework', module: 'spring-web'
-		}
-		api 'org.springframework:spring-web'
+		api 'org.springframework:spring-webflux'
 		api 'org.junit.jupiter:junit-jupiter-api'
 		api "org.assertj:assertj-core:$assertjVersion"
 		optionalApi "ch.qos.logback:logback-classic:$logbackVersion"
 		optionalApi 'org.apache.logging.log4j:log4j-core'
 		compileOnly 'org.apiguardian:apiguardian-api:1.0.0'
-
+		testRuntimeOnly 'com.fasterxml.jackson.core:jackson-core'
+		testRuntimeOnly 'com.fasterxml.jackson.core:jackson-databind'
 	}
 
 }

spring-rabbit-junit/src/main/java/org/springframework/amqp/rabbit/junit/BrokerRunningSupport.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2020 the original author or authors.
+ * Copyright 2002-2022 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.
@@ -17,8 +17,11 @@
 package org.springframework.amqp.rabbit.junit;
 
 import java.io.IOException;
+import java.net.URI;
 import java.net.URISyntaxException;
 import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashMap;
@@ -30,13 +33,17 @@
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
+import org.springframework.core.ParameterizedTypeReference;
+import org.springframework.http.MediaType;
 import org.springframework.util.Base64Utils;
 import org.springframework.util.StringUtils;
+import org.springframework.web.reactive.function.client.ExchangeFilterFunctions;
+import org.springframework.web.reactive.function.client.WebClient;
+import org.springframework.web.util.UriUtils;
 
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ConnectionFactory;
-import com.rabbitmq.http.client.Client;
 
 /**
  * A class that can be used to prevent integration tests from failing if the Rabbit broker application is
@@ -372,15 +379,33 @@ private Channel createQueues(Connection connection) throws IOException, URISynta
 			}
 		}
 		if (this.management) {
-			Client client = new Client(getAdminUri(), this.adminUser, this.adminPassword);
-			if (!client.alivenessTest("/")) {
+			if (!alivenessTest()) {
 				throw new BrokerNotAliveException("Aliveness test failed for localhost:15672 guest/quest; "
 						+ "management not available");
 			}
 		}
 		return channel;
 	}
 
+	private boolean alivenessTest() throws URISyntaxException {
+		WebClient client = WebClient.builder()
+				.filter(ExchangeFilterFunctions.basicAuthentication(this.adminUser, this.adminPassword))
+				.build();
+		URI uri = new URI(getAdminUri())
+				.resolve("/api/aliveness-test/" + UriUtils.encodePathSegment("/", StandardCharsets.UTF_8));
+		HashMap<String, String> result = client.get()
+				.uri(uri)
+				.accept(MediaType.APPLICATION_JSON)
+				.retrieve()
+				.bodyToMono(new ParameterizedTypeReference<HashMap<String, String>>() {
+				})
+				.block(Duration.ofSeconds(10)); // NOSONAR magic#
+		if (result != null) {
+			return result.get("status").equals("ok");
+		}
+		return false;
+	}
+
 	public static boolean fatal() {
 		String serversRequired = System.getenv(BROKER_REQUIRED);
 		if (Boolean.parseBoolean(serversRequired)) {

spring-rabbit-junit/src/test/java/org/springframework/amqp/rabbit/junit/RabbitAvailableTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2019 the original author or authors.
+ * Copyright 2017-2022 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.
@@ -30,7 +30,7 @@
  * @since 2.0.2
  *
  */
-@RabbitAvailable(queues = "rabbitAvailableTests.queue")
+@RabbitAvailable(queues = "rabbitAvailableTests.queue", management = true)
 public class RabbitAvailableTests {
 
 	@Test

spring-rabbit-stream/src/test/java/org/springframework/rabbit/stream/listener/RabbitListenerTests.java

@@ -18,14 +18,18 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
-import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 
 import org.springframework.amqp.core.Queue;
@@ -42,6 +46,8 @@
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.DependsOn;
+import org.springframework.core.ParameterizedTypeReference;
+import org.springframework.http.MediaType;
 import org.springframework.rabbit.stream.config.StreamRabbitListenerContainerFactory;
 import org.springframework.rabbit.stream.producer.RabbitStreamTemplate;
 import org.springframework.rabbit.stream.retry.StreamRetryOperationsInterceptorFactoryBean;
@@ -50,9 +56,10 @@
 import org.springframework.retry.interceptor.RetryOperationsInterceptor;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+import org.springframework.web.reactive.function.client.ExchangeFilterFunctions;
+import org.springframework.web.reactive.function.client.WebClient;
+import org.springframework.web.util.UriUtils;
 
-import com.rabbitmq.http.client.Client;
-import com.rabbitmq.http.client.domain.QueueInfo;
 import com.rabbitmq.stream.Address;
 import com.rabbitmq.stream.Environment;
 import com.rabbitmq.stream.Message;
@@ -99,12 +106,38 @@ void nativeMsg(@Autowired RabbitTemplate template) throws InterruptedException {
 		assertThat(this.config.latch4.await(10, TimeUnit.SECONDS)).isTrue();
 	}
 
+	@SuppressWarnings("unchecked")
 	@Test
-	@Disabled("Temporary until SF uses Micrometer snaps")
 	void queueOverAmqp() throws Exception {
-		Client client = new Client("http://guest:guest@localhost:" + managementPort() + "/api");
-		QueueInfo queue = client.getQueue("/", "stream.created.over.amqp");
-		assertThat(queue.getArguments().get("x-queue-type")).isEqualTo("stream");
+		WebClient client = WebClient.builder()
+				.filter(ExchangeFilterFunctions.basicAuthentication("guest", "guest"))
+				.build();
+		Map<String, Object> queue = queueInfo("stream.created.over.amqp");
+		assertThat(((Map<String, Object>) queue.get("arguments")).get("x-queue-type")).isEqualTo("stream");
+	}
+
+	private Map<String, Object> queueInfo(String queueName) throws URISyntaxException {
+		WebClient client = createClient("guest", "guest");
+		URI uri = queueUri(queueName);
+		return client.get()
+				.uri(uri)
+				.accept(MediaType.APPLICATION_JSON)
+				.retrieve()
+				.bodyToMono(new ParameterizedTypeReference<Map<String, Object>>() {
+				})
+				.block(Duration.ofSeconds(10));
+	}
+
+	private URI queueUri(String queue) throws URISyntaxException {
+		URI uri = new URI("http://localhost:" + managementPort() + "/api")
+				.resolve("/api/queues/" + UriUtils.encodePathSegment("/", StandardCharsets.UTF_8) + "/" + queue);
+		return uri;
+	}
+
+	private WebClient createClient(String adminUser, String adminPassword) {
+		return WebClient.builder()
+				.filter(ExchangeFilterFunctions.basicAuthentication(adminUser, adminPassword))
+				.build();
 	}
 
 	@Configuration(proxyBeanMethods = false)