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

* Break dependence on RabbitAutoConfiguration.

Browse Source 
* Restore Rabbit own-connection feature
This commit is contained in:
onobc
2022-01-27 16:54:36 -06:00
parent 1116040c26
commit 702d23c841
4 changed files with 112 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
applications/sink/rabbit-sink/src/test/java/org/springframework/cloud/stream/app/sink/rabbit/OwnConnectionTest.java
View File

@@ -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));

33
applications/source/rabbit-source/src/test/java/org/springframework/cloud/stream/app/source/rabbit/RabbitSourceListenerTests.java
View File

@@ -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<byte[]> 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(

72
functions/consumer/rabbit-consumer/src/main/java/org/springframework/cloud/fn/consumer/rabbit/RabbitConsumerConfiguration.java
View File

@@ -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> 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> credentialsProvider,
ObjectProvider<CredentialsRefreshService> credentialsRefreshService,
ObjectProvider<ConnectionFactoryCustomizer> 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;
}
}

67
functions/supplier/rabbit-supplier/src/main/java/org/springframework/cloud/fn/supplier/rabbit/RabbitSupplierConfiguration.java
View File

@@ -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> 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<Message<byte[]>> 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> credentialsProvider,
ObjectProvider<CredentialsRefreshService> credentialsRefreshService,
ObjectProvider<ConnectionFactoryCustomizer> 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;
}
}