diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-test-support/src/main/java/org/springframework/cloud/stream/binder/test/junit/rabbit/RabbitTestSupport.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-test-support/src/main/java/org/springframework/cloud/stream/binder/test/junit/rabbit/RabbitTestSupport.java index 37c59f69a..989042ec1 100644 --- a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-test-support/src/main/java/org/springframework/cloud/stream/binder/test/junit/rabbit/RabbitTestSupport.java +++ b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-test-support/src/main/java/org/springframework/cloud/stream/binder/test/junit/rabbit/RabbitTestSupport.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2019 the original author or authors. + * Copyright 2015-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. @@ -44,22 +44,31 @@ public class RabbitTestSupport extends AbstractExternalResourceTestSupport { private final boolean management; + private final int ampqPort; + private final int managmentPort; public RabbitTestSupport() { this(false); } public RabbitTestSupport(boolean management) { + this(management, 5672, 15672); + } + + public RabbitTestSupport(boolean management, int amqpPort, int managementPort) { super("RABBIT"); this.management = management; + this.ampqPort = amqpPort; + this.managmentPort = managementPort; + } @Override protected void obtainResource() throws Exception { - resource = new CachingConnectionFactory("localhost"); + resource = new CachingConnectionFactory("localhost", this.ampqPort); resource.createConnection().close(); if (management) { - Socket socket = SocketFactory.getDefault().createSocket("localhost", 15672); + Socket socket = SocketFactory.getDefault().createSocket("localhost", this.managmentPort); socket.close(); } } @@ -96,6 +105,10 @@ public class RabbitTestSupport } public void start() throws IOException { + start(5672); + } + + public void start(int amqpPort) throws IOException { this.serverSocket = ServerSocketFactory.getDefault() .createServerSocket(this.port, 10); LOGGER.info("Proxy started"); @@ -114,7 +127,7 @@ public class RabbitTestSupport try { final Socket rabbitSocket = SocketFactory .getDefault() - .createSocket("localhost", 5672); + .createSocket("localhost", amqpPort); socketExec.execute(new Runnable() { @Override diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/LocalizedQueueConnectionFactoryIntegrationTests.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/LocalizedQueueConnectionFactoryIntegrationTests.java index acb13785c..27878051e 100644 --- a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/LocalizedQueueConnectionFactoryIntegrationTests.java +++ b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/LocalizedQueueConnectionFactoryIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2016 the original author or authors. + * Copyright 2015-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. @@ -21,6 +21,7 @@ import java.util.UUID; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.RegisterExtension; +import org.testcontainers.containers.RabbitMQContainer; import org.springframework.amqp.core.Queue; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; @@ -34,20 +35,22 @@ import static org.assertj.core.api.Assertions.assertThat; /** * @author Gary Russell + * @author Chris Bono */ public class LocalizedQueueConnectionFactoryIntegrationTests { + private static final RabbitMQContainer RABBITMQ = RabbitTestContainer.sharedInstance(); + @RegisterExtension - public static RabbitTestSupport rabbitAvailableRule = new RabbitTestSupport(true); + private RabbitTestSupport rabbitTestSupport = new RabbitTestSupport(true, RABBITMQ.getAmqpPort(), RABBITMQ.getHttpPort()); private LocalizedQueueConnectionFactory lqcf; @BeforeEach public void setup() { - ConnectionFactory defaultConnectionFactory = rabbitAvailableRule.getResource(); - String[] addresses = new String[] { "localhost:9999", "localhost:5672" }; - String[] adminAddresses = new String[] { "http://localhost:15672", - "http://localhost:15672" }; + ConnectionFactory defaultConnectionFactory = rabbitTestSupport.getResource(); + String[] addresses = new String[] { "localhost:9999", "localhost:" + RABBITMQ.getAmqpPort() }; + String[] adminAddresses = new String[] { RABBITMQ.getHttpUrl(), RABBITMQ.getHttpUrl() }; String[] nodes = new String[] { "foo@bar", "rabbit@localhost" }; String vhost = "/"; String username = "guest"; 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 dc5e5cf7e..f7472b47c 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 @@ -1,5 +1,5 @@ /* - * Copyright 2015-2019 the original author or authors. + * Copyright 2015-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. @@ -22,13 +22,13 @@ import java.util.List; import java.util.Map; 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.jupiter.api.Test; import org.junit.jupiter.api.extension.RegisterExtension; +import org.testcontainers.containers.RabbitMQContainer; import org.springframework.amqp.core.Base64UrlNamingStrategy; import org.springframework.amqp.core.BindingBuilder; @@ -48,17 +48,19 @@ import static org.junit.Assert.fail; /** * @author Gary Russell + * @author Chris Bono * @since 1.2 */ public class RabbitBinderCleanerTests { + private static final RabbitMQContainer RABBITMQ = RabbitTestContainer.sharedInstance(); + private static final String BINDER_PREFIX = "binder."; private static final Client client; - static { try { - client = new Client("http://localhost:15672/api", "guest", "guest"); + client = new Client(RABBITMQ.getHttpUrl() + "/api", "guest", "guest"); } catch (MalformedURLException | URISyntaxException e) { throw new RabbitAdminException("Couldn't create a Client", e); @@ -66,7 +68,7 @@ public class RabbitBinderCleanerTests { } @RegisterExtension - public RabbitTestSupport rabbitWithMgmtEnabled = new RabbitTestSupport(true); + private RabbitTestSupport rabbitTestSupport = new RabbitTestSupport(true, RABBITMQ.getAmqpPort(), RABBITMQ.getHttpPort()); @Test public void testCleanStream() { @@ -74,7 +76,7 @@ public class RabbitBinderCleanerTests { final String stream1 = new Base64UrlNamingStrategy("foo").generateName(); String stream2 = stream1 + "-1"; String firstQueue = null; - CachingConnectionFactory connectionFactory = rabbitWithMgmtEnabled.getResource(); + CachingConnectionFactory connectionFactory = rabbitTestSupport.getResource(); RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory); for (int i = 0; i < 5; i++) { String queue1Name = AbstractBinder.applyPrefix(BINDER_PREFIX, @@ -120,7 +122,7 @@ public class RabbitBinderCleanerTests { new DefaultConsumer(channel)); try { waitForConsumerStateNot(queueName, 0); - cleaner.clean(stream1, false); + doClean(cleaner, stream1, false); fail("Expected exception"); } catch (RabbitAdminException e) { @@ -130,7 +132,7 @@ public class RabbitBinderCleanerTests { channel.basicCancel(consumerTag); waitForConsumerStateNot(queueName, 1); try { - cleaner.clean(stream1, false); + doClean(cleaner, stream1, false); fail("Expected exception"); } catch (RabbitAdminException e) { @@ -157,7 +159,7 @@ public class RabbitBinderCleanerTests { rabbitAdmin.declareExchange(topic1); rabbitAdmin.deleteQueue(foreignQueue); connectionFactory.destroy(); - Map> cleanedMap = cleaner.clean(stream1, false); + Map> cleanedMap = doClean(cleaner, stream1, false); assertThat(cleanedMap).hasSize(2); List cleanedQueues = cleanedMap.get("queues"); // should *not* clean stream2 @@ -172,7 +174,7 @@ public class RabbitBinderCleanerTests { assertThat(cleanedExchanges).hasSize(6); // wild card *should* clean stream2 - cleanedMap = cleaner.clean(stream1 + "*", false); + cleanedMap = doClean(cleaner, stream1 + "*", false); assertThat(cleanedMap).hasSize(2); cleanedQueues = cleanedMap.get("queues"); assertThat(cleanedQueues).hasSize(5); @@ -184,34 +186,8 @@ public class RabbitBinderCleanerTests { assertThat(cleanedExchanges).hasSize(6); } - public static class AmqpQueue { - - private boolean autoDelete; - - private boolean durable; - - public AmqpQueue(boolean autoDelete, boolean durable) { - this.autoDelete = autoDelete; - this.durable = durable; - } - - @JsonProperty("auto_delete") - protected boolean isAutoDelete() { - return autoDelete; - } - - protected void setAutoDelete(boolean autoDelete) { - this.autoDelete = autoDelete; - } - - protected boolean isDurable() { - return durable; - } - - protected void setDurable(boolean durable) { - this.durable = durable; - } - + private static Map> doClean(RabbitBindingCleaner cleaner, String entity, boolean isJob) { + return cleaner.clean(RABBITMQ.getHttpUrl() + "/api", "guest", "guest", "/", 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 fe2724561..1ee65618d 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 @@ -49,6 +49,7 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.TestInfo; import org.junit.jupiter.api.extension.RegisterExtension; import org.mockito.ArgumentCaptor; +import org.testcontainers.containers.RabbitMQContainer; import org.springframework.amqp.AmqpIOException; import org.springframework.amqp.ImmediateAcknowledgeAmqpException; @@ -147,30 +148,30 @@ import static org.mockito.Mockito.when; * @author Gary Russell * @author David Turanski * @author Artem Bilan + * @author Chris Bono */ // @checkstyle:off public class RabbitBinderTests extends PartitionCapableBinderTests, ExtendedProducerProperties> { - private final String CLASS_UNDER_TEST_NAME = RabbitMessageChannelBinder.class - .getSimpleName(); + private static final RabbitMQContainer RABBITMQ = RabbitTestContainer.sharedInstance(); - public static final String TEST_PREFIX = "bindertest."; + private static final String CLASS_UNDER_TEST_NAME = RabbitMessageChannelBinder.class.getSimpleName(); - private static final String BIG_EXCEPTION_MESSAGE = new String(new byte[10_000]) - .replaceAll("\u0000", "x"); - - private int maxStackTraceSize; + private static final String TEST_PREFIX = "bindertest."; + private static final String BIG_EXCEPTION_MESSAGE = new String(new byte[10_000]).replaceAll("\u0000", "x"); @RegisterExtension - RabbitTestSupport rabbitAvailableRule = new RabbitTestSupport(true); + private RabbitTestSupport rabbitTestSupport = new RabbitTestSupport(true, RABBITMQ.getAmqpPort(), RABBITMQ.getHttpPort()); + + private int maxStackTraceSize; @Override protected RabbitTestBinder getBinder() { if (this.testBinder == null) { RabbitProperties rabbitProperties = new RabbitProperties(); - this.testBinder = new RabbitTestBinder(this.rabbitAvailableRule.getResource(), rabbitProperties); + this.testBinder = new RabbitTestBinder(this.rabbitTestSupport.getResource(), rabbitProperties); } return this.testBinder; } @@ -249,7 +250,7 @@ public class RabbitBinderTests extends @Test public void testProducerErrorChannel(TestInfo testInfo) throws Exception { RabbitTestBinder binder = getBinder(); - CachingConnectionFactory ccf = this.rabbitAvailableRule.getResource(); + CachingConnectionFactory ccf = this.rabbitTestSupport.getResource(); ccf.setPublisherReturns(true); ccf.setPublisherConfirmType(ConfirmType.CORRELATED); ccf.resetConnection(); @@ -332,7 +333,7 @@ public class RabbitBinderTests extends @Test public void testProducerAckChannel(TestInfo testInfo) throws Exception { RabbitTestBinder binder = getBinder(); - CachingConnectionFactory ccf = this.rabbitAvailableRule.getResource(); + CachingConnectionFactory ccf = this.rabbitTestSupport.getResource(); ccf.setPublisherReturns(true); ccf.setPublisherConfirmType(ConfirmType.CORRELATED); ccf.resetConnection(); @@ -363,7 +364,7 @@ public class RabbitBinderTests extends @Test public void testProducerConfirmHeader(TestInfo testInfo) throws Exception { RabbitTestBinder binder = getBinder(); - CachingConnectionFactory ccf = this.rabbitAvailableRule.getResource(); + CachingConnectionFactory ccf = this.rabbitTestSupport.getResource(); ccf.setPublisherReturns(true); ccf.setPublisherConfirmType(ConfirmType.CORRELATED); ccf.resetConnection(); @@ -509,7 +510,7 @@ public class RabbitBinderTests extends @Test public void testConsumerPropertiesWithUserInfrastructureNoBind() throws Exception { - RabbitAdmin admin = new RabbitAdmin(this.rabbitAvailableRule.getResource()); + RabbitAdmin admin = new RabbitAdmin(this.rabbitTestSupport.getResource()); Queue queue = new Queue("propsUser1.infra"); admin.declareQueue(queue); DirectExchange exchange = new DirectExchange("propsUser1"); @@ -532,7 +533,7 @@ public class RabbitBinderTests extends assertThat(container.isRunning()).isTrue(); consumerBinding.unbind(); assertThat(container.isRunning()).isFalse(); - Client client = new Client("http://guest:guest@localhost:15672/api/"); + Client client = new Client(adminUri()); List bindings = client.getBindingsBySource("/", exchange.getName()); assertThat(bindings.size()).isEqualTo(1); } @@ -604,7 +605,7 @@ public class RabbitBinderTests extends consumerBinding.unbind(); assertThat(container.isRunning()).isFalse(); assertThat(container.getQueueNames()[0]).isEqualTo(group); - Client client = new Client("http://guest:guest@localhost:15672/api/"); + Client client = new Client(adminUri()); List bindings = client.getBindingsBySource("/", "propsUser2"); int n = 0; while (n++ < 100 && bindings == null || bindings.size() < 1) { @@ -632,6 +633,9 @@ public class RabbitBinderTests extends verifyAutoDeclareContextClear(binder); } + private String adminUri() { + return String.format("http://guest:guest@localhost:%d/api", RABBITMQ.getHttpPort()); + } @Test public void testConsumerPropertiesWithUserInfrastructureCustomQueueArgs() @@ -675,7 +679,7 @@ public class RabbitBinderTests extends SimpleMessageListenerContainer container = TestUtils.getPropertyValue(endpoint, "messageListenerContainer", SimpleMessageListenerContainer.class); assertThat(container.isRunning()).isTrue(); - Client client = new Client("http://guest:guest@localhost:15672/api"); + Client client = new Client(adminUri()); List bindings = client.getBindingsBySource("/", "propsUser3"); int n = 0; while (n++ < 100 && bindings == null || bindings.size() < 1) { @@ -787,7 +791,7 @@ public class RabbitBinderTests extends consumerBinding.unbind(); assertThat(container.isRunning()).isFalse(); assertThat(container.getQueueNames()[0]).isEqualTo("propsHeader." + group); - Client client = new Client("http://guest:guest@localhost:15672/api/"); + Client client = new Client(adminUri()); List bindings = client.getBindingsBySource("/", "propsHeader"); int n = 0; while (n++ < 100 && bindings == null || bindings.size() < 1) { @@ -865,7 +869,7 @@ public class RabbitBinderTests extends producerBinding, "lifecycle.amqpTemplate.connectionFactory", ConnectionFactory.class); - assertThat(this.rabbitAvailableRule.getResource()) + assertThat(this.rabbitTestSupport.getResource()) .isSameAs(producerConnectionFactory); endpoint = extractEndpoint(producerBinding); @@ -882,7 +886,7 @@ public class RabbitBinderTests extends verifyFooRequestProducer(endpoint); channel.send(new GenericMessage<>("foo")); org.springframework.amqp.core.Message received = new RabbitTemplate( - this.rabbitAvailableRule.getResource()) + this.rabbitTestSupport.getResource()) .receive("foo.props.0.prodPropsRequired-0", 10_000); assertThat(received).isNotNull(); assertThat(received.getMessageProperties().getReceivedDelay()).isEqualTo(42); @@ -896,7 +900,7 @@ public class RabbitBinderTests extends @Test public void testDurablePubSubWithAutoBindDLQ() throws Exception { - RabbitAdmin admin = new RabbitAdmin(this.rabbitAvailableRule.getResource()); + RabbitAdmin admin = new RabbitAdmin(this.rabbitTestSupport.getResource()); RabbitTestBinder binder = getBinder(); @@ -920,7 +924,7 @@ public class RabbitBinderTests extends "tgroup", moduleInputChannel, consumerProperties); RabbitTemplate template = new RabbitTemplate( - this.rabbitAvailableRule.getResource()); + this.rabbitTestSupport.getResource()); template.convertAndSend(TEST_PREFIX + "durabletest.0", "", "foo"); int n = 0; @@ -945,7 +949,7 @@ public class RabbitBinderTests extends @Test public void testNonDurablePubSubWithAutoBindDLQ() throws Exception { - RabbitAdmin admin = new RabbitAdmin(this.rabbitAvailableRule.getResource()); + RabbitAdmin admin = new RabbitAdmin(this.rabbitTestSupport.getResource()); RabbitTestBinder binder = getBinder(); ExtendedConsumerProperties consumerProperties = createConsumerProperties(); @@ -1007,7 +1011,7 @@ public class RabbitBinderTests extends assertThat(container.getQueueNames().length).isEqualTo(2); RabbitTemplate template = new RabbitTemplate( - this.rabbitAvailableRule.getResource()); + this.rabbitTestSupport.getResource()); template.convertAndSend("", TEST_PREFIX + "dlqtest.default", "foo"); int n = 0; @@ -1066,7 +1070,7 @@ public class RabbitBinderTests extends DirectChannel moduleInputChannel = createBindableChannel("input", bindingProperties); moduleInputChannel.setBeanName("dlqTestManual"); - Client client = new Client("http://guest:guest@localhost:15672/api"); + Client client = new Client(adminUri()); moduleInputChannel.subscribe(new MessageHandler() { @Override @@ -1091,7 +1095,7 @@ public class RabbitBinderTests extends "default", moduleInputChannel, consumerProperties); RabbitTemplate template = new RabbitTemplate( - this.rabbitAvailableRule.getResource()); + this.rabbitTestSupport.getResource()); template.convertAndSend("", TEST_PREFIX + "dlqTestManual.default", "foo"); int n = 0; @@ -1205,7 +1209,7 @@ public class RabbitBinderTests extends output.send(new GenericMessage<>(1)); RabbitTemplate template = new RabbitTemplate( - this.rabbitAvailableRule.getResource()); + this.rabbitTestSupport.getResource()); template.setReceiveTimeout(10000); String streamDLQName = "bindertest.partDLQ.0.dlqPartGrp.dlq"; @@ -1356,7 +1360,7 @@ public class RabbitBinderTests extends output.send(new GenericMessage<>(1)); RabbitTemplate template = new RabbitTemplate( - this.rabbitAvailableRule.getResource()); + this.rabbitTestSupport.getResource()); template.setReceiveTimeout(10000); String streamDLQName = "bindertest.partPubDLQ.0.dlqPartGrp.dlq"; @@ -1470,7 +1474,7 @@ public class RabbitBinderTests extends output.send(new GenericMessage(1)); RabbitTemplate template = new RabbitTemplate( - this.rabbitAvailableRule.getResource()); + this.rabbitTestSupport.getResource()); template.setReceiveTimeout(10000); String streamDLQName = "bindertest.partDLQ.1.dlqPartGrp.dlq"; @@ -1505,7 +1509,7 @@ public class RabbitBinderTests extends @Test public void testAutoBindDLQwithRepublish() throws Exception { this.maxStackTraceSize = RabbitUtils - .getMaxFrame(rabbitAvailableRule.getResource()) - 20_000; + .getMaxFrame(rabbitTestSupport.getResource()) - 20_000; assertThat(this.maxStackTraceSize).isGreaterThan(0); RabbitTestBinder binder = getBinder(); @@ -1540,7 +1544,7 @@ public class RabbitBinderTests extends consumerProperties); RabbitTemplate template = new RabbitTemplate( - this.rabbitAvailableRule.getResource()); + this.rabbitTestSupport.getResource()); template.convertAndSend("", TEST_PREFIX + "foo.dlqpubtest.foo", "foo"); template.setReceiveTimeout(10_000); @@ -1598,7 +1602,7 @@ public class RabbitBinderTests extends Binding consumerBinding = binder.bindConsumer( "foo.dlqpubtestTx", "foo", moduleInputChannel, consumerProperties); - RabbitTemplate template = new RabbitTemplate(this.rabbitAvailableRule.getResource()); + RabbitTemplate template = new RabbitTemplate(this.rabbitTestSupport.getResource()); template.convertAndSend("", TEST_PREFIX + "foo.dlqpubtestTx.foo", "foo"); template.setReceiveTimeout(10_000); @@ -1625,7 +1629,7 @@ public class RabbitBinderTests extends @SuppressWarnings("unchecked") @Test public void testAutoBindDLQwithRepublishSimpleConfirms() throws Exception { - CachingConnectionFactory ccf = this.rabbitAvailableRule.getResource(); + CachingConnectionFactory ccf = this.rabbitTestSupport.getResource(); ccf.setPublisherReturns(true); ccf.setPublisherConfirmType(ConfirmType.SIMPLE); ccf.resetConnection(); @@ -1650,7 +1654,7 @@ public class RabbitBinderTests extends Binding consumerBinding = binder.bindConsumer( "foo.dlqpubtestSimple", "foo", moduleInputChannel, consumerProperties); - RabbitTemplate template = new RabbitTemplate(this.rabbitAvailableRule.getResource()); + RabbitTemplate template = new RabbitTemplate(this.rabbitTestSupport.getResource()); template.convertAndSend("", TEST_PREFIX + "foo.dlqpubtestSimple.foo", "foo"); template.setReceiveTimeout(10_000); @@ -1676,7 +1680,7 @@ public class RabbitBinderTests extends @SuppressWarnings("unchecked") @Test public void testAutoBindDLQwithRepublishCorrelatedConfirms() throws Exception { - CachingConnectionFactory ccf = this.rabbitAvailableRule.getResource(); + CachingConnectionFactory ccf = this.rabbitTestSupport.getResource(); ccf.setPublisherReturns(true); ccf.setPublisherConfirmType(ConfirmType.CORRELATED); ccf.resetConnection(); @@ -1701,7 +1705,7 @@ public class RabbitBinderTests extends Binding consumerBinding = binder.bindConsumer( "foo.dlqpubtestCorrelated", "foo", moduleInputChannel, consumerProperties); - RabbitTemplate template = new RabbitTemplate(this.rabbitAvailableRule.getResource()); + RabbitTemplate template = new RabbitTemplate(this.rabbitTestSupport.getResource()); template.convertAndSend("", TEST_PREFIX + "foo.dlqpubtestCorrelated.foo", "foo"); template.setReceiveTimeout(10_000); @@ -1894,11 +1898,11 @@ public class RabbitBinderTests extends ExtendedConsumerProperties consumerProperties = createConsumerProperties(); Binding consumerBinding = binder.bindConsumer("propagate.0", "propagate", input, consumerProperties); - RabbitAdmin admin = new RabbitAdmin(rabbitAvailableRule.getResource()); + RabbitAdmin admin = new RabbitAdmin(rabbitTestSupport.getResource()); admin.declareQueue(new Queue("propagate")); admin.declareBinding(new org.springframework.amqp.core.Binding("propagate", DestinationType.QUEUE, "propagate.1", "#", null)); - RabbitTemplate template = new RabbitTemplate(this.rabbitAvailableRule.getResource()); + RabbitTemplate template = new RabbitTemplate(this.rabbitTestSupport.getResource()); template.convertAndSend("propagate.0.propagate", "foo"); output.send(input.receive(10_000)); org.springframework.amqp.core.Message received = template.receive("propagate", 10_000); @@ -1997,7 +2001,7 @@ public class RabbitBinderTests extends Binding durableConsumerBinding = binder.bindConsumer("latePubSub", "lateDurableGroup", durablePubSubInputChannel, noDlqConsumerProperties); - proxy.start(); + proxy.start(RABBITMQ.getAmqpPort()); moduleOutputChannel.send(MessageBuilder.withPayload("foo") .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN) @@ -2052,7 +2056,7 @@ public class RabbitBinderTests extends proxy.stop(); cf.destroy(); - this.rabbitAvailableRule.getResource().destroy(); + this.rabbitTestSupport.getResource().destroy(); verifyAutoDeclareContextClear(binder); } @@ -2071,10 +2075,10 @@ public class RabbitBinderTests extends bf.initializeBean(provisioner, "provisioner"); bf.registerSingleton("provisioner", provisioner); context.addApplicationListener(provisioner); - RabbitAdmin admin = new RabbitAdmin(rabbitAvailableRule.getResource()); + RabbitAdmin admin = new RabbitAdmin(rabbitTestSupport.getResource()); admin.declareQueue(new Queue("testBadUserDeclarationsFatal")); // reset the connection and configure the "user" admin to auto declare queues... - rabbitAvailableRule.getResource().resetConnection(); + rabbitTestSupport.getResource().resetConnection(); bf.initializeBean(admin, "rabbitAdmin"); bf.registerSingleton("rabbitAdmin", admin); admin.afterPropertiesSet(); @@ -2113,7 +2117,7 @@ public class RabbitBinderTests extends Binding producerBinding = binder.bindProducer("rke", output, producerProperties); - RabbitAdmin admin = new RabbitAdmin(this.rabbitAvailableRule.getResource()); + RabbitAdmin admin = new RabbitAdmin(this.rabbitTestSupport.getResource()); Queue queue = new AnonymousQueue(); TopicExchange exchange = new TopicExchange("rke"); org.springframework.amqp.core.Binding binding = BindingBuilder.bind(queue) @@ -2159,7 +2163,7 @@ public class RabbitBinderTests extends Binding producerBinding = binder.bindProducer("rke", output, producerProperties); - RabbitAdmin admin = new RabbitAdmin(this.rabbitAvailableRule.getResource()); + RabbitAdmin admin = new RabbitAdmin(this.rabbitTestSupport.getResource()); Queue queue = new AnonymousQueue(); DirectExchange exchange = new DirectExchange("rke"); org.springframework.amqp.core.Binding binding = BindingBuilder.bind(queue) @@ -2199,7 +2203,7 @@ public class RabbitBinderTests extends Binding producerBinding = binder.bindProducer("rkep", output, producerProperties); - RabbitAdmin admin = new RabbitAdmin(this.rabbitAvailableRule.getResource()); + RabbitAdmin admin = new RabbitAdmin(this.rabbitTestSupport.getResource()); Queue queue = new AnonymousQueue(); TopicExchange exchange = new TopicExchange("rkep"); org.springframework.amqp.core.Binding binding = BindingBuilder.bind(queue) @@ -2243,7 +2247,7 @@ public class RabbitBinderTests extends Binding> binding = binder.bindPollableConsumer( "pollable", "group", inboundBindTarget, createConsumerProperties()); RabbitTemplate template = new RabbitTemplate( - this.rabbitAvailableRule.getResource()); + this.rabbitTestSupport.getResource()); template.convertAndSend("pollable.group", "testPollable"); boolean polled = inboundBindTarget.poll(m -> { assertThat(m.getPayload()).isEqualTo("testPollable"); @@ -2270,7 +2274,7 @@ public class RabbitBinderTests extends Binding> binding = binder.bindPollableConsumer( "pollableRequeue", "group", inboundBindTarget, properties); RabbitTemplate template = new RabbitTemplate( - this.rabbitAvailableRule.getResource()); + this.rabbitTestSupport.getResource()); template.convertAndSend("pollableRequeue.group", "testPollable"); try { boolean polled = false; @@ -2307,7 +2311,7 @@ public class RabbitBinderTests extends Binding> binding = binder.bindPollableConsumer( "pollableDlq", "group", inboundBindTarget, properties); RabbitTemplate template = new RabbitTemplate( - this.rabbitAvailableRule.getResource()); + this.rabbitTestSupport.getResource()); template.convertAndSend("pollableDlq.group", "testPollable"); try { int n = 0; @@ -2344,7 +2348,7 @@ public class RabbitBinderTests extends Binding> binding = binder.bindPollableConsumer( "pollableDlqNoRetry", "group", inboundBindTarget, properties); RabbitTemplate template = new RabbitTemplate( - this.rabbitAvailableRule.getResource()); + this.rabbitTestSupport.getResource()); template.convertAndSend("pollableDlqNoRetry.group", "testPollable"); try { int n = 0; @@ -2380,7 +2384,7 @@ public class RabbitBinderTests extends Binding> binding = binder.bindPollableConsumer( "pollableDlqRePub", "group", inboundBindTarget, properties); RabbitTemplate template = new RabbitTemplate( - this.rabbitAvailableRule.getResource()); + this.rabbitTestSupport.getResource()); template.convertAndSend("pollableDlqRePub.group", "testPollable"); boolean polled = false; int n = 0; @@ -2512,7 +2516,7 @@ public class RabbitBinderTests extends @Override public Spy spyOn(final String queue) { final RabbitTemplate template = new RabbitTemplate( - this.rabbitAvailableRule.getResource()); + this.rabbitTestSupport.getResource()); template.setAfterReceivePostProcessors( new DelegatingDecompressingPostProcessor()); return new Spy() { diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestContainer.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestContainer.java new file mode 100644 index 000000000..26df3f449 --- /dev/null +++ b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestContainer.java @@ -0,0 +1,52 @@ +/* + * Copyright 2022-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.time.Duration; + +import org.testcontainers.containers.RabbitMQContainer; + +/** + * Provides a static {@link RabbitMQContainer} that can be shared across test classes. + * + * @author Chris Bono + */ +public class RabbitTestContainer { + + private static final RabbitMQContainer RABBITMQ; + static { + String image = "rabbitmq:management"; + String cache = System.getenv().get("IMAGE_CACHE"); + if (cache != null) { + image = cache + image; + } + RABBITMQ = new RabbitMQContainer(image) + .withExposedPorts(5672, 15672, 5552) + .withEnv("RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS", "-rabbitmq_stream advertised_host localhost") + .withPluginsEnabled("rabbitmq_stream") + .withStartupTimeout(Duration.ofMinutes(2)); + RABBITMQ.start(); + } + + /** + * Should be called early by test that wants to ensure a shared {@link RabbitMQContainer} is up and running. + */ + public static RabbitMQContainer sharedInstance() { + return RABBITMQ; + } + +} diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RepublishUnitTests.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RepublishUnitTests.java index 873348d63..65f65e8ca 100644 --- a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RepublishUnitTests.java +++ b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RepublishUnitTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2020-2020 the original author or authors. + * Copyright 2020-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. @@ -20,7 +20,7 @@ import java.io.IOException; import java.util.Collections; import com.rabbitmq.client.Channel; -import org.junit.Test; +import org.junit.jupiter.api.Test; import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessageProperties; @@ -62,7 +62,7 @@ public class RepublishUnitTests { RabbitMessageChannelBinder binder = new RabbitMessageChannelBinder(cf, props, null); RabbitConsumerProperties extension = new RabbitConsumerProperties(); ExtendedConsumerProperties bindingProps = - new ExtendedConsumerProperties(extension); + new ExtendedConsumerProperties<>(extension); MessageHandler handler = binder.getErrorMessageHandler(mock(ConsumerDestination.class), "foo", bindingProps); ErrorMessage message = new ErrorMessage(new RuntimeException("test"), Collections.singletonMap(IntegrationMessageHeaderAccessor.SOURCE_DATA, 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 56e78fdb1..2e1b36b1b 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 @@ -1,5 +1,5 @@ /* - * Copyright 2015-2019 the original author or authors. + * Copyright 2015-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. @@ -32,6 +32,7 @@ import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.RegisterExtension; import org.mockito.Mockito; +import org.testcontainers.containers.RabbitMQContainer; import org.springframework.amqp.core.DeclarableCustomizer; import org.springframework.amqp.core.ExchangeTypes; @@ -58,6 +59,7 @@ import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; 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.properties.RabbitConsumerProperties; import org.springframework.cloud.stream.binder.rabbit.properties.RabbitProducerProperties; import org.springframework.cloud.stream.binder.test.junit.rabbit.RabbitTestSupport; @@ -89,17 +91,19 @@ import static org.mockito.Mockito.verify; * @author Gary Russell * @author Artem Bilan * @author Soby Chacko + * @author Chris Bono */ public class RabbitBinderModuleTests { + private static final RabbitMQContainer RABBITMQ = RabbitTestContainer.sharedInstance(); + + private static final ConnectionFactory MOCK_CONNECTION_FACTORY = mock(ConnectionFactory.class, Mockito.RETURNS_MOCKS); + @RegisterExtension - public static RabbitTestSupport rabbitTestSupport = new RabbitTestSupport(); + private RabbitTestSupport rabbitTestSupport = new RabbitTestSupport(true, RABBITMQ.getAmqpPort(), RABBITMQ.getHttpPort()); private ConfigurableApplicationContext context; - public static final ConnectionFactory MOCK_CONNECTION_FACTORY = mock( - ConnectionFactory.class, Mockito.RETURNS_MOCKS); - @AfterEach public void tearDown() { if (context != null) { @@ -114,47 +118,48 @@ public class RabbitBinderModuleTests { @Test public void testParentConnectionFactoryInheritedByDefault() throws Exception { context = new SpringApplicationBuilder(SimpleProcessor.class) - .web(WebApplicationType.NONE).run("--server.port=0", - "--spring.cloud.function.definition=process", - "--spring.cloud.stream.rabbit.binder.connection-name-prefix=foo", - "--spring.cloud.stream.rabbit.bindings.process-in-0.consumer.single-active-consumer=true"); + .web(WebApplicationType.NONE).run("--server.port=0", + "--spring.rabbitmq.port=" + RABBITMQ.getAmqpPort(), + "--spring.cloud.function.definition=process", + "--spring.cloud.stream.rabbit.binder.connection-name-prefix=foo", + "--spring.cloud.stream.rabbit.bindings.process-in-0.consumer.single-active-consumer=true"); BinderFactory binderFactory = context.getBean(BinderFactory.class); Binder binder = binderFactory.getBinder(null, MessageChannel.class); assertThat(binder).isInstanceOf(RabbitMessageChannelBinder.class); DirectFieldAccessor binderFieldAccessor = new DirectFieldAccessor(binder); CachingConnectionFactory binderConnectionFactory = (CachingConnectionFactory) binderFieldAccessor - .getPropertyValue("connectionFactory"); + .getPropertyValue("connectionFactory"); assertThat(binderConnectionFactory).isInstanceOf(CachingConnectionFactory.class); ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class); assertThat(binderConnectionFactory).isSameAs(connectionFactory); CompositeHealthContributor bindersHealthIndicator = context - .getBean("bindersHealthContributor", CompositeHealthContributor.class); + .getBean("bindersHealthContributor", CompositeHealthContributor.class); assertThat(bindersHealthIndicator).isNotNull(); RabbitHealthIndicator indicator = (RabbitHealthIndicator) bindersHealthIndicator.getContributor("rabbit"); assertThat(indicator).isNotNull(); assertThat(indicator.health().getStatus()) - .isEqualTo(Status.UP); + .isEqualTo(Status.UP); ConnectionFactory publisherConnectionFactory = binderConnectionFactory - .getPublisherConnectionFactory(); + .getPublisherConnectionFactory(); assertThat(TestUtils.getPropertyValue(publisherConnectionFactory, - "connection.target")).isNull(); + "connection.target")).isNull(); DirectChannel checkPf = new DirectChannel(); Binding binding = ((RabbitMessageChannelBinder) binder) - .bindProducer("checkPF", checkPf, - new ExtendedProducerProperties<>(new RabbitProducerProperties())); + .bindProducer("checkPF", checkPf, + new ExtendedProducerProperties<>(new RabbitProducerProperties())); checkPf.send(new GenericMessage<>("foo".getBytes())); binding.unbind(); assertThat(TestUtils.getPropertyValue(publisherConnectionFactory, - "connection.target")).isNotNull(); + "connection.target")).isNotNull(); CachingConnectionFactory cf = this.context - .getBean(CachingConnectionFactory.class); + .getBean(CachingConnectionFactory.class); ConnectionNameStrategy cns = TestUtils.getPropertyValue(cf, - "connectionNameStrategy", ConnectionNameStrategy.class); + "connectionNameStrategy", ConnectionNameStrategy.class); assertThat(cns.obtainNewConnectionName(cf)).isEqualTo("foo#2"); new RabbitAdmin(rabbitTestSupport.getResource()).deleteExchange("checkPF"); checkCustomizedArgs(); @@ -164,7 +169,7 @@ public class RabbitBinderModuleTests { } private void checkCustomizedArgs() throws MalformedURLException, URISyntaxException, InterruptedException { - Client client = new Client("http://guest:guest@localhost:15672/api"); + Client client = new Client(String.format("http://guest:guest@localhost:%d/api", RABBITMQ.getHttpPort())); List bindings = client.getBindingsBySource("/", "process-in-0"); int n = 0; while (n++ < 100 && bindings == null || bindings.size() < 1) { @@ -184,59 +189,60 @@ public class RabbitBinderModuleTests { @SuppressWarnings("unchecked") public void testParentConnectionFactoryInheritedByDefaultAndRabbitSettingsPropagated() { context = new SpringApplicationBuilder(SimpleProcessor.class) - .web(WebApplicationType.NONE).run("--server.port=0", - "--spring.cloud.function.definition=process", - "--spring.cloud.stream.bindings.source.group=someGroup", - "--spring.cloud.stream.bindings.process-in-0.group=someGroup", - "--spring.cloud.stream.rabbit.bindings.process-in-0.consumer.transacted=true", - "--spring.cloud.stream.rabbit.bindings.process-out-0.producer.transacted=true"); + .web(WebApplicationType.NONE).run("--server.port=0", + "--spring.rabbitmq.port=" + RABBITMQ.getAmqpPort(), + "--spring.cloud.function.definition=process", + "--spring.cloud.stream.bindings.source.group=someGroup", + "--spring.cloud.stream.bindings.process-in-0.group=someGroup", + "--spring.cloud.stream.rabbit.bindings.process-in-0.consumer.transacted=true", + "--spring.cloud.stream.rabbit.bindings.process-out-0.producer.transacted=true"); BinderFactory binderFactory = context.getBean(BinderFactory.class); Binder binder = binderFactory.getBinder(null, MessageChannel.class); assertThat(binder).isInstanceOf(RabbitMessageChannelBinder.class); BindingService bindingService = context.getBean(BindingService.class); DirectFieldAccessor channelBindingServiceAccessor = new DirectFieldAccessor( - bindingService); + bindingService); // @checkstyle:off Map>> consumerBindings = (Map>>) channelBindingServiceAccessor - .getPropertyValue("consumerBindings"); + .getPropertyValue("consumerBindings"); // @checkstyle:on Binding inputBinding = consumerBindings.get("process-in-0").get(0); assertThat(TestUtils.getPropertyValue(inputBinding, "lifecycle.beanName")) - .isEqualTo("setByCustomizer:someGroup"); + .isEqualTo("setByCustomizer:someGroup"); SimpleMessageListenerContainer container = TestUtils.getPropertyValue( - inputBinding, "lifecycle.messageListenerContainer", - SimpleMessageListenerContainer.class); + inputBinding, "lifecycle.messageListenerContainer", + SimpleMessageListenerContainer.class); assertThat(TestUtils.getPropertyValue(container, "beanName")) - .isEqualTo("setByCustomizerForQueue:process-in-0.someGroup,andGroup:someGroup"); + .isEqualTo("setByCustomizerForQueue:process-in-0.someGroup,andGroup:someGroup"); assertThat(TestUtils.getPropertyValue(container, "transactional", Boolean.class)) - .isTrue(); + .isTrue(); Map> producerBindings = (Map>) TestUtils - .getPropertyValue(bindingService, "producerBindings"); + .getPropertyValue(bindingService, "producerBindings"); Binding outputBinding = producerBindings.get("process-out-0"); assertThat(TestUtils.getPropertyValue(outputBinding, - "lifecycle.amqpTemplate.transactional", Boolean.class)).isTrue(); + "lifecycle.amqpTemplate.transactional", Boolean.class)).isTrue(); assertThat(TestUtils.getPropertyValue(outputBinding, "lifecycle.beanName")) - .isEqualTo("setByCustomizer:process-out-0"); + .isEqualTo("setByCustomizer:process-out-0"); DirectFieldAccessor binderFieldAccessor = new DirectFieldAccessor(binder); ConnectionFactory binderConnectionFactory = (ConnectionFactory) binderFieldAccessor - .getPropertyValue("connectionFactory"); + .getPropertyValue("connectionFactory"); assertThat(binderConnectionFactory).isInstanceOf(CachingConnectionFactory.class); ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class); assertThat(binderConnectionFactory).isSameAs(connectionFactory); CompositeHealthContributor bindersHealthIndicator = context - .getBean("bindersHealthContributor", CompositeHealthContributor.class); + .getBean("bindersHealthContributor", CompositeHealthContributor.class); assertThat(bindersHealthIndicator).isNotNull(); RabbitHealthIndicator indicator = (RabbitHealthIndicator) bindersHealthIndicator.getContributor("rabbit"); assertThat(indicator).isNotNull(); assertThat(indicator.health().getStatus()) - .isEqualTo(Status.UP); + .isEqualTo(Status.UP); CachingConnectionFactory cf = this.context - .getBean(CachingConnectionFactory.class); + .getBean(CachingConnectionFactory.class); ConnectionNameStrategy cns = TestUtils.getPropertyValue(cf, - "connectionNameStrategy", ConnectionNameStrategy.class); + "connectionNameStrategy", ConnectionNameStrategy.class); assertThat(cns.obtainNewConnectionName(cf)).startsWith("rabbitConnectionFactory"); // assertThat(TestUtils.getPropertyValue(consumerBindings.get("source").get(0), // "target.source.h.advised.targetSource.target.beanName")) @@ -246,25 +252,25 @@ public class RabbitBinderModuleTests { @Test public void testParentConnectionFactoryInheritedIfOverridden() { context = new SpringApplicationBuilder(SimpleProcessor.class, - ConnectionFactoryConfiguration.class).web(WebApplicationType.NONE) - .run("--server.port=0"); + ConnectionFactoryConfiguration.class).web(WebApplicationType.NONE) + .run("--server.port=0", "--spring.rabbitmq.port=" + RABBITMQ.getAmqpPort()); BinderFactory binderFactory = context.getBean(BinderFactory.class); Binder binder = binderFactory.getBinder(null, MessageChannel.class); assertThat(binder).isInstanceOf(RabbitMessageChannelBinder.class); DirectFieldAccessor binderFieldAccessor = new DirectFieldAccessor(binder); ConnectionFactory binderConnectionFactory = (ConnectionFactory) binderFieldAccessor - .getPropertyValue("connectionFactory"); + .getPropertyValue("connectionFactory"); assertThat(binderConnectionFactory).isSameAs(MOCK_CONNECTION_FACTORY); ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class); assertThat(binderConnectionFactory).isSameAs(connectionFactory); CompositeHealthContributor bindersHealthIndicator = context - .getBean("bindersHealthContributor", CompositeHealthContributor.class); + .getBean("bindersHealthContributor", CompositeHealthContributor.class); assertThat(bindersHealthIndicator).isNotNull(); RabbitHealthIndicator indicator = (RabbitHealthIndicator) bindersHealthIndicator.getContributor("rabbit"); assertThat(indicator).isNotNull(); // mock connection factory behaves as if down assertThat(indicator.health().getStatus()) - .isEqualTo(Status.DOWN); + .isEqualTo(Status.DOWN); } @Test @@ -276,43 +282,44 @@ public class RabbitBinderModuleTests { params.add("--spring.cloud.stream.binders.custom.type=rabbit"); params.add("--spring.cloud.stream.binders.custom.environment.foo=bar"); params.add("--server.port=0"); + params.add("--spring.rabbitmq.port=" + RABBITMQ.getAmqpPort()); params.add("--spring.rabbitmq.template.retry.enabled=true"); params.add("--spring.rabbitmq.template.retry.maxAttempts=2"); params.add("--spring.rabbitmq.template.retry.initial-interval=1000"); params.add("--spring.rabbitmq.template.retry.multiplier=1.1"); params.add("--spring.rabbitmq.template.retry.max-interval=3000"); context = new SpringApplicationBuilder(SimpleProcessor.class) - .web(WebApplicationType.NONE) - .run(params.toArray(new String[params.size()])); + .web(WebApplicationType.NONE) + .run(params.toArray(new String[0])); BinderFactory binderFactory = context.getBean(BinderFactory.class); // @checkstyle:off @SuppressWarnings("unchecked") Binder, ExtendedProducerProperties> binder = (Binder, ExtendedProducerProperties>) binderFactory - .getBinder(null, MessageChannel.class); + .getBinder(null, MessageChannel.class); // @checkstyle:on assertThat(binder).isInstanceOf(RabbitMessageChannelBinder.class); DirectFieldAccessor binderFieldAccessor = new DirectFieldAccessor(binder); ConnectionFactory binderConnectionFactory = (ConnectionFactory) binderFieldAccessor - .getPropertyValue("connectionFactory"); + .getPropertyValue("connectionFactory"); ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class); assertThat(binderConnectionFactory).isNotSameAs(connectionFactory); CompositeHealthContributor bindersHealthIndicator = context - .getBean("bindersHealthContributor", CompositeHealthContributor.class); - assertThat(bindersHealthIndicator); + .getBean("bindersHealthContributor", CompositeHealthContributor.class); + assertThat(bindersHealthIndicator).isNotNull(); RabbitHealthIndicator indicator = (RabbitHealthIndicator) bindersHealthIndicator.getContributor("custom"); assertThat(indicator).isNotNull(); assertThat(indicator.health().getStatus()).isEqualTo(Status.UP); String name = UUID.randomUUID().toString(); Binding binding = binder.bindProducer(name, new DirectChannel(), - new ExtendedProducerProperties<>(new RabbitProducerProperties())); + new ExtendedProducerProperties<>(new RabbitProducerProperties())); RetryTemplate template = TestUtils.getPropertyValue(binding, - "lifecycle.amqpTemplate.retryTemplate", RetryTemplate.class); + "lifecycle.amqpTemplate.retryTemplate", RetryTemplate.class); assertThat(template).isNotNull(); SimpleRetryPolicy retryPolicy = TestUtils.getPropertyValue(template, - "retryPolicy", SimpleRetryPolicy.class); + "retryPolicy", SimpleRetryPolicy.class); ExponentialBackOffPolicy backOff = TestUtils.getPropertyValue(template, - "backOffPolicy", ExponentialBackOffPolicy.class); + "backOffPolicy", ExponentialBackOffPolicy.class); assertThat(retryPolicy.getMaxAttempts()).isEqualTo(2); assertThat(backOff.getInitialInterval()).isEqualTo(1000L); assertThat(backOff.getMultiplier()).isEqualTo(1.1); @@ -325,23 +332,23 @@ public class RabbitBinderModuleTests { @Test public void testCloudProfile() { this.context = new SpringApplicationBuilder(SimpleProcessor.class, - MockCloudConfiguration.class).web(WebApplicationType.NONE) - .profiles("cloud").run(); + MockCloudConfiguration.class).web(WebApplicationType.NONE) + .profiles("cloud").run(); BinderFactory binderFactory = this.context.getBean(BinderFactory.class); Binder binder = binderFactory.getBinder(null, MessageChannel.class); assertThat(binder).isInstanceOf(RabbitMessageChannelBinder.class); DirectFieldAccessor binderFieldAccessor = new DirectFieldAccessor(binder); ConnectionFactory binderConnectionFactory = (ConnectionFactory) binderFieldAccessor - .getPropertyValue("connectionFactory"); + .getPropertyValue("connectionFactory"); ConnectionFactory connectionFactory = this.context - .getBean(ConnectionFactory.class); + .getBean(ConnectionFactory.class); assertThat(binderConnectionFactory).isNotSameAs(connectionFactory); assertThat(TestUtils.getPropertyValue(connectionFactory, "addresses")) - .isNotNull(); + .isNotNull(); assertThat(TestUtils.getPropertyValue(binderConnectionFactory, "addresses")) - .isNull(); + .isNull(); Cloud cloud = this.context.getBean(Cloud.class); @@ -351,30 +358,31 @@ public class RabbitBinderModuleTests { @Test public void testExtendedProperties() { context = new SpringApplicationBuilder(SimpleProcessor.class) - .web(WebApplicationType.NONE).run("--server.port=0", - "--spring.cloud.function.definition=process", - "--spring.cloud.stream.rabbit.default.producer.routing-key-expression=fooRoutingKey", - "--spring.cloud.stream.rabbit.default.consumer.exchange-type=direct", - "--spring.cloud.stream.rabbit.bindings.process-out-0.producer.batch-size=512", - "--spring.cloud.stream.rabbit.default.consumer.max-concurrency=4", - "--spring.cloud.stream.rabbit.bindings.process-in-0.consumer.exchange-type=fanout"); + .web(WebApplicationType.NONE).run("--server.port=0", + "--spring.rabbitmq.port=" + RABBITMQ.getAmqpPort(), + "--spring.cloud.function.definition=process", + "--spring.cloud.stream.rabbit.default.producer.routing-key-expression=fooRoutingKey", + "--spring.cloud.stream.rabbit.default.consumer.exchange-type=direct", + "--spring.cloud.stream.rabbit.bindings.process-out-0.producer.batch-size=512", + "--spring.cloud.stream.rabbit.default.consumer.max-concurrency=4", + "--spring.cloud.stream.rabbit.bindings.process-in-0.consumer.exchange-type=fanout"); BinderFactory binderFactory = context.getBean(BinderFactory.class); Binder rabbitBinder = binderFactory.getBinder(null, - MessageChannel.class); + MessageChannel.class); RabbitProducerProperties rabbitProducerProperties = (RabbitProducerProperties) ((ExtendedPropertiesBinder) rabbitBinder) - .getExtendedProducerProperties("process-out-0"); + .getExtendedProducerProperties("process-out-0"); assertThat( - rabbitProducerProperties.getRoutingKeyExpression().getExpressionString()) - .isEqualTo("fooRoutingKey"); + rabbitProducerProperties.getRoutingKeyExpression().getExpressionString()) + .isEqualTo("fooRoutingKey"); assertThat(rabbitProducerProperties.getBatchSize()).isEqualTo(512); RabbitConsumerProperties rabbitConsumerProperties = (RabbitConsumerProperties) ((ExtendedPropertiesBinder) rabbitBinder) - .getExtendedConsumerProperties("process-in-0"); + .getExtendedConsumerProperties("process-in-0"); assertThat(rabbitConsumerProperties.getExchangeType()) - .isEqualTo(ExchangeTypes.FANOUT); + .isEqualTo(ExchangeTypes.FANOUT); assertThat(rabbitConsumerProperties.getMaxConcurrency()).isEqualTo(4); } @@ -384,7 +392,7 @@ public class RabbitBinderModuleTests { @Bean public ListenerContainerCustomizer containerCustomizer() { return (c, q, g) -> ((AbstractMessageListenerContainer) c).setBeanName( - "setByCustomizerForQueue:" + q + (g == null ? "" : ",andGroup:" + g)); + "setByCustomizerForQueue:" + q + (g == null ? "" : ",andGroup:" + g)); } @Bean @@ -432,7 +440,7 @@ public class RabbitBinderModuleTests { Cloud cloud = mock(Cloud.class); willReturn(new CachingConnectionFactory("localhost")).given(cloud) - .getSingletonServiceConnector(ConnectionFactory.class, null); + .getSingletonServiceConnector(ConnectionFactory.class, null); return cloud; } diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/stream/AbstractIntegrationTests.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/stream/AbstractIntegrationTests.java deleted file mode 100644 index ebcb6d4f6..000000000 --- a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/stream/AbstractIntegrationTests.java +++ /dev/null @@ -1,62 +0,0 @@ -/* - * Copyright 2021-2021 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.stream; - -import java.time.Duration; - -import org.testcontainers.containers.GenericContainer; -import org.testcontainers.utility.DockerImageName; - -/** - * @author Gary Russell - * @since 3.2 - * - */ -public abstract class AbstractIntegrationTests { - - static final GenericContainer RABBITMQ; - - static { - if (System.getProperty("spring.rabbit.use.local.server") == null) { - String image = "pivotalrabbitmq/rabbitmq-stream"; - String cache = System.getenv().get("IMAGE_CACHE"); - if (cache != null) { - image = cache + image; - } - RABBITMQ = new GenericContainer<>(DockerImageName.parse(image)) - .withExposedPorts(5672, 15672, 5552) - .withStartupTimeout(Duration.ofMinutes(2)); - RABBITMQ.start(); - } - else { - RABBITMQ = null; - } - } - - static int amqpPort() { - return RABBITMQ != null ? RABBITMQ.getMappedPort(5672) : 5672; - } - - static int managementPort() { - return RABBITMQ != null ? RABBITMQ.getMappedPort(15672) : 15672; - } - - static int streamPort() { - return RABBITMQ != null ? RABBITMQ.getMappedPort(5552) : 5552; - } - -} diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/stream/RabbitStreamMessageHandlerTests.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/stream/RabbitStreamMessageHandlerTests.java index 780ea5f4b..575de2ccd 100644 --- a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/stream/RabbitStreamMessageHandlerTests.java +++ b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/stream/RabbitStreamMessageHandlerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2021-2021 the original author or authors. + * Copyright 2021-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. @@ -26,8 +26,10 @@ import com.rabbitmq.stream.Environment; import com.rabbitmq.stream.OffsetSpecification; import org.junit.jupiter.api.Disabled; import org.junit.jupiter.api.Test; +import org.testcontainers.containers.RabbitMQContainer; import org.springframework.cloud.stream.binder.rabbit.RabbitStreamMessageHandler; +import org.springframework.cloud.stream.binder.rabbit.RabbitTestContainer; import org.springframework.integration.support.MessageBuilder; import org.springframework.rabbit.stream.producer.RabbitStreamTemplate; @@ -35,16 +37,18 @@ import static org.assertj.core.api.Assertions.assertThat; /** * @author Gary Russell + * @author Chris Bono * @since 3.2 - * */ -public class RabbitStreamMessageHandlerTests extends AbstractIntegrationTests { +public class RabbitStreamMessageHandlerTests { + + private static final RabbitMQContainer RABBITMQ = RabbitTestContainer.sharedInstance(); @Test void convertAndSend() throws InterruptedException { Environment env = Environment.builder() .lazyInitialization(true) - .addressResolver(add -> new Address("localhost", streamPort())) + .addressResolver(add -> new Address("localhost", RABBITMQ.getMappedPort(5552))) .build(); try { env.deleteStream("stream.stream"); diff --git a/bom/spring-cloud-starter-parent/pom.xml b/bom/spring-cloud-starter-parent/pom.xml index 4bf57cd72..a3ed4b420 100644 --- a/bom/spring-cloud-starter-parent/pom.xml +++ b/bom/spring-cloud-starter-parent/pom.xml @@ -7,6 +7,7 @@ org.springframework.boot spring-boot-starter-parent 3.0.0-M2 + org.springframework.cloud spring-cloud-stream-starter-parent