diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/pom.xml b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/pom.xml index b3d24ec9d..8203d1462 100644 --- a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/pom.xml +++ b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/pom.xml @@ -47,13 +47,22 @@ test - org.apache.httpcomponents - httpclient + org.springframework.boot + spring-boot-starter-webflux - com.rabbitmq - http-client - 2.1.0.RELEASE + io.projectreactor.netty + reactor-netty + + + org.springframework.boot + spring-boot-starter-test + test + + + io.projectreactor + reactor-test + test diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/admin/RabbitBindingCleaner.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/admin/RabbitBindingCleaner.java index 13c239a0b..d1e0dcf78 100644 --- a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/admin/RabbitBindingCleaner.java +++ b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/admin/RabbitBindingCleaner.java @@ -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> doClean(Client client, - String vhost, String binderPrefix, String entity, boolean isJob) { + private Map> doClean(WebClient client, + URI uri, String vhost, String binderPrefix, String entity, boolean isJob) { LinkedList removedQueues = isJob ? null - : findStreamQueues(client, vhost, binderPrefix, entity); - List removedExchanges = findExchanges(client, vhost, binderPrefix, entity); + : findStreamQueues(client, uri, vhost, binderPrefix, entity); + List 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 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 findStreamQueues(WebClient client, URI uri, String vhost, String binderPrefix, + String stream) { + String queueNamePrefix = adjustPrefix(AbstractBinder.applyPrefix(binderPrefix, stream)); - List queues = client.getQueues(vhost); + List> 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> 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>>() { + }) + .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 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 findExchanges(Client client, String vhost, String binderPrefix, String entity) { - List exchanges = client.getExchanges(vhost); + private List findExchanges(WebClient client, URI uri, String vhost, String binderPrefix, String entity) { + List> exchanges = getExchanges(client, uri, vhost); String exchangeNamePrefix = adjustPrefix(AbstractBinder.applyPrefix(binderPrefix, entity)); List 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 bindingsBySource = client.getBindingsBySource(vhost, e.getName()); - return Collections.singletonMap(e.getName(), bindingsBySource); + List> 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> bindings, String exchangeNamePrefix) { - Entry> next = bindings.entrySet().iterator().next(); - for (BindingInfo binding : next.getValue()) { - if (!"queue".equals(binding.getDestinationType()) - || !binding.getDestination().startsWith(exchangeNamePrefix)) { + private List> 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>>() { + }) + .block(Duration.ofSeconds(10)); + } + + private List> 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>>() { + }) + .block(Duration.ofSeconds(10)); + } + + private List> 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>>() { + }) + .block(Duration.ofSeconds(10)); + } + + private String hasNoForeignBindings(Map>> bindings, String exchangeNamePrefix) { + Entry>> next = bindings.entrySet().iterator().next(); + for (Map 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); } diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderCleanerTests.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderCleanerTests.java index f7472b47c..4cb59bcc3 100644 --- a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderCleanerTests.java +++ b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderCleanerTests.java @@ -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 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 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 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>() { + }) + .block(Duration.ofSeconds(10)); + } + private static Map> 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); } } diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java index 2cffc1cd9..6840d3c92 100644 --- a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java +++ b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java @@ -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, ExtendedProducerProperties> { - 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 bindings = client.getBindingsBySource("/", "propsUser2"); + List> 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 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 bindings = client.getBindingsBySource("/", "propsUser3"); + List> 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 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 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 args = queue.getArguments(); + Map args = (Map) 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) 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 bindings = client.getBindingsBySource("/", "propsHeader"); + List> 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 args = (Map) 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) 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 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 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 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> getBindingsBySource(String vhost, String name) { + return RestUtils.getBindingsBySource(client, uri, vhost, name); + } + + private Map getExchange(String vhost, String name) { + return RestUtils.getExchange(client, uri, vhost, name); + } + + private Map getQueue(String vhost, String name) { + return RestUtils.getQueue(client, uri, vhost, name); + } + + private boolean requiredConsumerState(long state, Map 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 { diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RestUtils.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RestUtils.java new file mode 100644 index 000000000..41f2c53c4 --- /dev/null +++ b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RestUtils.java @@ -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> 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>>() { + }) + .block(Duration.ofSeconds(10)); + } + + public static Map 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>() { + }) + .block(Duration.ofSeconds(10)); + } + + public static Map 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>() { + }) + .block(Duration.ofSeconds(10)); + } + +} diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java index e6370f9c8..52a91fe40 100644 --- a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java +++ b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java @@ -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 bindings = client.getBindingsBySource("/", "process-in-0"); + List> 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) bindings.get(0).get("arguments")).contains(entry("added.by", "customizer")); + Map exchange = RestUtils.getExchange(client, uri, "/", "process-in-0"); + assertThat((Map) exchange.get("arguments")).contains(entry("added.by", "customizer")); + Map queue = RestUtils.getQueue(client, uri, "/", (String) bindings.get(0).get("destination")); + Map args = (Map) queue.get("arguments"); + assertThat(args).contains(entry("added.by", "customizer")); + assertThat(args).contains(entry("x-single-active-consumer", Boolean.TRUE)); } @Test