SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Remove Usage of RabbitMQ http-client

Browse Source 
Replace with `WebClient`.
This commit is contained in:
Gary Russell
2022-10-12 12:16:55 -04:00
committed by Oleg Zhurakousky
parent 9039bdc5e5
commit 48c8550753
6 changed files with 389 additions and 143 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/pom.xml
View File

@@ -47,13 +47,22 @@
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.httpcomponents</groupId>
<artifactId>httpclient</artifactId>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-webflux</artifactId>
</dependency>
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>http-client</artifactId>
<version>2.1.0.RELEASE</version>
<groupId>io.projectreactor.netty</groupId>
<artifactId>reactor-netty</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
</project>

145
binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/admin/RabbitBindingCleaner.java
View File

@@ -16,8 +16,10 @@
package org.springframework.cloud.stream.binder.rabbit.admin;
import java.net.MalformedURLException;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
@@ -26,15 +28,15 @@ import java.util.Map;
import java.util.Map.Entry;
import java.util.stream.Collectors;
import com.rabbitmq.http.client.Client;
import com.rabbitmq.http.client.domain.BindingInfo;
import com.rabbitmq.http.client.domain.ExchangeInfo;
import com.rabbitmq.http.client.domain.QueueInfo;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.cloud.stream.binder.AbstractBinder;
import org.springframework.cloud.stream.binder.BindingCleaner;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.web.reactive.function.client.ExchangeFilterFunctions;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.web.util.UriUtils;
/**
* Implementation of {@link org.springframework.cloud.stream.binder.BindingCleaner} for
@@ -65,28 +67,31 @@ public class RabbitBindingCleaner implements BindingCleaner {
String vhost, String binderPrefix, String entity, boolean isJob) {
try {
Client client = new Client(adminUri, user, pw);
return doClean(client,
WebClient client = WebClient.builder()
.filter(ExchangeFilterFunctions.basicAuthentication(user, pw))
.build();
URI uri = new URI(adminUri);
return doClean(client, uri,
vhost == null ? "/" : vhost,
binderPrefix == null ? BINDER_PREFIX : binderPrefix, entity, isJob);
}
catch (MalformedURLException | URISyntaxException e) {
catch (URISyntaxException e) {
throw new RabbitAdminException("Couldn't create a Client", e);
}
}
private Map<String, List<String>> doClean(Client client,
String vhost, String binderPrefix, String entity, boolean isJob) {
private Map<String, List<String>> doClean(WebClient client,
URI uri, String vhost, String binderPrefix, String entity, boolean isJob) {
LinkedList<String> removedQueues = isJob ? null
: findStreamQueues(client, vhost, binderPrefix, entity);
List<String> removedExchanges = findExchanges(client, vhost, binderPrefix, entity);
: findStreamQueues(client, uri, vhost, binderPrefix, entity);
List<String> removedExchanges = findExchanges(client, uri, vhost, binderPrefix, entity);
// Delete the queues in reverse order to enable re-running after a partial
// success.
// The queue search above starts with 0 and terminates on a not found.
if (removedQueues != null) {
removedQueues.descendingIterator().forEachRemaining(q -> {
client.deleteQueue(vhost, q);
deleteQueue(client, uri, vhost, q);
if (logger.isDebugEnabled()) {
logger.debug("deleted queue: " + q);
}
@@ -98,7 +103,7 @@ public class RabbitBindingCleaner implements BindingCleaner {
}
// Fanout exchanges for taps
removedExchanges.forEach(exchange -> {
client.deleteExchange(vhost, exchange);
deleteExchange(client, uri, vhost, exchange);
if (logger.isDebugEnabled()) {
logger.debug("deleted exchange: " + exchange);
}
@@ -109,15 +114,48 @@ public class RabbitBindingCleaner implements BindingCleaner {
return results;
}
private LinkedList<String> findStreamQueues(Client client, String vhost, String binderPrefix, String stream) {
private void deleteQueue(WebClient client, URI uri, String vhost, String q) {
URI deleteURI = uri
.resolve("/api/queues/" + UriUtils.encodePathSegment(vhost, StandardCharsets.UTF_8) + "/" + q);
client.delete()
.uri(deleteURI)
.retrieve()
.toEntity(Void.class)
.block(Duration.ofSeconds(10));
}
private void deleteExchange(WebClient client, URI uri, String vhost, String ex) {
URI deleteURI = uri
.resolve("/api/exchanges/" + UriUtils.encodePathSegment(vhost, StandardCharsets.UTF_8) + "/" + ex);
client.delete()
.uri(deleteURI)
.retrieve()
.toEntity(Void.class)
.block(Duration.ofSeconds(10));
}
private LinkedList<String> findStreamQueues(WebClient client, URI uri, String vhost, String binderPrefix,
String stream) {
String queueNamePrefix = adjustPrefix(AbstractBinder.applyPrefix(binderPrefix, stream));
List<QueueInfo> queues = client.getQueues(vhost);
List<Map<String, Object>> queues = getQueues(client, uri, vhost);
return queues.stream()
.filter(q -> q.getName().startsWith(queueNamePrefix))
.filter(q -> ((String) q.get("name")).startsWith(queueNamePrefix))
.map(q -> checkNoConsumers(q))
.collect(Collectors.toCollection(LinkedList::new));
}
private List<Map<String, Object>> getQueues(WebClient client, URI uri, String vhost) {
URI getUri = uri
.resolve("/api/queues/" + UriUtils.encodePathSegment(vhost, StandardCharsets.UTF_8) + "/");
return client.get()
.uri(getUri)
.retrieve()
.bodyToMono(new ParameterizedTypeReference<List<Map<String, Object>>>() {
})
.block(Duration.ofSeconds(10));
}
private String adjustPrefix(String prefix) {
if (prefix.endsWith("*")) {
return prefix.substring(0, prefix.length() - 1);
@@ -127,41 +165,80 @@ public class RabbitBindingCleaner implements BindingCleaner {
}
}
private String checkNoConsumers(QueueInfo queue) {
if (queue.getConsumerCount() != 0) {
throw new RabbitAdminException("Queue " + queue.getName() + " is in use");
private String checkNoConsumers(Map<String, Object> queue) {
if ((Integer) queue.get("consumers") != 0) {
throw new RabbitAdminException("Queue " + queue.get("name") + " is in use");
}
return queue.getName();
return (String) queue.get("name");
}
private List<String> findExchanges(Client client, String vhost, String binderPrefix, String entity) {
List<ExchangeInfo> exchanges = client.getExchanges(vhost);
private List<String> findExchanges(WebClient client, URI uri, String vhost, String binderPrefix, String entity) {
List<Map<String, Object>> exchanges = getExchanges(client, uri, vhost);
String exchangeNamePrefix = adjustPrefix(AbstractBinder.applyPrefix(binderPrefix, entity));
List<String> exchangesToRemove = exchanges.stream()
.filter(e -> e.getName().startsWith(exchangeNamePrefix))
.filter(e -> ((String) e.get("name")).startsWith(exchangeNamePrefix))
.map(e -> {
System.out.println(e.getName());
List<BindingInfo> bindingsBySource = client.getBindingsBySource(vhost, e.getName());
return Collections.singletonMap(e.getName(), bindingsBySource);
List<Map<String, Object>> bindingsBySource =
getBindingsBySource(client, uri, vhost, (String) e.get("name"));
return Collections.singletonMap((String) e.get("name"), bindingsBySource);
})
.map(bindingsMap -> hasNoForeignBindings(bindingsMap, exchangeNamePrefix))
.collect(Collectors.toList());
exchangesToRemove.stream()
.map(exchange -> client.getExchangeBindingsByDestination(vhost, exchange))
.map(exchange -> getExchangeBindingsByDestination(client, uri, vhost, exchange))
.forEach(bindings -> {
if (bindings.size() > 0) {
throw new RabbitAdminException("Cannot delete exchange "
+ bindings.get(0).getDestination() + "; it is a destination: " + bindings);
+ bindings.get(0).get("destination") + "; it is a destination: " + bindings);
}
});
return exchangesToRemove;
}
private String hasNoForeignBindings(Map<String, List<BindingInfo>> bindings, String exchangeNamePrefix) {
Entry<String, List<BindingInfo>> next = bindings.entrySet().iterator().next();
for (BindingInfo binding : next.getValue()) {
if (!"queue".equals(binding.getDestinationType())
|| !binding.getDestination().startsWith(exchangeNamePrefix)) {
private List<Map<String, Object>> getExchangeBindingsByDestination(WebClient client, URI uri, String vhost,
String name) {
String exchange = "".equals(name) ? "amq.default" : name;
URI getUri = uri
.resolve("/api/exchanges/" + UriUtils.encodePathSegment(vhost, StandardCharsets.UTF_8) + "/"
+ UriUtils.encodePathSegment(exchange, StandardCharsets.UTF_8) + "/bindings/destination");
return client.get()
.uri(getUri)
.retrieve()
.bodyToMono(new ParameterizedTypeReference<List<Map<String, Object>>>() {
})
.block(Duration.ofSeconds(10));
}
private List<Map<String, Object>> getBindingsBySource(WebClient client, URI uri, String vhost, String name) {
String exchange = "".equals(name) ? "amq.default" : name;
URI getUri = uri
.resolve("/api/exchanges/" + UriUtils.encodePathSegment(vhost, StandardCharsets.UTF_8) + "/"
+ UriUtils.encodePathSegment(exchange, StandardCharsets.UTF_8) + "/bindings/source");
return client.get()
.uri(getUri)
.retrieve()
.bodyToMono(new ParameterizedTypeReference<List<Map<String, Object>>>() {
})
.block(Duration.ofSeconds(10));
}
private List<Map<String, Object>> getExchanges(WebClient client, URI uri, String vhost) {
URI getUri = uri
.resolve("/api/exchanges/" + UriUtils.encodePathSegment(vhost, StandardCharsets.UTF_8) + "/");
return client.get()
.uri(getUri)
.retrieve()
.bodyToMono(new ParameterizedTypeReference<List<Map<String, Object>>>() {
})
.block(Duration.ofSeconds(10));
}
private String hasNoForeignBindings(Map<String, List<Map<String, Object>>> bindings, String exchangeNamePrefix) {
Entry<String, List<Map<String, Object>>> next = bindings.entrySet().iterator().next();
for (Map<String, Object> binding : next.getValue()) {
if (!"queue".equals(binding.get("destination_type"))
|| !((String) binding.get("destination")).startsWith(exchangeNamePrefix)) {
throw new RabbitAdminException("Cannot delete exchange "
+ next.getKey() + "; it has bindings: " + bindings);
}

61
binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderCleanerTests.java
View File

@@ -16,16 +16,16 @@
package org.springframework.cloud.stream.binder.rabbit;
import java.net.MalformedURLException;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.http.client.Client;
import com.rabbitmq.http.client.domain.QueueInfo;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
import org.testcontainers.containers.RabbitMQContainer;
@@ -42,6 +42,10 @@ import org.springframework.cloud.stream.binder.AbstractBinder;
import org.springframework.cloud.stream.binder.rabbit.admin.RabbitAdminException;
import org.springframework.cloud.stream.binder.rabbit.admin.RabbitBindingCleaner;
import org.springframework.cloud.stream.binder.test.junit.rabbit.RabbitTestSupport;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.web.reactive.function.client.ExchangeFilterFunctions;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.web.util.UriUtils;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.Assert.fail;
@@ -57,18 +61,18 @@ public class RabbitBinderCleanerTests {
private static final String BINDER_PREFIX = "binder.";
private static final Client client;
private static final WebClient client;
static {
try {
client = new Client(RABBITMQ.getHttpUrl() + "/api", "guest", "guest");
}
catch (MalformedURLException | URISyntaxException e) {
throw new RabbitAdminException("Couldn't create a Client", e);
}
client = WebClient.builder()
.filter(ExchangeFilterFunctions
.basicAuthentication(RABBITMQ.getAdminUsername(), RABBITMQ.getAdminPassword()))
.build();
}
@RegisterExtension
private RabbitTestSupport rabbitTestSupport = new RabbitTestSupport(true, RABBITMQ.getAmqpPort(), RABBITMQ.getHttpPort());
private final RabbitTestSupport rabbitTestSupport = new RabbitTestSupport(true, RABBITMQ.getAmqpPort(),
RABBITMQ.getHttpPort());
@Test
public void testCleanStream() {
@@ -121,7 +125,7 @@ public class RabbitBinderCleanerTests {
String consumerTag = channel.basicConsume(queueName,
new DefaultConsumer(channel));
try {
waitForConsumerStateNot(queueName, 0);
waitForConsumerState(queueName, 1);
doClean(cleaner, stream1, false);
fail("Expected exception");
}
@@ -130,7 +134,7 @@ public class RabbitBinderCleanerTests {
.hasMessageContaining("Queue " + queueName + " is in use");
}
channel.basicCancel(consumerTag);
waitForConsumerStateNot(queueName, 1);
waitForConsumerState(queueName, 0);
try {
doClean(cleaner, stream1, false);
fail("Expected exception");
@@ -142,18 +146,27 @@ public class RabbitBinderCleanerTests {
return null;
}
private void waitForConsumerStateNot(String queueName, long state) throws InterruptedException {
private void waitForConsumerState(String queueName, long state)
throws InterruptedException, URISyntaxException {
int n = 0;
QueueInfo queue = client.getQueue("/", queueName);
while (n++ < 100 && (queue == null || queue.getConsumerCount() == state)) {
Map<String, Object> queue = getQueue("/", queueName);
while (n++ < 100 && !requiredState(state, queue)) {
Thread.sleep(100);
queue = client.getQueue("/", queueName);
queue = getQueue("/", queueName);
}
assertThat(n).withFailMessage(
"Consumer state remained at " + state + " after 10 seconds")
.isLessThan(100);
}
private boolean requiredState(long state, Map<String, Object> queue) {
Object consumers = queue.get("consumers");
return state == 0
? consumers == null || (Integer) consumers == 0
: consumers != null && (Integer) consumers == state;
}
});
rabbitAdmin.deleteExchange(topic1.getName()); // easier than deleting the binding
rabbitAdmin.declareExchange(topic1);
@@ -186,8 +199,20 @@ public class RabbitBinderCleanerTests {
assertThat(cleanedExchanges).hasSize(6);
}
protected Map<String, Object> getQueue(String string, String queueName) throws URISyntaxException {
URI uri = new URI(RABBITMQ.getHttpUrl())
.resolve("/api/queues/" + UriUtils.encodePathSegment("/", StandardCharsets.UTF_8) + "/" + queueName);
return client.get()
.uri(uri)
.retrieve()
.bodyToMono(new ParameterizedTypeReference<Map<String, Object>>() {
})
.block(Duration.ofSeconds(10));
}
private static Map<String, List<String>> doClean(RabbitBindingCleaner cleaner, String entity, boolean isJob) {
return cleaner.clean(RABBITMQ.getHttpUrl() + "/api", "guest", "guest", "/", BINDER_PREFIX, entity, isJob);
return cleaner.clean(RABBITMQ.getHttpUrl() + "/api", RABBITMQ.getAdminUsername(), RABBITMQ.getAdminPassword(),
"/", BINDER_PREFIX, entity, isJob);
}
}

186
binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java
View File

@@ -19,6 +19,8 @@ package org.springframework.cloud.stream.binder.rabbit;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.lang.reflect.Constructor;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
@@ -39,10 +41,6 @@ import java.util.stream.Collectors;
import java.util.zip.Deflater;
import com.rabbitmq.client.LongString;
import com.rabbitmq.http.client.Client;
import com.rabbitmq.http.client.domain.BindingInfo;
import com.rabbitmq.http.client.domain.ExchangeInfo;
import com.rabbitmq.http.client.domain.QueueInfo;
import org.apache.commons.logging.Log;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
@@ -137,6 +135,8 @@ import org.springframework.messaging.support.GenericMessage;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.util.MimeTypeUtils;
import org.springframework.util.ReflectionUtils;
import org.springframework.web.reactive.function.client.ExchangeFilterFunctions;
import org.springframework.web.reactive.function.client.WebClient;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.fail;
@@ -156,7 +156,7 @@ import static org.mockito.Mockito.when;
public class RabbitBinderTests extends
PartitionCapableBinderTests<RabbitTestBinder, ExtendedConsumerProperties<RabbitConsumerProperties>, ExtendedProducerProperties<RabbitProducerProperties>> {
private static final RabbitMQContainer RABBITMQ = RabbitTestContainer.sharedInstance();
protected static final RabbitMQContainer RABBITMQ = RabbitTestContainer.sharedInstance();
private static final String CLASS_UNDER_TEST_NAME = RabbitMessageChannelBinder.class.getSimpleName();
@@ -164,6 +164,23 @@ public class RabbitBinderTests extends
private static final String BIG_EXCEPTION_MESSAGE = new String(new byte[10_000]).replaceAll("\u0000", "x");
private static final WebClient client;
private static final URI uri;
static {
client = WebClient.builder()
.filter(ExchangeFilterFunctions
.basicAuthentication(RABBITMQ.getAdminUsername(), RABBITMQ.getAdminPassword()))
.build();
try {
uri = new URI(RABBITMQ.getHttpUrl() + "/api/");
}
catch (URISyntaxException ex) {
throw new IllegalStateException(ex);
}
}
@RegisterExtension
private final RabbitTestSupport rabbitTestSupport = new RabbitTestSupport(true, RABBITMQ.getAmqpPort(), RABBITMQ.getHttpPort());
@@ -566,8 +583,7 @@ public class RabbitBinderTests extends
assertThat(container.isRunning()).isTrue();
consumerBinding.unbind();
assertThat(container.isRunning()).isFalse();
Client client = new Client(adminUri());
List<?> bindings = client.getBindingsBySource("/", exchange.getName());
List<?> bindings = getBindingsBySource("/", exchange.getName());
assertThat(bindings.size()).isEqualTo(1);
}
@@ -642,30 +658,29 @@ public class RabbitBinderTests extends
consumerBinding.unbind();
assertThat(container.isRunning()).isFalse();
assertThat(container.getQueueNames()[0]).isEqualTo(group);
Client client = new Client(adminUri());
List<BindingInfo> bindings = client.getBindingsBySource("/", "propsUser2");
List<Map<String, Object>> bindings = getBindingsBySource("/", "propsUser2");
int n = 0;
while (n++ < 100 && bindings == null || bindings.size() < 1) {
while (n++ < 100 && bindings == null || bindings.size() < 2) {
Thread.sleep(100);
bindings = client.getBindingsBySource("/", "propsUser2");
bindings = getBindingsBySource("/", "propsUser2");
}
assertThat(bindings.size()).isEqualTo(2);
assertThat(bindings.get(0).getSource()).isEqualTo("propsUser2");
assertThat(bindings.get(0).getDestination()).isEqualTo(group);
assertThat(bindings.get(0).getRoutingKey()).isIn("foo", "bar");
assertThat(bindings.get(1).getSource()).isEqualTo("propsUser2");
assertThat(bindings.get(1).getDestination()).isEqualTo(group);
assertThat(bindings.get(1).getRoutingKey()).isIn("foo", "bar");
assertThat(bindings.get(1).getRoutingKey()).isNotEqualTo(bindings.get(0).getRoutingKey());
assertThat(bindings.get(0).get("source")).isEqualTo("propsUser2");
assertThat(bindings.get(0).get("destination")).isEqualTo(group);
assertThat(bindings.get(0).get("routing_key")).isIn("foo", "bar");
assertThat(bindings.get(1).get("source")).isEqualTo("propsUser2");
assertThat(bindings.get(1).get("destination")).isEqualTo(group);
assertThat(bindings.get(1).get("routing_key")).isIn("foo", "bar");
assertThat(bindings.get(1).get("routing_key")).isNotEqualTo(bindings.get(0).get("routing_key"));
ExchangeInfo exchange = client.getExchange("/", "propsUser2");
Map<String, Object> exchange = getExchange("/", "propsUser2");
while (n++ < 100 && exchange == null) {
Thread.sleep(100);
exchange = client.getExchange("/", "propsUser2");
exchange = getExchange("/", "propsUser2");
}
assertThat(exchange.getType()).isEqualTo("direct");
assertThat(exchange.isDurable()).isEqualTo(true);
assertThat(exchange.isAutoDelete()).isEqualTo(false);
assertThat(exchange.get("type")).isEqualTo("direct");
assertThat(exchange.get("durable")).isEqualTo(true);
assertThat(exchange.get("auto_delete")).isEqualTo(false);
verifyAutoDeclareContextClear(binder);
}
@@ -716,57 +731,56 @@ public class RabbitBinderTests extends
SimpleMessageListenerContainer container = TestUtils.getPropertyValue(endpoint,
"messageListenerContainer", SimpleMessageListenerContainer.class);
assertThat(container.isRunning()).isTrue();
Client client = new Client(adminUri());
List<BindingInfo> bindings = client.getBindingsBySource("/", "propsUser3");
List<Map<String, Object>> bindings = getBindingsBySource("/", "propsUser3");
int n = 0;
while (n++ < 100 && bindings == null || bindings.size() < 1) {
Thread.sleep(100);
bindings = client.getBindingsBySource("/", "propsUser3");
bindings = getBindingsBySource("/", "propsUser3");
}
assertThat(bindings.size()).isEqualTo(1);
assertThat(bindings.get(0).getSource()).isEqualTo("propsUser3");
assertThat(bindings.get(0).getDestination()).isEqualTo("propsUser3.infra");
assertThat(bindings.get(0).getRoutingKey()).isEqualTo("foo");
assertThat(bindings.get(0).get("source")).isEqualTo("propsUser3");
assertThat(bindings.get(0).get("destination")).isEqualTo("propsUser3.infra");
assertThat(bindings.get(0).get("routing_key")).isEqualTo("foo");
bindings = client.getBindingsBySource("/", "customDLX");
bindings = getBindingsBySource("/", "customDLX");
n = 0;
while (n++ < 100 && bindings == null || bindings.size() < 1) {
Thread.sleep(100);
bindings = client.getBindingsBySource("/", "customDLX");
bindings = getBindingsBySource("/", "customDLX");
}
// assertThat(bindings.size()).isEqualTo(1);
assertThat(bindings.get(0).getSource()).isEqualTo("customDLX");
assertThat(bindings.get(0).getDestination()).isEqualTo("customDLQ");
assertThat(bindings.get(0).getRoutingKey()).isEqualTo("customDLRK");
assertThat(bindings.get(0).get("source")).isEqualTo("customDLX");
assertThat(bindings.get(0).get("destination")).isEqualTo("customDLQ");
assertThat(bindings.get(0).get("routing_key")).isEqualTo("customDLRK");
ExchangeInfo exchange = client.getExchange("/", "propsUser3");
Map<String, Object> exchange = getExchange("/", "propsUser3");
n = 0;
while (n++ < 100 && exchange == null) {
Thread.sleep(100);
exchange = client.getExchange("/", "propsUser3");
exchange = getExchange("/", "propsUser3");
}
assertThat(exchange.getType()).isEqualTo("direct");
assertThat(exchange.isDurable()).isEqualTo(false);
assertThat(exchange.isAutoDelete()).isEqualTo(true);
assertThat(exchange.get("type")).isEqualTo("direct");
assertThat(exchange.get("durable")).isEqualTo(false);
assertThat(exchange.get("auto_delete")).isEqualTo(true);
exchange = client.getExchange("/", "customDLX");
exchange = getExchange("/", "customDLX");
n = 0;
while (n++ < 100 && exchange == null) {
Thread.sleep(100);
exchange = client.getExchange("/", "customDLX");
exchange = getExchange("/", "customDLX");
}
assertThat(exchange.getType()).isEqualTo("topic");
assertThat(exchange.isDurable()).isEqualTo(true);
assertThat(exchange.isAutoDelete()).isEqualTo(false);
assertThat(exchange.get("type")).isEqualTo("topic");
assertThat(exchange.get("durable")).isEqualTo(true);
assertThat(exchange.get("auto_delete")).isEqualTo(false);
QueueInfo queue = client.getQueue("/", "propsUser3.infra");
Map<String, Object> queue = getQueue("/", "propsUser3.infra");
n = 0;
while (n++ < 100 && queue == null || queue.getConsumerCount() == 0) {
while (n++ < 100 && queue == null || !requiredConsumerState(1, queue)) {
Thread.sleep(100);
queue = client.getQueue("/", "propsUser3.infra");
queue = getQueue("/", "propsUser3.infra");
}
assertThat(queue).isNotNull();
Map<String, Object> args = queue.getArguments();
Map<String, Object> args = (Map<String, Object>) queue.get("arguments");
assertThat(args.get("x-expires")).isEqualTo(30_000);
assertThat(args.get("x-max-length")).isEqualTo(10_000);
assertThat(args.get("x-max-length-bytes")).isEqualTo(100_000);
@@ -776,17 +790,17 @@ public class RabbitBinderTests extends
assertThat(args.get("x-dead-letter-exchange")).isEqualTo("customDLX");
assertThat(args.get("x-dead-letter-routing-key")).isEqualTo("customDLRK");
assertThat(args.get("x-queue-mode")).isEqualTo("lazy");
assertThat(queue.getExclusiveConsumerTag()).isEqualTo("testConsumerTag#0");
assertThat(queue.get("exclusive_consumer_tag")).isEqualTo("testConsumerTag#0");
queue = client.getQueue("/", "customDLQ");
queue = getQueue("/", "customDLQ");
n = 0;
while (n++ < 100 && queue == null) {
Thread.sleep(100);
queue = client.getQueue("/", "customDLQ");
queue = getQueue("/", "customDLQ");
}
assertThat(queue).isNotNull();
args = queue.getArguments();
args = (Map<String, Object>) queue.get("arguments");
assertThat(args.get("x-expires")).isEqualTo(60_000);
assertThat(args.get("x-max-length")).isEqualTo(20_000);
assertThat(args.get("x-max-length-bytes")).isEqualTo(40_000);
@@ -804,6 +818,7 @@ public class RabbitBinderTests extends
}
@SuppressWarnings("unchecked")
@Test
public void testConsumerPropertiesWithHeaderExchanges() throws Exception {
RabbitTestBinder binder = getBinder();
@@ -828,30 +843,31 @@ public class RabbitBinderTests extends
consumerBinding.unbind();
assertThat(container.isRunning()).isFalse();
assertThat(container.getQueueNames()[0]).isEqualTo("propsHeader." + group);
Client client = new Client(adminUri());
List<BindingInfo> bindings = client.getBindingsBySource("/", "propsHeader");
List<Map<String, Object>> bindings = getBindingsBySource("/", "propsHeader");
int n = 0;
while (n++ < 100 && bindings == null || bindings.size() < 1) {
Thread.sleep(100);
bindings = client.getBindingsBySource("/", "propsHeader");
bindings = getBindingsBySource("/", "propsHeader");
}
assertThat(bindings.size()).isEqualTo(1);
assertThat(bindings.get(0).getSource()).isEqualTo("propsHeader");
assertThat(bindings.get(0).getDestination()).isEqualTo("propsHeader." + group);
assertThat(bindings.get(0).getArguments()).hasEntrySatisfying("x-match", v -> assertThat(v).isEqualTo("any"));
assertThat(bindings.get(0).getArguments()).hasEntrySatisfying("foo", v -> assertThat(v).isEqualTo("bar"));
assertThat(bindings.get(0).get("source")).isEqualTo("propsHeader");
assertThat(bindings.get(0).get("destination")).isEqualTo("propsHeader." + group);
Map<String, Object> args = (Map<String, Object>) bindings.get(0).get("arguments");
assertThat(args).hasEntrySatisfying("x-match", v -> assertThat(v).isEqualTo("any"));
assertThat(args).hasEntrySatisfying("foo", v -> assertThat(v).isEqualTo("bar"));
bindings = client.getBindingsBySource("/", "propsHeader.dlx");
bindings = getBindingsBySource("/", "propsHeader.dlx");
n = 0;
while (n++ < 100 && bindings == null || bindings.size() < 1) {
Thread.sleep(100);
bindings = client.getBindingsBySource("/", "propsHeader.dlx");
bindings = getBindingsBySource("/", "propsHeader.dlx");
}
assertThat(bindings.size()).isEqualTo(1);
assertThat(bindings.get(0).getSource()).isEqualTo("propsHeader.dlx");
assertThat(bindings.get(0).getDestination()).isEqualTo("propsHeader." + group + ".dlq");
assertThat(bindings.get(0).getArguments()).hasEntrySatisfying("x-match", v -> assertThat(v).isEqualTo("any"));
assertThat(bindings.get(0).getArguments()).hasEntrySatisfying("foo", v -> assertThat(v).isEqualTo("bar"));
assertThat(bindings.get(0).get("source")).isEqualTo("propsHeader.dlx");
assertThat(bindings.get(0).get("destination")).isEqualTo("propsHeader." + group + ".dlq");
args = (Map<String, Object>) bindings.get(0).get("arguments");
assertThat(args).hasEntrySatisfying("x-match", v -> assertThat(v).isEqualTo("any"));
assertThat(args).hasEntrySatisfying("foo", v -> assertThat(v).isEqualTo("bar"));
verifyAutoDeclareContextClear(binder);
}
@@ -1107,22 +1123,22 @@ public class RabbitBinderTests extends
DirectChannel moduleInputChannel = createBindableChannel("input",
bindingProperties);
moduleInputChannel.setBeanName("dlqTestManual");
Client client = new Client(adminUri());
moduleInputChannel.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
// Wait until the unacked state is reflected in the admin
QueueInfo info = client.getQueue("/", TEST_PREFIX + "dlqTestManual.default");
Map<String, Object> info = getQueue("/", TEST_PREFIX + "dlqTestManual.default");
int n = 0;
while (n++ < 100 && info.getMessagesUnacknowledged() < 1L) {
while (n++ < 100 && ( info.get("messages_unacknowledged") == null
|| ((Long) info.get("messages_unacknowledged")) < 1L)) {
try {
Thread.sleep(100);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
info = client.getQueue("/", TEST_PREFIX + "dlqTestManual.default");
info = getQueue("/", TEST_PREFIX + "dlqTestManual.default");
}
throw new RuntimeException("foo");
}
@@ -1148,12 +1164,12 @@ public class RabbitBinderTests extends
assertThat(n).isLessThan(100);
n = 0;
QueueInfo info = client.getQueue("/", TEST_PREFIX + "dlqTestManual.default");
while (n++ < 100 && info.getMessagesUnacknowledged() > 0L) {
Map<String, Object> info = getQueue("/", TEST_PREFIX + "dlqTestManual.default");
while (n++ < 100 && unackedMessages(info)) {
Thread.sleep(100);
info = client.getQueue("/", TEST_PREFIX + "dlqTestManual.default");
info = getQueue("/", TEST_PREFIX + "dlqTestManual.default");
}
assertThat(info.getMessagesUnacknowledged()).isEqualTo(0L);
assertThat(unackedMessages(info)).isFalse();
consumerBinding.unbind();
@@ -1170,6 +1186,11 @@ public class RabbitBinderTests extends
verifyAutoDeclareContextClear(binder);
}
protected boolean unackedMessages(Map<String, Object> info) {
Object unack = info.get("messages_unacknowledged");
return unack != null && ((Integer) unack) > 0L;
}
@Test
public void testAutoBindDLQPartionedConsumerFirst(TestInfo testInfo) throws Exception {
@@ -2600,6 +2621,25 @@ public class RabbitBinderTests extends
assertThat(ctx.getBeansOfType(Declarable.class)).isEmpty();
}
private List<Map<String, Object>> getBindingsBySource(String vhost, String name) {
return RestUtils.getBindingsBySource(client, uri, vhost, name);
}
private Map<String, Object> getExchange(String vhost, String name) {
return RestUtils.getExchange(client, uri, vhost, name);
}
private Map<String, Object> getQueue(String vhost, String name) {
return RestUtils.getQueue(client, uri, vhost, name);
}
private boolean requiredConsumerState(long state, Map<String, Object> queue) {
Object consumers = queue.get("consumers");
return state == 0
? consumers == null || (Integer) consumers == 0
: consumers != null && (Integer) consumers == state;
}
public static class TestPartitionKeyExtractorClass
implements PartitionKeyExtractorStrategy {

77
binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RestUtils.java Normal file
View File

@@ -0,0 +1,77 @@
/*
* 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.cloud.stream.binder.rabbit;
import java.net.URI;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.List;
import java.util.Map;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.web.util.UriUtils;
/**
* @author Gary Russell
* @since 4.0
*
*/
public final class RestUtils {
private RestUtils() {
}
public static List<Map<String, Object>> getBindingsBySource(WebClient client, URI uri, String vhost, String name) {
String exchange = "".equals(name) ? "amq.default" : name;
URI getUri = uri
.resolve("/api/exchanges/" + UriUtils.encodePathSegment(vhost, StandardCharsets.UTF_8) + "/"
+ UriUtils.encodePathSegment(exchange, StandardCharsets.UTF_8) + "/bindings/source");
return client.get()
.uri(getUri)
.retrieve()
.bodyToMono(new ParameterizedTypeReference<List<Map<String, Object>>>() {
})
.block(Duration.ofSeconds(10));
}
public static Map<String, Object> getExchange(WebClient client, URI uri, String vhost, String name) {
String exchange = "".equals(name) ? "amq.default" : name;
URI getUri = uri
.resolve("/api/exchanges/" + UriUtils.encodePathSegment(vhost, StandardCharsets.UTF_8) + "/"
+ UriUtils.encodePathSegment(exchange, StandardCharsets.UTF_8));
return client.get()
.uri(getUri)
.retrieve()
.bodyToMono(new ParameterizedTypeReference<Map<String, Object>>() {
})
.block(Duration.ofSeconds(10));
}
public static Map<String, Object> getQueue(WebClient client, URI uri, String vhost, String name) {
URI getUri = uri
.resolve("/api/queues/" + UriUtils.encodePathSegment(vhost, StandardCharsets.UTF_8) + "/"
+ UriUtils.encodePathSegment(name, StandardCharsets.UTF_8));
return client.get()
.uri(getUri)
.retrieve()
.bodyToMono(new ParameterizedTypeReference<Map<String, Object>>() {
})
.block(Duration.ofSeconds(10));
}
}

