diff --git a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java index 430bd229a..2ebb20174 100644 --- a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java +++ b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java @@ -129,8 +129,6 @@ public class RabbitMessageChannelBinder private ConnectionFactory connectionFactory; - private ConnectionFactory producerConnectionFactory; - private MessagePostProcessor decompressingPostProcessor = new DelegatingDecompressingPostProcessor(); private MessagePostProcessor compressingPostProcessor = new GZipPostProcessor(); @@ -152,15 +150,6 @@ public class RabbitMessageChannelBinder this.rabbitProperties = rabbitProperties; } - /** - * Specify a distinct {@link ConnectionFactory} for the non-transactional producers to avoid dead locks - * on blocked connections. - * @param producerConnectionFactory the ConnectionFactory to use for non-transactional producers. - */ - public void setProducerConnectionFactory(ConnectionFactory producerConnectionFactory) { - this.producerConnectionFactory = producerConnectionFactory; - } - /** * Set a {@link MessagePostProcessor} to decompress messages. Defaults to a * {@link DelegatingDecompressingPostProcessor} with its default delegates. @@ -219,7 +208,6 @@ public class RabbitMessageChannelBinder if (this.destroyConnectionFactory) { ((DisposableBean) this.connectionFactory).destroy(); } - ((DisposableBean) this.producerConnectionFactory).destroy(); } } @@ -439,7 +427,11 @@ public class RabbitMessageChannelBinder return new MessageHandler() { private final RabbitTemplate template = new RabbitTemplate( - RabbitMessageChannelBinder.this.producerConnectionFactory); + RabbitMessageChannelBinder.this.connectionFactory); + + { + this.template.setUsePublisherConnection(true); + } private final String exchange = deadLetterExchangeName(properties.getExtension()); @@ -584,12 +576,8 @@ public class RabbitMessageChannelBinder rabbitTemplate = new RabbitTemplate(); } rabbitTemplate.setChannelTransacted(properties.isTransacted()); - if (rabbitTemplate.isChannelTransacted()) { - rabbitTemplate.setConnectionFactory(this.connectionFactory); - } - else { - rabbitTemplate.setConnectionFactory(this.producerConnectionFactory); - } + rabbitTemplate.setConnectionFactory(this.connectionFactory); + rabbitTemplate.setUsePublisherConnection(true); if (properties.isCompress()) { rabbitTemplate.setBeforePublishPostProcessors(this.compressingPostProcessor); } @@ -598,9 +586,9 @@ public class RabbitMessageChannelBinder Retry retry = rabbitProperties.getTemplate().getRetry(); RetryPolicy retryPolicy = new SimpleRetryPolicy(retry.getMaxAttempts()); ExponentialBackOffPolicy backOff = new ExponentialBackOffPolicy(); - backOff.setInitialInterval(retry.getInitialInterval()); + backOff.setInitialInterval(retry.getInitialInterval().toMillis()); backOff.setMultiplier(retry.getMultiplier()); - backOff.setMaxInterval(retry.getMaxInterval()); + backOff.setMaxInterval(retry.getMaxInterval().toMillis()); RetryTemplate retryTemplate = new RetryTemplate(); retryTemplate.setRetryPolicy(retryPolicy); retryTemplate.setBackOffPolicy(backOff); diff --git a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java index e80415adb..e3ead9dbd 100644 --- a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java +++ b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2017 the original author or authors. + * Copyright 2015-2018 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -16,12 +16,8 @@ package org.springframework.cloud.stream.binder.rabbit.config; -import java.time.Duration; - import org.springframework.amqp.core.MessagePostProcessor; -import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.connection.ConnectionFactory; -import org.springframework.amqp.rabbit.connection.RabbitConnectionFactoryBean; import org.springframework.amqp.support.postprocessor.DelegatingDecompressingPostProcessor; import org.springframework.amqp.support.postprocessor.GZipPostProcessor; import org.springframework.beans.factory.annotation.Autowired; @@ -44,6 +40,7 @@ import org.springframework.context.annotation.Import; * @author Vinicius Carvalho * @author Artem Bilan * @author Oleg Zhurakousky + * @author Gary Russell */ @Configuration @Import({ PropertyPlaceholderAutoConfiguration.class }) @@ -69,7 +66,6 @@ public class RabbitMessageChannelBinderConfiguration { RabbitMessageChannelBinder rabbitMessageChannelBinder() throws Exception { RabbitMessageChannelBinder binder = new RabbitMessageChannelBinder(this.rabbitConnectionFactory, this.rabbitProperties, provisioningProvider()); - binder.setProducerConnectionFactory(buildProducerConnectionFactory()); binder.setAdminAddresses(this.rabbitBinderConfigurationProperties.getAdminAddresses()); binder.setCompressingPostProcessor(gZipPostProcessor()); binder.setDecompressingPostProcessor(deCompressingPostProcessor()); @@ -78,66 +74,6 @@ public class RabbitMessageChannelBinderConfiguration { return binder; } - /** - * @see org.springframework.boot.autoconfigure.amqp.RabbitAutoConfiguration.RabbitConnectionFactoryCreator - */ - private CachingConnectionFactory buildProducerConnectionFactory() throws Exception { - com.rabbitmq.client.ConnectionFactory rabbitConnectionFactory; - if (this.rabbitConnectionFactory instanceof CachingConnectionFactory) { - rabbitConnectionFactory = ((CachingConnectionFactory) this.rabbitConnectionFactory) - .getRabbitConnectionFactory(); - } - else { - RabbitConnectionFactoryBean factory = new RabbitConnectionFactoryBean(); - String host = this.rabbitProperties.determineHost(); - if (host != null) { - factory.setHost(host); - } - factory.setPort(this.rabbitProperties.determinePort()); - String user = this.rabbitProperties.determineUsername(); - if (user != null) { - factory.setUsername(user); - } - String password = this.rabbitProperties.determinePassword(); - if (password != null) { - factory.setPassword(password); - } - String vHost = this.rabbitProperties.determineVirtualHost(); - if (vHost != null) { - factory.setVirtualHost(vHost); - } - Duration requestedHeartbeatDuration = this.rabbitProperties.getRequestedHeartbeat(); - if (requestedHeartbeatDuration != null) { - factory.setRequestedHeartbeat((int) requestedHeartbeatDuration.getSeconds()); - } - RabbitProperties.Ssl ssl = this.rabbitProperties.getSsl(); - if (ssl.isEnabled()) { - factory.setUseSSL(true); - if (ssl.getAlgorithm() != null) { - factory.setSslAlgorithm(ssl.getAlgorithm()); - } - factory.setKeyStore(ssl.getKeyStore()); - factory.setKeyStorePassphrase(ssl.getKeyStorePassword()); - factory.setTrustStore(ssl.getTrustStore()); - factory.setTrustStorePassphrase(ssl.getTrustStorePassword()); - } - Duration connectionTimeoutDuration = this.rabbitProperties.getConnectionTimeout(); - if (connectionTimeoutDuration != null) { - factory.setConnectionTimeout((int) connectionTimeoutDuration.getSeconds()); - } - factory.afterPropertiesSet(); - - rabbitConnectionFactory = factory.getObject(); - } - - CachingConnectionFactory connectionFactory = new CachingConnectionFactory(rabbitConnectionFactory); - - RabbitServiceAutoConfiguration.configureCachingConnectionFactory(connectionFactory, this.applicationContext, - this.rabbitProperties); - - return connectionFactory; - } - @Bean MessagePostProcessor deCompressingPostProcessor() { return new DelegatingDecompressingPostProcessor(); diff --git a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitServiceAutoConfiguration.java b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitServiceAutoConfiguration.java index d56ac2333..50336573b 100644 --- a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitServiceAutoConfiguration.java +++ b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitServiceAutoConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2017 the original author or authors. + * Copyright 2015-2018 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. @@ -52,6 +52,7 @@ import org.springframework.util.StringUtils; * @author Marius Bogoevici * @author Ilayaperumal Gopinathan * @author Artem Bilan + * @author Gary Russell */ @Configuration @ConditionalOnMissingBean(Binder.class) @@ -186,7 +187,7 @@ public class RabbitServiceAutoConfiguration { } if (rabbitProperties.getCache().getChannel().getCheckoutTimeout() != null) { connectionFactory.setChannelCheckoutTimeout( - rabbitProperties.getCache().getChannel().getCheckoutTimeout()); + rabbitProperties.getCache().getChannel().getCheckoutTimeout().toMillis()); } connectionFactory.setApplicationContext(applicationContext); applicationContext.addApplicationListener(connectionFactory); diff --git a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestBinder.java b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestBinder.java index 825ddc84d..e8010bf89 100644 --- a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestBinder.java +++ b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestBinder.java @@ -67,7 +67,6 @@ public class RabbitTestBinder extends public RabbitTestBinder(ConnectionFactory connectionFactory, RabbitMessageChannelBinder binder) { this.applicationContext = new AnnotationConfigApplicationContext(Config.class); binder.setApplicationContext(this.applicationContext); - binder.setProducerConnectionFactory(connectionFactory); this.setPollableConsumerBinder(binder); this.rabbitAdmin = new RabbitAdmin(connectionFactory); } diff --git a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java index 3b96633df..3693c2b46 100644 --- a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java +++ b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java @@ -16,11 +16,6 @@ package org.springframework.cloud.stream.binder.rabbit.integration; -import static org.assertj.core.api.Assertions.assertThat; -import static org.mockito.BDDMockito.willReturn; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.verify; - import java.util.ArrayList; import java.util.List; import java.util.Map; @@ -59,10 +54,16 @@ import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.integration.channel.DirectChannel; import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.support.GenericMessage; import org.springframework.retry.backoff.ExponentialBackOffPolicy; import org.springframework.retry.policy.SimpleRetryPolicy; import org.springframework.retry.support.RetryTemplate; +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.BDDMockito.willReturn; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; + /** * @author Marius Bogoevici * @author Gary Russell @@ -106,10 +107,6 @@ public class RabbitBinderModuleTests { ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class); assertThat(binderConnectionFactory).isSameAs(connectionFactory); - ConnectionFactory producerConnectionFactory = (ConnectionFactory) binderFieldAccessor - .getPropertyValue("producerConnectionFactory"); - assertThat(producerConnectionFactory).isNotSameAs(connectionFactory); - CompositeHealthIndicator bindersHealthIndicator = context.getBean("bindersHealthIndicator", CompositeHealthIndicator.class); DirectFieldAccessor directFieldAccessor = new DirectFieldAccessor(bindersHealthIndicator); @@ -119,6 +116,16 @@ public class RabbitBinderModuleTests { .getPropertyValue("indicators"); assertThat(healthIndicators).containsKey(("rabbit")); assertThat(healthIndicators.get("rabbit").health().getStatus()).isEqualTo((Status.UP)); + + ConnectionFactory publisherConnectionFactory = binderConnectionFactory.getPublisherConnectionFactory(); + assertThat(TestUtils.getPropertyValue(publisherConnectionFactory, "connection.target")).isNull(); + DirectChannel checkPf = new DirectChannel(); + Binding binding = ((RabbitMessageChannelBinder) binder).bindProducer("checkPF", checkPf, + new ExtendedProducerProperties<>( + new RabbitProducerProperties())); + checkPf.send(new GenericMessage<>("foo".getBytes())); + binding.unbind(); + assertThat(TestUtils.getPropertyValue(publisherConnectionFactory, "connection.target")).isNotNull(); } @Test @@ -254,16 +261,8 @@ public class RabbitBinderModuleTests { assertThat(binderConnectionFactory).isNotSameAs(connectionFactory); - ConnectionFactory producerConnectionFactory = (ConnectionFactory) binderFieldAccessor - .getPropertyValue("producerConnectionFactory"); - - assertThat(producerConnectionFactory).isNotSameAs(connectionFactory); - - assertThat(binderConnectionFactory).isNotSameAs(producerConnectionFactory); - assertThat(TestUtils.getPropertyValue(connectionFactory, "addresses")).isNotNull(); assertThat(TestUtils.getPropertyValue(binderConnectionFactory, "addresses")).isNull(); - assertThat(TestUtils.getPropertyValue(producerConnectionFactory, "addresses")).isNull(); Cloud cloud = this.context.getBean(Cloud.class);