From 18a6663bb0dced3d9bc7f5932182077828f0b4ef Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 26 Feb 2019 15:00:13 -0500 Subject: [PATCH] GH-195: Remove usage of RabbitManagementTemplate Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-rabbit/issues/195 Also convert the binder cleaner to use Hop instead of directly using a `RestTemplate`. Also, temporarily drop back to the 4.5.6 Http client (see https://github.com/spring-projects/spring-boot/issues/16043) because 4.5.7 broke RabbitMQ REST calls for the `/` virtual host. Resolves #196 --- .../rabbit/admin/RabbitBindingCleaner.java | 182 ++++++++---------- .../rabbit/admin/RabbitManagementUtils.java | 90 --------- spring-cloud-stream-binder-rabbit/pom.xml | 6 + .../rabbit/RabbitBinderCleanerTests.java | 66 +++---- .../binder/rabbit/RabbitBinderTests.java | 60 +++--- 5 files changed, 134 insertions(+), 270 deletions(-) delete mode 100644 spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/admin/RabbitManagementUtils.java diff --git a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/admin/RabbitBindingCleaner.java b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/admin/RabbitBindingCleaner.java index 9a34ab04a..c02738ae5 100644 --- a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/admin/RabbitBindingCleaner.java +++ b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/admin/RabbitBindingCleaner.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2016 the original author or authors. + * Copyright 2015-2019 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. @@ -16,19 +16,25 @@ package org.springframework.cloud.stream.binder.rabbit.admin; -import java.net.URI; -import java.util.ArrayList; +import java.net.MalformedURLException; +import java.net.URISyntaxException; +import java.util.Collections; import java.util.HashMap; +import java.util.LinkedList; import java.util.List; 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.web.client.RestTemplate; -import org.springframework.web.util.UriComponentsBuilder; /** * Implementation of {@link org.springframework.cloud.stream.binder.BindingCleaner} for @@ -51,81 +57,65 @@ public class RabbitBindingCleaner implements BindingCleaner { @Override public Map> clean(String entity, boolean isJob) { - return clean("http://localhost:15672", "guest", "guest", "/", BINDER_PREFIX, + return clean("http://localhost:15672/api", "guest", "guest", "/", BINDER_PREFIX, entity, isJob); } public Map> clean(String adminUri, String user, String pw, String vhost, String binderPrefix, String entity, boolean isJob) { - return doClean(adminUri == null ? "http://localhost:15672" : adminUri, - user == null ? "guest" : user, pw == null ? "guest" : pw, - vhost == null ? "/" : vhost, - binderPrefix == null ? BINDER_PREFIX : binderPrefix, entity, isJob); + + try { + Client client = new Client(adminUri, user, pw); + return doClean(client, + vhost == null ? "/" : vhost, + binderPrefix == null ? BINDER_PREFIX : binderPrefix, entity, isJob); + } + catch (MalformedURLException | URISyntaxException e) { + throw new RabbitAdminException("Couldn't create a Client", e); + } } - private Map> doClean(String adminUri, String user, String pw, + private Map> doClean(Client client, String vhost, String binderPrefix, String entity, boolean isJob) { - RestTemplate restTemplate = RabbitManagementUtils.buildRestTemplate(adminUri, - user, pw); - List removedQueues = isJob ? null - : findStreamQueues(adminUri, vhost, binderPrefix, entity, restTemplate); - List removedExchanges = findExchanges(adminUri, vhost, binderPrefix, - entity, restTemplate); + + LinkedList removedQueues = isJob ? null + : findStreamQueues(client, vhost, binderPrefix, entity); + List removedExchanges = findExchanges(client, 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. - for (int i = removedQueues.size() - 1; i >= 0; i--) { - String queueName = removedQueues.get(i); - URI uri = UriComponentsBuilder.fromUriString(adminUri + "/api") - .pathSegment("queues", "{vhost}", "{stream}") - .buildAndExpand(vhost, queueName).encode().toUri(); - restTemplate.delete(uri); - if (logger.isDebugEnabled()) { - logger.debug("deleted queue: " + queueName); - } + if (removedQueues != null) { + removedQueues.descendingIterator().forEachRemaining(q -> { + client.deleteQueue(vhost, q); + if (logger.isDebugEnabled()) { + logger.debug("deleted queue: " + q); + } + }); } Map> results = new HashMap<>(); if (removedQueues.size() > 0) { results.put("queues", removedQueues); } // Fanout exchanges for taps - for (String exchange : removedExchanges) { - URI uri = UriComponentsBuilder.fromUriString(adminUri + "/api") - .pathSegment("exchanges", "{vhost}", "{name}") - .buildAndExpand(vhost, exchange).encode().toUri(); - restTemplate.delete(uri); + removedExchanges.forEach(exchange -> { + client.deleteExchange(vhost, exchange); if (logger.isDebugEnabled()) { logger.debug("deleted exchange: " + exchange); } - } + }); if (removedExchanges.size() > 0) { results.put("exchanges", removedExchanges); } return results; } - private List findStreamQueues(String adminUri, String vhost, - String binderPrefix, String stream, RestTemplate restTemplate) { - String queueNamePrefix = adjustPrefix( - AbstractBinder.applyPrefix(binderPrefix, stream)); - List> queues = listAllQueues(adminUri, vhost, restTemplate); - List removedQueues = new ArrayList<>(); - for (Map queue : queues) { - String queueName = (String) queue.get("name"); - if (queueName.startsWith(queueNamePrefix)) { - checkNoConsumers(queueName, queue); - removedQueues.add(queueName); - } - } - return removedQueues; - } - - private List> listAllQueues(String adminUri, String vhost, - RestTemplate restTemplate) { - URI uri = UriComponentsBuilder.fromUriString(adminUri + "/api") - .pathSegment("queues", "{vhost}").buildAndExpand(vhost).encode().toUri(); - List> queues = restTemplate.getForObject(uri, List.class); - return queues; + private LinkedList findStreamQueues(Client client, String vhost, String binderPrefix, String stream) { + String queueNamePrefix = adjustPrefix(AbstractBinder.applyPrefix(binderPrefix, stream)); + List queues = client.getQueues(vhost); + return queues.stream() + .filter(q -> q.getName().startsWith(queueNamePrefix)) + .map(q -> checkNoConsumers(q)) + .collect(Collectors.toCollection(LinkedList::new)); } private String adjustPrefix(String prefix) { @@ -137,68 +127,46 @@ public class RabbitBindingCleaner implements BindingCleaner { } } - private void checkNoConsumers(String queueName, Map queue) { - if (!queue.get("consumers").equals(Integer.valueOf(0))) { - throw new RabbitAdminException("Queue " + queueName + " is in use"); + private String checkNoConsumers(QueueInfo queue) { + if (queue.getConsumerCount() != 0) { + throw new RabbitAdminException("Queue " + queue.getName() + " is in use"); } + return queue.getName(); } - private List findExchanges(String adminUri, String vhost, String binderPrefix, - String entity, RestTemplate restTemplate) { - List removedExchanges = new ArrayList<>(); - URI uri = UriComponentsBuilder.fromUriString(adminUri + "/api") - .pathSegment("exchanges", "{vhost}").buildAndExpand(vhost).encode() - .toUri(); - List> exchanges = restTemplate.getForObject(uri, List.class); - String exchangeNamePrefix = adjustPrefix( - AbstractBinder.applyPrefix(binderPrefix, entity)); - for (Map exchange : exchanges) { - String exchangeName = (String) exchange.get("name"); - if (exchangeName.startsWith(exchangeNamePrefix)) { - uri = UriComponentsBuilder.fromUriString(adminUri + "/api") - .pathSegment("exchanges", "{vhost}", "{name}", "bindings", - "source") - .buildAndExpand(vhost, exchangeName).encode().toUri(); - List> bindings = restTemplate.getForObject(uri, - List.class); - if (hasNoForeignBindings(bindings, exchangeNamePrefix)) { - uri = UriComponentsBuilder.fromUriString(adminUri + "/api") - .pathSegment("exchanges", "{vhost}", "{name}", "bindings", - "destination") - .buildAndExpand(vhost, exchangeName).encode().toUri(); - bindings = restTemplate.getForObject(uri, List.class); - if (bindings.size() == 0) { - removedExchanges.add((String) exchange.get("name")); - } - else { + private List findExchanges(Client client, String vhost, String binderPrefix, String entity) { + List exchanges = client.getExchanges(vhost); + String exchangeNamePrefix = adjustPrefix(AbstractBinder.applyPrefix(binderPrefix, entity)); + List exchangesToRemove = exchanges.stream() + .filter(e -> e.getName().startsWith(exchangeNamePrefix)) + .map(e -> { + System.out.println(e.getName()); + List bindingsBySource = client.getBindingsBySource(vhost, e.getName()); + return Collections.singletonMap(e.getName(), bindingsBySource); + }) + .map(bindingsMap -> hasNoForeignBindings(bindingsMap, exchangeNamePrefix)) + .collect(Collectors.toList()); + exchangesToRemove.stream() + .map(exchange -> client.getExchangeBindingsByDestination(vhost, exchange)) + .forEach(bindings -> { + if (bindings.size() > 0) { throw new RabbitAdminException("Cannot delete exchange " - + exchangeName + "; it is a destination: " + bindings); + + bindings.get(0).getDestination() + "; it is a destination: " + bindings); } - } - else { - throw new RabbitAdminException("Cannot delete exchange " - + exchangeName + "; it has bindings: " + bindings); - } - } - } - return removedExchanges; + }); + return exchangesToRemove; } - private boolean hasNoForeignBindings(List> bindings, - String exchangeNamePrefix) { - if (bindings.size() == 0) { - return true; - } - boolean noForeign = true; - for (Map binding : bindings) { - if (!("queue".equals(binding.get("destination_type"))) - || !((String) binding.get("destination")) - .startsWith(exchangeNamePrefix)) { - noForeign = false; - break; + 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)) { + throw new RabbitAdminException("Cannot delete exchange " + + next.getKey() + "; it has bindings: " + bindings); } } - return noForeign; + return next.getKey(); } } diff --git a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/admin/RabbitManagementUtils.java b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/admin/RabbitManagementUtils.java deleted file mode 100644 index 09c139479..000000000 --- a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/admin/RabbitManagementUtils.java +++ /dev/null @@ -1,90 +0,0 @@ -/* - * Copyright 2015-2019 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 - * - * http://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.admin; - -import java.net.URI; -import java.net.URISyntaxException; -import java.util.Collections; - -import org.apache.http.HttpHost; -import org.apache.http.auth.AuthScope; -import org.apache.http.auth.UsernamePasswordCredentials; -import org.apache.http.client.AuthCache; -import org.apache.http.client.HttpClient; -import org.apache.http.client.protocol.HttpClientContext; -import org.apache.http.impl.auth.BasicScheme; -import org.apache.http.impl.client.BasicAuthCache; -import org.apache.http.impl.client.BasicCredentialsProvider; -import org.apache.http.impl.client.HttpClients; -import org.apache.http.protocol.HttpContext; - -import org.springframework.http.HttpMethod; -import org.springframework.http.client.HttpComponentsClientHttpRequestFactory; -import org.springframework.http.converter.HttpMessageConverter; -import org.springframework.http.converter.json.MappingJackson2HttpMessageConverter; -import org.springframework.web.client.RestTemplate; - -/** - * @author Gary Russell - * @since 1.2 - */ -public abstract class RabbitManagementUtils { - - public static RestTemplate buildRestTemplate(String adminUri, String user, - String password) { - BasicCredentialsProvider credsProvider = new BasicCredentialsProvider(); - credsProvider.setCredentials( - new AuthScope(AuthScope.ANY_HOST, AuthScope.ANY_PORT), - new UsernamePasswordCredentials(user, password)); - HttpClient httpClient = HttpClients.custom() - .setDefaultCredentialsProvider(credsProvider).build(); - // Set up pre-emptive basic Auth because the rabbit plugin doesn't currently - // support challenge/response for PUT - // Create AuthCache instance - AuthCache authCache = new BasicAuthCache(); - // Generate BASIC scheme object and add it to the local; from the apache docs... - // auth cache - BasicScheme basicAuth = new BasicScheme(); - URI uri; - try { - uri = new URI(adminUri); - } - catch (URISyntaxException e) { - throw new RabbitAdminException("Invalid URI", e); - } - authCache.put(new HttpHost(uri.getHost(), uri.getPort(), uri.getScheme()), - basicAuth); - // Add AuthCache to the execution context - final HttpClientContext localContext = HttpClientContext.create(); - localContext.setAuthCache(authCache); - RestTemplate restTemplate = new RestTemplate( - new HttpComponentsClientHttpRequestFactory(httpClient) { - - @Override - protected HttpContext createHttpContext(HttpMethod httpMethod, - URI uri) { - return localContext; - } - - }); - restTemplate - .setMessageConverters(Collections.>singletonList( - new MappingJackson2HttpMessageConverter())); - return restTemplate; - } - -} diff --git a/spring-cloud-stream-binder-rabbit/pom.xml b/spring-cloud-stream-binder-rabbit/pom.xml index a0bab2e4f..8ce00516d 100644 --- a/spring-cloud-stream-binder-rabbit/pom.xml +++ b/spring-cloud-stream-binder-rabbit/pom.xml @@ -85,5 +85,11 @@ spring-cloud-stream-binder-rabbit-test-support test + + + org.apache.httpcomponents + httpclient + 4.5.6 + diff --git a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderCleanerTests.java b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderCleanerTests.java index f1000d621..3acefafb3 100644 --- a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderCleanerTests.java +++ b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderCleanerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2016 the original author or authors. + * Copyright 2015-2019 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. @@ -16,7 +16,8 @@ package org.springframework.cloud.stream.binder.rabbit; -import java.net.URI; +import java.net.MalformedURLException; +import java.net.URISyntaxException; import java.util.List; import java.util.Map; import java.util.UUID; @@ -24,9 +25,12 @@ import java.util.UUID; import com.fasterxml.jackson.annotation.JsonProperty; 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.Rule; import org.junit.Test; +import org.springframework.amqp.core.Base64UrlNamingStrategy; import org.springframework.amqp.core.BindingBuilder; import org.springframework.amqp.core.Queue; import org.springframework.amqp.core.TopicExchange; @@ -37,10 +41,7 @@ import org.springframework.amqp.rabbit.core.RabbitTemplate; 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.rabbit.admin.RabbitManagementUtils; import org.springframework.cloud.stream.binder.test.junit.rabbit.RabbitTestSupport; -import org.springframework.web.client.RestTemplate; -import org.springframework.web.util.UriComponentsBuilder; import static org.assertj.core.api.Assertions.assertThat; import static org.junit.Assert.fail; @@ -49,19 +50,29 @@ import static org.junit.Assert.fail; * @author Gary Russell * @since 1.2 */ +//@Ignore public class RabbitBinderCleanerTests { private static final String BINDER_PREFIX = "binder."; + private static final Client client; + + static { + try { + client = new Client("http://localhost:15672/api", "guest", "guest"); + } + catch (MalformedURLException | URISyntaxException e) { + throw new RabbitAdminException("Couldn't create a Client", e); + } + } + @Rule public RabbitTestSupport rabbitWithMgmtEnabled = new RabbitTestSupport(true); @Test public void testCleanStream() { final RabbitBindingCleaner cleaner = new RabbitBindingCleaner(); - final RestTemplate template = RabbitManagementUtils - .buildRestTemplate("http://localhost:15672", "guest", "guest"); - final String stream1 = UUID.randomUUID().toString(); + final String stream1 = new Base64UrlNamingStrategy("foo").generateName(); String stream2 = stream1 + "-1"; String firstQueue = null; CachingConnectionFactory connectionFactory = rabbitWithMgmtEnabled.getResource(); @@ -74,20 +85,9 @@ public class RabbitBinderCleanerTests { if (firstQueue == null) { firstQueue = queue1Name; } - URI uri = UriComponentsBuilder - .fromUriString("http://localhost:15672/api/queues") - .pathSegment("{vhost}", "{queue}").buildAndExpand("/", queue1Name) - .encode().toUri(); - template.put(uri, new AmqpQueue(false, true)); - uri = UriComponentsBuilder.fromUriString("http://localhost:15672/api/queues") - .pathSegment("{vhost}", "{queue}").buildAndExpand("/", queue2Name) - .encode().toUri(); - template.put(uri, new AmqpQueue(false, true)); - uri = UriComponentsBuilder.fromUriString("http://localhost:15672/api/queues") - .pathSegment("{vhost}", "{queue}") - .buildAndExpand("/", AbstractBinder.constructDLQName(queue1Name)) - .encode().toUri(); - template.put(uri, new AmqpQueue(false, true)); + rabbitAdmin.declareQueue(new Queue(queue1Name, true, false, false)); + rabbitAdmin.declareQueue(new Queue(queue2Name, true, false, false)); + rabbitAdmin.declareQueue(new Queue(AbstractBinder.constructDLQName(queue1Name), true, false, false)); TopicExchange exchange = new TopicExchange(queue1Name); rabbitAdmin.declareExchange(exchange); rabbitAdmin.declareBinding(BindingBuilder.bind(new Queue(queue1Name)) @@ -141,25 +141,13 @@ public class RabbitBinderCleanerTests { return null; } - private void waitForConsumerStateNot(String queueName, int state) - throws InterruptedException { + private void waitForConsumerStateNot(String queueName, long state) throws InterruptedException { int n = 0; - URI uri = UriComponentsBuilder - .fromUriString("http://localhost:15672/api/queues") - .pathSegment("{vhost}", "{queue}").buildAndExpand("/", queueName) - .encode().toUri(); - - Object consumers = null; - while (n++ < 100 && (consumers == null - || consumers.equals(Integer.valueOf(state)))) { - Map queueInfo = template.getForObject(uri, Map.class); - consumers = queueInfo.get("consumers"); - if (consumers == null || consumers.equals(Integer.valueOf(state))) { - Thread.sleep(100); - } + QueueInfo queue = client.getQueue("/", queueName); + while (n++ < 100 && (queue == null || queue.getConsumerCount() == state)) { + Thread.sleep(100); + queue = client.getQueue("/", queueName); } - assertThat(consumers).isNotNull(); - assertThat(n).withFailMessage( "Consumer state remained at " + state + " after 10 seconds") .isLessThan(100); diff --git a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java index 025c453fa..282916f4b 100644 --- a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java +++ b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java @@ -30,6 +30,9 @@ import java.util.concurrent.atomic.AtomicReference; 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.Rule; @@ -44,7 +47,6 @@ import org.springframework.amqp.core.AmqpTemplate; import org.springframework.amqp.core.AnonymousQueue; import org.springframework.amqp.core.BindingBuilder; import org.springframework.amqp.core.DirectExchange; -import org.springframework.amqp.core.Exchange; import org.springframework.amqp.core.ExchangeTypes; import org.springframework.amqp.core.MessageDeliveryMode; import org.springframework.amqp.core.Queue; @@ -435,9 +437,8 @@ public class RabbitBinderTests extends assertThat(container.isRunning()).isTrue(); consumerBinding.unbind(); assertThat(container.isRunning()).isFalse(); - org.springframework.amqp.rabbit.core.RabbitManagementTemplate rmt = new org.springframework.amqp.rabbit.core.RabbitManagementTemplate(); - List bindings = rmt - .getBindingsForExchange("/", exchange.getName()); + Client client = new Client("http://guest:guest@localhost:15672/api/"); + List bindings = client.getBindingsBySource("/", exchange.getName()); assertThat(bindings.size()).isEqualTo(1); } @@ -482,32 +483,24 @@ public class RabbitBinderTests extends consumerBinding.unbind(); assertThat(container.isRunning()).isFalse(); assertThat(container.getQueueNames()[0]).isEqualTo(group); - org.springframework.amqp.rabbit.core.RabbitManagementTemplate rmt = new org.springframework.amqp.rabbit.core.RabbitManagementTemplate(); - List bindings = rmt - .getBindingsForExchange("/", "propsUser2"); + Client client = new Client("http://guest:guest@localhost:15672/api/"); + List bindings = client.getBindingsBySource("/", "propsUser2"); int n = 0; while (n++ < 100 && bindings == null || bindings.size() < 1) { Thread.sleep(100); - bindings = rmt.getBindingsForExchange("/", "propsUser2"); + bindings = client.getBindingsBySource("/", "propsUser2"); } assertThat(bindings.size()).isEqualTo(1); - assertThat(bindings.get(0).getExchange()).isEqualTo("propsUser2"); + assertThat(bindings.get(0).getSource()).isEqualTo("propsUser2"); assertThat(bindings.get(0).getDestination()).isEqualTo(group); assertThat(bindings.get(0).getRoutingKey()).isEqualTo("foo"); - // // TODO: AMQP-696 - // // Exchange exchange = rmt.getExchange("propsUser2"); - // ExchangeInfo ei = rmt.getClient().getExchange("/", "propsUser2"); // requires - // delayed message exchange plugin - // assertThat(ei.getType()).isEqualTo("x-delayed-message"); - // assertThat(ei.getArguments().get("x-delayed-type")).isEqualTo("direct"); - - Exchange exchange = rmt.getExchange("propsUser2"); + ExchangeInfo exchange = client.getExchange("/", "propsUser2"); while (n++ < 100 && exchange == null) { Thread.sleep(100); - exchange = rmt.getExchange("propsUser2"); + exchange = client.getExchange("/", "propsUser2"); } - assertThat(exchange).isInstanceOf(DirectExchange.class); + assertThat(exchange.getType()).isEqualTo("direct"); assertThat(exchange.isDurable()).isEqualTo(true); assertThat(exchange.isAutoDelete()).isEqualTo(false); } @@ -554,44 +547,43 @@ public class RabbitBinderTests extends SimpleMessageListenerContainer container = TestUtils.getPropertyValue(endpoint, "messageListenerContainer", SimpleMessageListenerContainer.class); assertThat(container.isRunning()).isTrue(); - org.springframework.amqp.rabbit.core.RabbitManagementTemplate rmt = new org.springframework.amqp.rabbit.core.RabbitManagementTemplate(); - List bindings = rmt - .getBindingsForExchange("/", "propsUser3"); + Client client = new Client("http://guest:guest@localhost:15672/api"); + List bindings = client.getBindingsBySource("/", "propsUser3"); int n = 0; while (n++ < 100 && bindings == null || bindings.size() < 1) { Thread.sleep(100); - bindings = rmt.getBindingsForExchange("/", "propsUser3"); + bindings = client.getBindingsBySource("/", "propsUser3"); } assertThat(bindings.size()).isEqualTo(1); - assertThat(bindings.get(0).getExchange()).isEqualTo("propsUser3"); + assertThat(bindings.get(0).getSource()).isEqualTo("propsUser3"); assertThat(bindings.get(0).getDestination()).isEqualTo("propsUser3.infra"); assertThat(bindings.get(0).getRoutingKey()).isEqualTo("foo"); - Exchange exchange = rmt.getExchange("propsUser3"); + ExchangeInfo exchange = client.getExchange("/", "propsUser3"); n = 0; while (n++ < 100 && exchange == null) { Thread.sleep(100); - exchange = rmt.getExchange("propsUser3"); + exchange = client.getExchange("/", "propsUser3"); } - assertThat(exchange).isInstanceOf(DirectExchange.class); + assertThat(exchange.getType()).isEqualTo("direct"); assertThat(exchange.isDurable()).isEqualTo(false); assertThat(exchange.isAutoDelete()).isEqualTo(true); - exchange = rmt.getExchange("customDLX"); + exchange = client.getExchange("/", "customDLX"); n = 0; while (n++ < 100 && exchange == null) { Thread.sleep(100); - exchange = rmt.getExchange("customDLX"); + exchange = client.getExchange("/", "customDLX"); } - assertThat(exchange).isInstanceOf(TopicExchange.class); + assertThat(exchange.getType()).isEqualTo("topic"); assertThat(exchange.isDurable()).isEqualTo(true); assertThat(exchange.isAutoDelete()).isEqualTo(false); - QueueInfo queue = rmt.getClient().getQueue("/", "propsUser3.infra"); + QueueInfo queue = client.getQueue("/", "propsUser3.infra"); n = 0; while (n++ < 100 && queue == null || queue.getConsumerCount() == 0) { Thread.sleep(100); - queue = rmt.getClient().getQueue("/", "propsUser3.infra"); + queue = client.getQueue("/", "propsUser3.infra"); } assertThat(queue).isNotNull(); Map args = queue.getArguments(); @@ -606,12 +598,12 @@ public class RabbitBinderTests extends assertThat(args.get("x-queue-mode")).isEqualTo("lazy"); assertThat(queue.getExclusiveConsumerTag()).isEqualTo("testConsumerTag#0"); - queue = rmt.getClient().getQueue("/", "customDLQ"); + queue = client.getQueue("/", "customDLQ"); n = 0; while (n++ < 100 && queue == null) { Thread.sleep(100); - queue = rmt.getClient().getQueue("/", "customDLQ"); + queue = client.getQueue("/", "customDLQ"); } assertThat(queue).isNotNull(); args = queue.getArguments();