Spring/spring-amqp
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-1419: Add RestTemplateNodeLocator

Browse Source 
- also remove hard dependency on `spring-webflux` from `spring-rabbit-junit`.

* Use reactor-netty-http.

* Fix Javadoc.
This commit is contained in:
Gary Russell
2022-10-11 14:39:39 -04:00
committed by GitHub
parent c143c5b022
commit b1ae21cad9
8 changed files with 206 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
build.gradle
View File

@@ -44,6 +44,7 @@ ext {
assertkVersion = '0.24'
awaitilityVersion = '4.2.0'
commonsCompressVersion = '1.20'
commonsHttpClientVersion = '5.1.3'
commonsPoolVersion = '2.11.1'
googleJsr305Version = '3.0.2'
hamcrestVersion = '2.2'
@@ -387,7 +388,9 @@ project('spring-rabbit') {
api 'org.springframework:spring-messaging'
api 'org.springframework:spring-tx'
optionalApi 'org.springframework:spring-webflux'
optionalApi "org.apache.httpcomponents.client5:httpclient5:$commonsHttpClientVersion"
optionalApi 'io.projectreactor:reactor-core'
optionalApi 'io.projectreactor.netty:reactor-netty-http'
optionalApi "ch.qos.logback:logback-classic:$logbackVersion"
optionalApi 'org.apache.logging.log4j:log4j-core'
optionalApi 'io.micrometer:micrometer-core'
@@ -473,6 +476,7 @@ project('spring-rabbit-stream') {
testRuntimeOnly "com.github.luben:zstd-jni:$zstdJniVersion"
testImplementation "org.testcontainers:rabbitmq:1.17.3"
testImplementation "org.apache.logging.log4j:log4j-slf4j-impl:$log4jVersion"
testImplementation 'org.springframework:spring-webflux'
}
}
@@ -488,14 +492,12 @@ project('spring-rabbit-junit') {
exclude group: 'org.hamcrest', module: 'hamcrest-core'
}
api "com.rabbitmq:amqp-client:$rabbitmqVersion"
api 'org.springframework:spring-webflux'
api 'org.springframework:spring-web'
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'
}
}

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