44
binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java
View File

@@ -17,6 +17,7 @@
package org.springframework.cloud.stream.binder.rabbit.integration;
import java.net.MalformedURLException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.List;
@@ -24,10 +25,6 @@ import java.util.Map;
import java.util.UUID;
import java.util.function.Function;
import com.rabbitmq.http.client.Client;
import com.rabbitmq.http.client.domain.BindingInfo;
import com.rabbitmq.http.client.domain.ExchangeInfo;
import com.rabbitmq.http.client.domain.QueueInfo;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
@@ -60,6 +57,7 @@ import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder;
import org.springframework.cloud.stream.binder.rabbit.RabbitMessageChannelBinder;
import org.springframework.cloud.stream.binder.rabbit.RabbitTestContainer;
import org.springframework.cloud.stream.binder.rabbit.RestUtils;
import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties;
import org.springframework.cloud.stream.binder.rabbit.properties.RabbitProducerProperties;
import org.springframework.cloud.stream.binder.rabbit.properties.RabbitProducerProperties.AlternateExchange;
@@ -80,6 +78,8 @@ import org.springframework.messaging.support.GenericMessage;
import org.springframework.retry.backoff.ExponentialBackOffPolicy;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.web.reactive.function.client.ExchangeFilterFunctions;
import org.springframework.web.reactive.function.client.WebClient;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.entry;
@@ -100,6 +100,23 @@ public class RabbitBinderModuleTests {
private static final ConnectionFactory MOCK_CONNECTION_FACTORY = mock(ConnectionFactory.class, Mockito.RETURNS_MOCKS);
private static final WebClient client;
private static final URI uri;
static {
client = WebClient.builder()
.filter(ExchangeFilterFunctions
.basicAuthentication(RABBITMQ.getAdminUsername(), RABBITMQ.getAdminPassword()))
.build();
try {
uri = new URI(RABBITMQ.getHttpUrl() + "/api/");
}
catch (URISyntaxException ex) {
throw new IllegalStateException(ex);
}
}
@RegisterExtension
private final RabbitTestSupport rabbitTestSupport = new RabbitTestSupport(true, RABBITMQ.getAmqpPort(), RABBITMQ.getHttpPort());
@@ -169,21 +186,22 @@ public class RabbitBinderModuleTests {
checkCustomizedArgs();
}
@SuppressWarnings("unchecked")
private void checkCustomizedArgs() throws MalformedURLException, URISyntaxException, InterruptedException {
Client client = new Client(String.format("http://guest:guest@localhost:%d/api", RABBITMQ.getHttpPort()));
List<BindingInfo> bindings = client.getBindingsBySource("/", "process-in-0");
List<Map<String, Object>> bindings = RestUtils.getBindingsBySource(client, uri, "/", "process-in-0");
int n = 0;
while (n++ < 100 && bindings == null || bindings.size() < 1) {
Thread.sleep(100);
bindings = client.getBindingsBySource("/", "process-in-0");
bindings = RestUtils.getBindingsBySource(client, uri, "/", "process-in-0");
}
assertThat(bindings).isNotNull();
assertThat(bindings.get(0).getArguments()).contains(entry("added.by", "customizer"));
ExchangeInfo exchange = client.getExchange("/", "process-in-0");
assertThat(exchange.getArguments()).contains(entry("added.by", "customizer"));
QueueInfo queue = client.getQueue("/", bindings.get(0).getDestination());
assertThat(queue.getArguments()).contains(entry("added.by", "customizer"));
assertThat(queue.getArguments()).contains(entry("x-single-active-consumer", Boolean.TRUE));
assertThat((Map<String, Object>) bindings.get(0).get("arguments")).contains(entry("added.by", "customizer"));
Map<String, Object> exchange = RestUtils.getExchange(client, uri, "/", "process-in-0");
assertThat((Map<String, Object>) exchange.get("arguments")).contains(entry("added.by", "customizer"));
Map<String, Object> queue = RestUtils.getQueue(client, uri, "/", (String) bindings.get(0).get("destination"));
Map<String, Object> args = (Map<String, Object>) queue.get("arguments");
assertThat(args).contains(entry("added.by", "customizer"));
assertThat(args).contains(entry("x-single-active-consumer", Boolean.TRUE));
}
@Test