From 702d23c8419fda1ed1d094208cd3dc019d6b29ba Mon Sep 17 00:00:00 2001 From: onobc Date: Thu, 27 Jan 2022 16:54:36 -0600 Subject: [PATCH] * Break dependence on RabbitAutoConfiguration. * Restore Rabbit own-connection feature --- .../app/sink/rabbit/OwnConnectionTest.java | 2 - .../rabbit/RabbitSourceListenerTests.java | 33 ++++++++- .../rabbit/RabbitConsumerConfiguration.java | 72 +++++++++++-------- .../rabbit/RabbitSupplierConfiguration.java | 67 +++++++++-------- 4 files changed, 112 insertions(+), 62 deletions(-) diff --git a/applications/sink/rabbit-sink/src/test/java/org/springframework/cloud/stream/app/sink/rabbit/OwnConnectionTest.java b/applications/sink/rabbit-sink/src/test/java/org/springframework/cloud/stream/app/sink/rabbit/OwnConnectionTest.java index c2e3aef5..d6680795 100644 --- a/applications/sink/rabbit-sink/src/test/java/org/springframework/cloud/stream/app/sink/rabbit/OwnConnectionTest.java +++ b/applications/sink/rabbit-sink/src/test/java/org/springframework/cloud/stream/app/sink/rabbit/OwnConnectionTest.java @@ -16,7 +16,6 @@ package org.springframework.cloud.stream.app.sink.rabbit; -import org.junit.jupiter.api.Disabled; import org.junit.jupiter.api.Test; import org.springframework.amqp.core.Message; @@ -31,7 +30,6 @@ import static org.assertj.core.api.Assertions.assertThat; public class OwnConnectionTest extends RabbitSinkIntegrationTests { @Test - @Disabled public void test() { this.rabbitAdmin.declareQueue( new Queue("scsapp-testOwn", false, false, true)); diff --git a/applications/source/rabbit-source/src/test/java/org/springframework/cloud/stream/app/source/rabbit/RabbitSourceListenerTests.java b/applications/source/rabbit-source/src/test/java/org/springframework/cloud/stream/app/source/rabbit/RabbitSourceListenerTests.java index ef6a06cf..6cd8f5fa 100644 --- a/applications/source/rabbit-source/src/test/java/org/springframework/cloud/stream/app/source/rabbit/RabbitSourceListenerTests.java +++ b/applications/source/rabbit-source/src/test/java/org/springframework/cloud/stream/app/source/rabbit/RabbitSourceListenerTests.java @@ -107,12 +107,13 @@ public class RabbitSourceListenerTests { "${spring.rabbitmq.test.port}" )) { - final RabbitTemplate rabbitTemplate = context.getBean(RabbitTemplate.class); + // Reset the boot connection factory -should not matter to container as it SHOULD be using its own connection factory final CachingConnectionFactory bootFactory = context.getBean(CachingConnectionFactory.class); - rabbitTemplate.convertAndSend("scsapp-testOwnSource", "foo"); - bootFactory.resetConnection(); + // Send a message on a separate connection - the container should still receive it. + sendMessageOnSeparateConnection("scsapp-testOwnSource", "foo", bootFactory); + OutputDestination target = context.getBean(OutputDestination.class); Message sourceMessage = target.receive(600000, "rabbitSupplier-out-0"); @@ -122,6 +123,32 @@ public class RabbitSourceListenerTests { } } + /** + * Sends a message on a separate connection. + * + * @param routingKey message routing key + * @param payload message content + * @param bootFactory the auto-configured connection factory used to get connection coordinates from + */ + private void sendMessageOnSeparateConnection(String routingKey, Object payload, CachingConnectionFactory bootFactory) { + CachingConnectionFactory copiedConnectionFactory = null; + try { + copiedConnectionFactory = new CachingConnectionFactory(bootFactory.getHost(), bootFactory.getPort()); + copiedConnectionFactory.setUsername(bootFactory.getUsername()); + copiedConnectionFactory.setPassword(bootFactory.getRabbitConnectionFactory().getPassword()); + if (bootFactory.getVirtualHost() != null) { + copiedConnectionFactory.setVirtualHost(bootFactory.getVirtualHost()); + } + RabbitTemplate rabbitTemplate = new RabbitTemplate(copiedConnectionFactory); + rabbitTemplate.convertAndSend(routingKey, payload); + } + finally { + if (copiedConnectionFactory != null) { + copiedConnectionFactory.resetConnection(); + } + } + } + @Test public void testPropertiesPopulated() { try (ConfigurableApplicationContext context = new SpringApplicationBuilder( diff --git a/functions/consumer/rabbit-consumer/src/main/java/org/springframework/cloud/fn/consumer/rabbit/RabbitConsumerConfiguration.java b/functions/consumer/rabbit-consumer/src/main/java/org/springframework/cloud/fn/consumer/rabbit/RabbitConsumerConfiguration.java index e6ce99fd..06b15247 100644 --- a/functions/consumer/rabbit-consumer/src/main/java/org/springframework/cloud/fn/consumer/rabbit/RabbitConsumerConfiguration.java +++ b/functions/consumer/rabbit-consumer/src/main/java/org/springframework/cloud/fn/consumer/rabbit/RabbitConsumerConfiguration.java @@ -24,7 +24,7 @@ import com.rabbitmq.client.impl.CredentialsRefreshService; import org.springframework.amqp.core.MessageDeliveryMode; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.connection.ConnectionFactory; -import org.springframework.amqp.rabbit.connection.ConnectionNameStrategy; +import org.springframework.amqp.rabbit.connection.RabbitConnectionFactoryBean; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter; import org.springframework.amqp.support.converter.MessageConverter; @@ -33,8 +33,9 @@ import org.springframework.beans.factory.ObjectProvider; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.beans.factory.annotation.Value; +import org.springframework.boot.autoconfigure.amqp.CachingConnectionFactoryConfigurer; import org.springframework.boot.autoconfigure.amqp.ConnectionFactoryCustomizer; -import org.springframework.boot.autoconfigure.amqp.RabbitAutoConfiguration; +import org.springframework.boot.autoconfigure.amqp.RabbitConnectionFactoryBeanConfigurer; import org.springframework.boot.autoconfigure.amqp.RabbitProperties; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.context.properties.EnableConfigurationProperties; @@ -47,6 +48,14 @@ import org.springframework.integration.amqp.dsl.AmqpOutboundChannelAdapterSpec; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandler; +/** + * A configuration for RabbitMQ Consumer function. Uses a + * {@link AmqpOutboundChannelAdapterSpec} to save payload contents to RabbitMQ. + * + * @author Soby Chako + * @author Nicolas Labrot + * @author Chris Bono + */ @EnableConfigurationProperties(RabbitConsumerProperties.class) @Configuration public class RabbitConsumerConfiguration implements DisposableBean { @@ -54,9 +63,6 @@ public class RabbitConsumerConfiguration implements DisposableBean { @Autowired private RabbitProperties bootProperties; - @Autowired - private ObjectProvider connectionNameStrategy; - @Autowired private ResourceLoader resourceLoader; @@ -86,12 +92,12 @@ public class RabbitConsumerConfiguration implements DisposableBean { } @Bean - public MessageHandler amqpChannelAdapter(ConnectionFactory rabbitConnectionFactory, CachingConnectionFactory cachingConnectionFactory) + public MessageHandler amqpChannelAdapter(ConnectionFactory rabbitConnectionFactory) throws Exception { AmqpOutboundChannelAdapterSpec handler = Amqp .outboundAdapter(rabbitTemplate(this.properties.isOwnConnection() - ? buildLocalConnectionFactory(cachingConnectionFactory) : rabbitConnectionFactory)) + ? buildLocalConnectionFactory() : rabbitConnectionFactory)) .mappedRequestHeaders(properties.getMappedRequestHeaders()) .defaultDeliveryMode(properties.getPersistentDeliveryMode() ? MessageDeliveryMode.PERSISTENT @@ -116,11 +122,6 @@ public class RabbitConsumerConfiguration implements DisposableBean { return handler.get(); } - private ConnectionFactory buildLocalConnectionFactory(CachingConnectionFactory cachingConnectionFactory) throws Exception { - this.ownConnectionFactory = new AutoConfig.Creator().rabbitConnectionFactory(cachingConnectionFactory); - return this.ownConnectionFactory; - } - @Bean public RabbitTemplate rabbitTemplate(ConnectionFactory rabbitConnectionFactory) { RabbitTemplate rabbitTemplate = new RabbitTemplate(rabbitConnectionFactory); @@ -130,6 +131,7 @@ public class RabbitConsumerConfiguration implements DisposableBean { return rabbitTemplate; } + @Bean @ConditionalOnProperty(name = "rabbit.converterBeanName", havingValue = RabbitConsumerProperties.JSON_CONVERTER) @@ -144,25 +146,39 @@ public class RabbitConsumerConfiguration implements DisposableBean { } } - private static class AutoConfig extends RabbitAutoConfiguration { + private ConnectionFactory buildLocalConnectionFactory() throws Exception { + this.ownConnectionFactory = rabbitConnectionFactory(this.bootProperties, this.resourceLoader, + this.credentialsProvider, this.credentialsRefreshService, this.connectionFactoryCustomizers); + return this.ownConnectionFactory; + } - static class Creator extends RabbitConnectionFactoryCreator { + private CachingConnectionFactory rabbitConnectionFactory(RabbitProperties properties, ResourceLoader resourceLoader, + ObjectProvider credentialsProvider, + ObjectProvider credentialsRefreshService, + ObjectProvider connectionFactoryCustomizers) throws Exception { -// @Override - public CachingConnectionFactory rabbitConnectionFactory(CachingConnectionFactory cachingConnectionFactory) - throws Exception { -// CachingConnectionFactory cf = super.rabbitConnectionFactory(config, resourceLoader, credentialsProvider, -// credentialsRefreshService, connectionNameStrategy, connectionFactoryCustomizers); - cachingConnectionFactory.setConnectionNameStrategy( - connectionFactory -> "rabbit.sink.own.connection"); - cachingConnectionFactory.afterPropertiesSet(); - return cachingConnectionFactory; - } + /* NOTE: This is based on RabbitAutoConfiguration.RabbitConnectionFactoryCreator + * https://github.com/spring-projects/spring-boot/blob/c820ad01a108d419d8548265b8a34ed7c5591f7c/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/RabbitAutoConfiguration.java#L95 + * [UPGRADE_CONSIDERATION] this should stay somewhat in sync w/ the functionality provided by its original source. + */ + RabbitConnectionFactoryBean connectionFactoryBean = new RabbitConnectionFactoryBean(); + RabbitConnectionFactoryBeanConfigurer connectionFactoryBeanConfigurer = new RabbitConnectionFactoryBeanConfigurer(resourceLoader, properties); + connectionFactoryBeanConfigurer.setCredentialsProvider(credentialsProvider.getIfUnique()); + connectionFactoryBeanConfigurer.setCredentialsRefreshService(credentialsRefreshService.getIfUnique()); + connectionFactoryBeanConfigurer.configure(connectionFactoryBean); + connectionFactoryBean.afterPropertiesSet(); - } + com.rabbitmq.client.ConnectionFactory connectionFactory = connectionFactoryBean.getObject(); + connectionFactoryCustomizers.orderedStream() + .forEach((customizer) -> customizer.customize(connectionFactory)); + + CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory(connectionFactory); + CachingConnectionFactoryConfigurer cachingConnectionFactoryConfigurer = new CachingConnectionFactoryConfigurer(properties); + cachingConnectionFactoryConfigurer.setConnectionNameStrategy(cf -> "rabbit.sink.own.connection"); + cachingConnectionFactoryConfigurer.configure(cachingConnectionFactory); + cachingConnectionFactory.afterPropertiesSet(); + + return cachingConnectionFactory; } } - - - diff --git a/functions/supplier/rabbit-supplier/src/main/java/org/springframework/cloud/fn/supplier/rabbit/RabbitSupplierConfiguration.java b/functions/supplier/rabbit-supplier/src/main/java/org/springframework/cloud/fn/supplier/rabbit/RabbitSupplierConfiguration.java index 9b7e1d2e..6506b025 100644 --- a/functions/supplier/rabbit-supplier/src/main/java/org/springframework/cloud/fn/supplier/rabbit/RabbitSupplierConfiguration.java +++ b/functions/supplier/rabbit-supplier/src/main/java/org/springframework/cloud/fn/supplier/rabbit/RabbitSupplierConfiguration.java @@ -30,7 +30,7 @@ import org.springframework.amqp.core.MessageProperties; import org.springframework.amqp.rabbit.config.RetryInterceptorBuilder; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.connection.ConnectionFactory; -import org.springframework.amqp.rabbit.connection.ConnectionNameStrategy; +import org.springframework.amqp.rabbit.connection.RabbitConnectionFactoryBean; import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; import org.springframework.amqp.rabbit.retry.RejectAndDontRequeueRecoverer; import org.springframework.amqp.rabbit.support.DefaultMessagePropertiesConverter; @@ -38,8 +38,9 @@ import org.springframework.amqp.rabbit.support.MessagePropertiesConverter; import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.ObjectProvider; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.amqp.CachingConnectionFactoryConfigurer; import org.springframework.boot.autoconfigure.amqp.ConnectionFactoryCustomizer; -import org.springframework.boot.autoconfigure.amqp.RabbitAutoConfiguration; +import org.springframework.boot.autoconfigure.amqp.RabbitConnectionFactoryBeanConfigurer; import org.springframework.boot.autoconfigure.amqp.RabbitProperties; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.context.annotation.Bean; @@ -57,6 +58,7 @@ import org.springframework.util.Assert; * @author Gary Russell * @author Chris Schaefer * @author Roger Perez + * @author Chris Bono */ @EnableConfigurationProperties(RabbitSupplierProperties.class) public class RabbitSupplierConfiguration implements DisposableBean { @@ -66,8 +68,8 @@ public class RabbitSupplierConfiguration implements DisposableBean { @Override public MessageProperties toMessageProperties(AMQP.BasicProperties source, - Envelope envelope, - String charset) { + Envelope envelope, + String charset) { MessageProperties properties = super.toMessageProperties(source, envelope, charset); properties.setDeliveryMode(null); return properties; @@ -77,9 +79,6 @@ public class RabbitSupplierConfiguration implements DisposableBean { @Autowired private RabbitProperties rabbitProperties; - @Autowired - private ObjectProvider connectionNameStrategy; - @Autowired private ResourceLoader resourceLoader; @@ -101,9 +100,9 @@ public class RabbitSupplierConfiguration implements DisposableBean { private CachingConnectionFactory ownConnectionFactory; @Bean - public SimpleMessageListenerContainer container(CachingConnectionFactory cf) { + public SimpleMessageListenerContainer container() { ConnectionFactory connectionFactory = this.properties.isOwnConnection() - ? buildLocalConnectionFactory(cf) + ? buildLocalConnectionFactory() : this.rabbitConnectionFactory; SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory); container.setAutoStartup(false); @@ -129,7 +128,6 @@ public class RabbitSupplierConfiguration implements DisposableBean { if (transactionSize != null) { container.setBatchSize(transactionSize); } - container.setDefaultRequeueRejected(this.properties.getRequeue()); container.setChannelTransacted(this.properties.getTransacted()); String[] queues = this.properties.getQueues(); @@ -146,9 +144,9 @@ public class RabbitSupplierConfiguration implements DisposableBean { @Bean public Publisher> rabbitPublisher(SimpleMessageListenerContainer container) { return IntegrationFlows.from( - Amqp.inboundAdapter(container) - .autoStartup(false) - .mappedRequestHeaders(properties.getMappedRequestHeaders())) + Amqp.inboundAdapter(container) + .autoStartup(false) + .mappedRequestHeaders(properties.getMappedRequestHeaders())) .toReactivePublisher(); } @@ -176,9 +174,11 @@ public class RabbitSupplierConfiguration implements DisposableBean { } } - private ConnectionFactory buildLocalConnectionFactory(CachingConnectionFactory cf) { + private ConnectionFactory buildLocalConnectionFactory() { try { - this.ownConnectionFactory = new AutoConfig.Creator().rabbitConnectionFactory(cf); + this.ownConnectionFactory = rabbitConnectionFactory( + this.rabbitProperties, this.resourceLoader, this.credentialsProvider, this.credentialsRefreshService, + this.connectionFactoryCustomizers); } catch (Exception exception) { @@ -187,24 +187,33 @@ public class RabbitSupplierConfiguration implements DisposableBean { return this.ownConnectionFactory; } -} -class AutoConfig extends RabbitAutoConfiguration { + /* NOTE: This is based on RabbitAutoConfiguration.RabbitConnectionFactoryCreator + * https://github.com/spring-projects/spring-boot/blob/c820ad01a108d419d8548265b8a34ed7c5591f7c/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/RabbitAutoConfiguration.java#L95 + * [UPGRADE_CONSIDERATION] this should stay somewhat in sync w/ the functionality provided by its original source. + */ + private CachingConnectionFactory rabbitConnectionFactory(RabbitProperties properties, ResourceLoader resourceLoader, + ObjectProvider credentialsProvider, + ObjectProvider credentialsRefreshService, + ObjectProvider connectionFactoryCustomizers) throws Exception { - static class Creator extends RabbitConnectionFactoryCreator { + RabbitConnectionFactoryBean connectionFactoryBean = new RabbitConnectionFactoryBean(); + RabbitConnectionFactoryBeanConfigurer connectionFactoryBeanConfigurer = new RabbitConnectionFactoryBeanConfigurer(resourceLoader, properties); + connectionFactoryBeanConfigurer.setCredentialsProvider(credentialsProvider.getIfUnique()); + connectionFactoryBeanConfigurer.setCredentialsRefreshService(credentialsRefreshService.getIfUnique()); + connectionFactoryBeanConfigurer.configure(connectionFactoryBean); + connectionFactoryBean.afterPropertiesSet(); - public CachingConnectionFactory rabbitConnectionFactory(CachingConnectionFactory cf) - throws Exception { + com.rabbitmq.client.ConnectionFactory connectionFactory = connectionFactoryBean.getObject(); + connectionFactoryCustomizers.orderedStream() + .forEach((customizer) -> customizer.customize(connectionFactory)); - cf.setConnectionNameStrategy(new ConnectionNameStrategy() { + CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory(connectionFactory); + CachingConnectionFactoryConfigurer cachingConnectionFactoryConfigurer = new CachingConnectionFactoryConfigurer(properties); + cachingConnectionFactoryConfigurer.setConnectionNameStrategy(cf -> "rabbit.supplier.own.connection"); + cachingConnectionFactoryConfigurer.configure(cachingConnectionFactory); + cachingConnectionFactory.afterPropertiesSet(); - @Override - public String obtainNewConnectionName(ConnectionFactory connectionFactory) { - return "rabbit.supplier.own.connection"; - } - }); - cf.afterPropertiesSet(); - return cf; - } + return cachingConnectionFactory; } }