@@ -17,11 +17,16 @@
package org.springframework.amqp.rabbit.junit;
import java.io.IOException;
import java.net.Authenticator;
import java.net.PasswordAuthentication;
import java.net.URI;
import java.net.URISyntaxException;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.net.http.HttpResponse.BodyHandlers;
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;
@@ -33,12 +38,8 @@ import java.util.concurrent.TimeoutException;
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;
@@ -388,22 +389,35 @@ public final class BrokerRunningSupport {
}
private boolean alivenessTest() throws URISyntaxException {
WebClient client = WebClient.builder()
.filter(ExchangeFilterFunctions.basicAuthentication(this.adminUser, this.adminPassword))
HttpClient client = HttpClient.newBuilder()
.authenticator(new Authenticator() {
@Override
protected PasswordAuthentication getPasswordAuthentication() {
return new PasswordAuthentication(getAdminUser(), getAdminPassword().toCharArray());
}
})
.build();
URI uri = new URI(getAdminUri())
.resolve("/api/aliveness-test/" + UriUtils.encodePathSegment("/", StandardCharsets.UTF_8));
HashMap<String, String> result = client.get()
HttpRequest request = HttpRequest.newBuilder()
.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");
.build();
HttpResponse<String> response;
try {
response = client.send(request, BodyHandlers.ofString());
}
return false;
catch (IOException ex) {
LOGGER.error("Exception checking admin aliveness", ex);
return false;
}
catch (InterruptedException ex) {
Thread.currentThread().interrupt();
return false;
}
return response.body().contentEquals("{\"status\":\"ok\"}");
}
public static boolean fatal() {

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

@@ -35,6 +35,7 @@ import org.springframework.core.io.Resource;
import org.springframework.core.log.LogAccessor;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
/**
* A {@link RoutingConnectionFactory} that determines the node on which a queue is located and
@@ -54,6 +55,13 @@ import org.springframework.util.Assert;
*/
public class LocalizedQueueConnectionFactory implements ConnectionFactory, RoutingConnectionFactory, DisposableBean {
private static final boolean USING_WEBFLUX;
static {
USING_WEBFLUX = ClassUtils.isPresent("org.springframework.web.reactive.function.client.WebClient",
LocalizedQueueConnectionFactory.class.getClassLoader());
}
private final Log logger = LogFactory.getLog(getClass());
private final Map<String, ConnectionFactory> nodeFactories = new HashMap<String, ConnectionFactory>();
@@ -82,7 +90,7 @@ public class LocalizedQueueConnectionFactory implements ConnectionFactory, Routi
private final String trustStorePassPhrase;
private NodeLocator<?> nodeLocator = new WebFluxNodeLocator();
private NodeLocator<?> nodeLocator;
/**
* @param defaultConnectionFactory the fallback connection factory to use if the queue
@@ -190,6 +198,12 @@ public class LocalizedQueueConnectionFactory implements ConnectionFactory, Routi
this.trustStore = trustStore;
this.keyStorePassPhrase = keyStorePassPhrase;
this.trustStorePassPhrase = trustStorePassPhrase;
if (USING_WEBFLUX) {
this.nodeLocator = new WebFluxNodeLocator();
}
else {
this.nodeLocator = new RestTemplateNodeLocator();
}
}
private static Map<String, String> nodesAddressesToMap(String[] nodes, String[] addresses) {
@@ -206,6 +220,7 @@ public class LocalizedQueueConnectionFactory implements ConnectionFactory, Routi
* @since 2.4.8
*/
public void setNodeLocator(NodeLocator<?> nodeLocator) {
Assert.notNull(nodeLocator, "'nodeLocator' cannot be null");
this.nodeLocator = nodeLocator;
}
@@ -378,7 +393,7 @@ public class LocalizedQueueConnectionFactory implements ConnectionFactory, Routi
try {
String uri = new URI(adminUri)
.resolve("/api/queues/").toString();
HashMap<String, Object> queueInfo = restCall(client, uri, vhost, queue);
Map<String, Object> queueInfo = restCall(client, uri, vhost, queue);
if (queueInfo != null) {
String node = (String) queueInfo.get("node");
if (node != null) {
@@ -423,13 +438,15 @@ public class LocalizedQueueConnectionFactory implements ConnectionFactory, Routi
/**
* Retrieve a map of queue properties using the RabbitMQ Management REST API.
* @param client the client.
* @param baseUri the base uri.
* @param vhost the virtual host.
* @param queue the queue name.
* @return the map of queue properties.
* @throws URISyntaxException if the syntax is bad.
*/
HashMap<String, Object> restCall(T client, String baseUri, String vhost, String queue)
@Nullable
Map<String, Object> restCall(T client, String baseUri, String vhost, String queue)
throws URISyntaxException;
}

41
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RestTemplateHolder.java Normal file
View File

@@ -0,0 +1,41 @@
/*
* Copyright 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.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.amqp.rabbit.connection;
import org.springframework.web.client.RestTemplate;
/**
* Holder for a {@link RestTemplate} and credentials.
*
* @author Gary Russell
* @since 2.4.8
*
*/
class RestTemplateHolder {
final String userName; // NOSONAR
final String password; // NOSONAR
RestTemplate template; // NOSONAR
RestTemplateHolder(String userName, String password) {
this.userName = userName;
this.password = password;
}
}

97
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RestTemplateNodeLocator.java Normal file
View File

@@ -0,0 +1,97 @@
/*
* Copyright 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.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.amqp.rabbit.connection;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.charset.StandardCharsets;
import java.util.Map;
import org.apache.hc.client5.http.auth.AuthCache;
import org.apache.hc.client5.http.impl.auth.BasicAuthCache;
import org.apache.hc.client5.http.impl.auth.BasicScheme;
import org.apache.hc.client5.http.protocol.HttpClientContext;
import org.apache.hc.core5.http.HttpHost;
import org.apache.hc.core5.http.protocol.BasicHttpContext;
import org.apache.hc.core5.http.protocol.HttpContext;
import org.springframework.amqp.rabbit.connection.LocalizedQueueConnectionFactory.NodeLocator;
import org.springframework.http.HttpMethod;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.http.client.HttpComponentsClientHttpRequestFactory;
import org.springframework.http.client.support.BasicAuthenticationInterceptor;
import org.springframework.lang.Nullable;
import org.springframework.web.client.RestTemplate;
import org.springframework.web.util.UriUtils;
/**
* A {@link NodeLocator} using the {@link RestTemplate}.
*
* @author Gary Russell
* @since 3.0
*
*/
public class RestTemplateNodeLocator implements NodeLocator<RestTemplateHolder> {
@Override
public RestTemplateHolder createClient(String userName, String password) {
return new RestTemplateHolder(userName, password);
}
@SuppressWarnings({ "unchecked", "rawtypes" })
@Override
@Nullable
public Map<String, Object> restCall(RestTemplateHolder client, String baseUri, String vhost, String queue)
throws URISyntaxException {
if (client.template == null) {
URI uri = new URI(baseUri);
HttpHost host = new HttpHost(uri.getHost(), uri.getPort());
client.template = new RestTemplate(new HttpComponentsClientHttpRequestFactory() {
@Override
@Nullable
protected HttpContext createHttpContext(HttpMethod httpMethod, URI uri) {
AuthCache cache = new BasicAuthCache();
BasicScheme scheme = new BasicScheme();
cache.put(host, scheme);
BasicHttpContext context = new BasicHttpContext();
context.setAttribute(HttpClientContext.AUTH_CACHE, cache);
return context;
}
});
client.template.getInterceptors().add(new BasicAuthenticationInterceptor(client.userName, client.password));
}
URI uri = new URI(baseUri)
.resolve("/api/queues/" + UriUtils.encodePathSegment(vhost, StandardCharsets.UTF_8) + "/" + queue);
ResponseEntity<Map> response = client.template.exchange(uri, HttpMethod.GET, null, Map.class);
return response.getStatusCode().equals(HttpStatus.OK) ? response.getBody() : null;
}
@Override
public void close(RestTemplateHolder client) {
try {
client.template.close();
}
catch (IOException e) {
}
}
}

5
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/WebFluxNodeLocator.java
View File

@@ -21,10 +21,12 @@ import java.net.URISyntaxException;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.HashMap;
import java.util.Map;
import org.springframework.amqp.rabbit.connection.LocalizedQueueConnectionFactory.NodeLocator;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.http.MediaType;
import org.springframework.lang.Nullable;
import org.springframework.web.reactive.function.client.ExchangeFilterFunctions;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.web.util.UriUtils;
@@ -39,7 +41,8 @@ import org.springframework.web.util.UriUtils;
public class WebFluxNodeLocator implements NodeLocator<WebClient> {
@Override
public HashMap<String, Object> restCall(WebClient client, String baseUri, String vhost, String queue)
@Nullable
public Map<String, Object> restCall(WebClient client, String baseUri, String vhost, String queue)
throws URISyntaxException {
URI uri = new URI(baseUri)

3
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/LocalizedQueueConnectionFactoryIntegrationTests.java
View File

@@ -83,6 +83,9 @@ public class LocalizedQueueConnectionFactoryIntegrationTests {
ConnectionFactory cf = lqcf.getTargetConnectionFactory("[local]");
RabbitAdmin admin = new RabbitAdmin(cf);
assertThat(admin.getQueueProperties("local")).isNotNull();
lqcf.setNodeLocator(new RestTemplateNodeLocator());
ConnectionFactory cf2 = lqcf.getTargetConnectionFactory("[local]");
assertThat(cf2).isSameAs(cf);
lqcf.destroy();
}

8
src/reference/asciidoc/amqp.adoc
View File

@@ -796,7 +796,9 @@ 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; which is not added to the class path by default.
Instead, by default, the `WebClient` from Spring Webflux is used if `spring-webflux` is on the class path; otherwise a `RestTemplate` is used.
To add `WebFlux` to the class path:
.Maven
====
@@ -821,7 +823,7 @@ You can also use other REST technology by implementing `LocalizedQueueConnection
====
[source, java]
----
lqcf.setNodeLocator(new DefaultNodeLocator<MyClient>() {
lqcf.setNodeLocator(new NodeLocator<MyClient>() {
@Override
public MyClient createClient(String userName, String password) {
@@ -837,6 +839,8 @@ lqcf.setNodeLocator(new DefaultNodeLocator<MyClient>() {
----
====
The framework provides the `WebFluxNodeLocator` and `RestTemplateNodeLocator`, with the default as discussed above.
[[cf-pub-conf-ret]]
===== Publisher Confirms and Returns