Rename DefaultNodeLocator; add generics.

Author: garyrussell

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/LocalizedQueueConnectionFactory.java

@@ -83,7 +83,7 @@ public class LocalizedQueueConnectionFactory implements ConnectionFactory, Routi
 
 	private final String trustStorePassPhrase;
 
-	private NodeLocator nodeLocator = new DefaultNodeLocator();
+	private NodeLocator<?> nodeLocator = new WebFluxNodeLocator();
 
 	/**
 	 * @param defaultConnectionFactory the fallback connection factory to use if the queue
@@ -206,7 +206,7 @@ private static Map<String, String> nodesAddressesToMap(String[] nodes, String[]
 	 * @param nodeLocator the locator.
 	 * @since 2.4.8
 	 */
-	public void setNodeLocator(NodeLocator nodeLocator) {
+	public void setNodeLocator(NodeLocator<?> nodeLocator) {
 		this.nodeLocator = nodeLocator;
 	}
 
@@ -254,7 +254,8 @@ public void clearConnectionListeners() {
 	public ConnectionFactory getTargetConnectionFactory(Object key) {
 		String queue = ((String) key);
 		queue = queue.substring(1, queue.length() - 1);
-		Assert.isTrue(!queue.contains(","), () -> "Cannot use LocalizedQueueConnectionFactory with more than one queue: " + key);
+		Assert.isTrue(!queue.contains(","),
+				() -> "Cannot use LocalizedQueueConnectionFactory with more than one queue: " + key);
 		ConnectionFactory connectionFactory = determineConnectionFactory(queue);
 		if (connectionFactory == null) {
 			return this.defaultConnectionFactory;
@@ -345,10 +346,11 @@ public void destroy() {
 
 	/**
 	 * Used to obtain a connection factory for the queue leader.
+	 *
+	 * @param <T> the client type.
 	 * @since 2.4.8
 	 */
-	@FunctionalInterface
-	public interface NodeLocator {
+	public interface NodeLocator<T> {
 
 		LogAccessor LOGGER = new LogAccessor(LogFactory.getLog(NodeLocator.class));
 
@@ -364,8 +366,10 @@ public interface NodeLocator {
 		 * @return a connection factory, if the leader node was found; null otherwise.
 		 */
 		@Nullable
-		default ConnectionFactory locate(String[] adminUris, Map<String, String> nodeToAddress, String vhost, String username,
-				String password, String queue, FactoryFinder factoryFunction) {
+		default ConnectionFactory locate(String[] adminUris, Map<String, String> nodeToAddress, String vhost,
+				String username, String password, String queue, FactoryFinder factoryFunction) {
+
+			T client = createClient(username, password);
 
 			for (int i = 0; i < adminUris.length; i++) {
 				String adminUri = adminUris[i];
@@ -376,12 +380,13 @@ default ConnectionFactory locate(String[] adminUris, Map<String, String> nodeToA
 					URI uri = new URI(adminUri)
 							.resolve("/api/queues/" + UriUtils.encodePathSegment(vhost, StandardCharsets.UTF_8) + "/"
 									+ queue);
-					HashMap<String, Object> queueInfo = restCall(username, password, uri);
+					HashMap<String, Object> queueInfo = restCall(client, uri);
 					if (queueInfo != null) {
 						String node = (String) queueInfo.get("node");
 						if (node != null) {
 							String nodeUri = nodeToAddress.get(node);
 							if (uri != null) {
+								close(client);
 								return factoryFunction.locate(queue, node, nodeUri);
 							}
 							if (LOGGER.isDebugEnabled()) {
@@ -399,17 +404,31 @@ default ConnectionFactory locate(String[] adminUris, Map<String, String> nodeToA
 				}
 			}
 			LOGGER.warn("Failed to determine queue location for: " + queue + ", using default connection factory");
+			close(client);
 			return null;
 		}
 
 		/**
-		 * Retrieve a map of queue properties using the RabbitMQ Management REST API.
-		 * @param username the user name.
+		 * Create a client for subsequent use.
+		 * @param userName the user name.
 		 * @param password the password.
+		 * @return the client.
+		 */
+		T createClient(String userName, String password);
+
+		/**
+		 * Close the client.
+		 * @param client the client.
+		 */
+		default void close(T client) {
+		}
+
+		/**
+		 * Retrieve a map of queue properties using the RabbitMQ Management REST API.
 		 * @param uri the uri.
 		 * @return the map of queue properties.
 		 */
-		HashMap<String, Object> restCall(String username, String password, URI uri);
+		HashMap<String, Object> restCall(T client, URI uri);
 
 	}
 

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/DefaultNodeLocator.java renamed to spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/WebFluxNodeLocator.java

@@ -27,17 +27,16 @@
 import org.springframework.web.reactive.function.client.WebClient;
 
 /**
- * Default {@link NodeLocator} using the Spring WebFlux {@link WebClient}.
+ * A {@link NodeLocator} using the Spring WebFlux {@link WebClient}.
  *
  * @author Gary Russell
  * @since 2.4.8
  *
  */
-public class DefaultNodeLocator implements NodeLocator {
+public class WebFluxNodeLocator implements NodeLocator<WebClient> {
 
 	@Override
-	public HashMap<String, Object> restCall(String username, String password, URI uri) {
-		WebClient client = createClient(username, password);
+	public HashMap<String, Object> restCall(WebClient client, URI uri) {
 		HashMap<String, Object> queueInfo = client.get()
 				.uri(uri)
 				.accept(MediaType.APPLICATION_JSON)
@@ -54,7 +53,8 @@ public HashMap<String, Object> restCall(String username, String password, URI ur
 	 * @param password the password.
 	 * @return The client.
 	 */
-	protected WebClient createClient(String username, String password) {
+	@Override
+	public WebClient createClient(String username, String password) {
 		return WebClient.builder()
 				.filter(ExchangeFilterFunctions.basicAuthentication(username, password))
 				.build();

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/LocalizedQueueConnectionFactoryTests.java

@@ -100,10 +100,10 @@ protected ConnectionFactory createConnectionFactory(String address, String node)
 			}
 
 		};
-		lqcf.setNodeLocator(new DefaultNodeLocator() {
+		lqcf.setNodeLocator(new WebFluxNodeLocator() {
 
 			@Override
-			protected WebClient createClient(String username, String password) {
+			public WebClient createClient(String username, String password) {
 				return firstServer.get() ? client1 : client2;
 			}
 

src/reference/asciidoc/amqp.adoc

@@ -796,7 +796,7 @@ Notice that the first three parameters are arrays of `addresses`, `adminUris`, a
 These are positional in that, when a container attempts to connect to a queue, it uses the admin API to determine which node is the lead for the queue and connects to the address in the same array position as that node.
 
 IMPORTANT: Starting with version 3.0, the RabbitMQ `http-client` is no longer used to access the Rest API.
-Instead, by default, the `WebClient` from Spring Webflux is used; it is not added to the class path by default.
+Instead, by default, the `WebClient` from Spring Webflux is used; which is not added to the class path by default.
 
 .Maven
 ====
@@ -816,17 +816,23 @@ compile 'org.springframework.amqp:spring-rabbit'
 ----
 ====
 
-You can also use other REST technology by extending `DefaultNodeLocator` and overriding its `restCall` method.
+You can also use other REST technology by implementing `LocalizedQueueConnectionFactory.NodeLocator` and overriding its `createClient, ``restCall`, and optionally, `close` methods.
 
 ====
 [source, java]
 ----
-lqcf.setNodeLocator(new DefaultNodeLocator() {
+lqcf.setNodeLocator(new DefaultNodeLocator<MyClient>() {
 
     @Override
-    public HashMap<String, Object> restCall(String username, String password, URI uri) {
+    public MyClient createClient(String userName, String password) {
+        ...
+    }
+
+    @Override
+    public HashMap<String, Object> restCall(MyClient client, URI uri) {
         ...
     });
+
 });
 ----
 ====