From fa035296f2338d024fcf5f1178d26e69d3a4c360 Mon Sep 17 00:00:00 2001
From: Gary Russell <grussell@pivotal.io>
Date: Thu, 17 Mar 2016 16:11:27 -0400
Subject: [PATCH] GH-439: Add 'transacted' to Rabbit Producer Props

Fixes #439
Resolves #401
---
 .../rabbit/RabbitMessageChannelBinder.java    | 149 ++++++++----------
 .../rabbit/RabbitProducerProperties.java      |  11 ++
 .../binder/rabbit/RabbitBinderTests.java      |   3 +
 3 files changed, 83 insertions(+), 80 deletions(-)

diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java
index cf1e4d0f1..72ccfafe4 100644
--- a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java
+++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java
@@ -86,7 +86,6 @@ import org.springframework.messaging.SubscribableChannel;
 import org.springframework.retry.interceptor.RetryOperationsInterceptor;
 import org.springframework.scheduling.TaskScheduler;
 import org.springframework.util.Assert;
-import org.springframework.util.ClassUtils;
 import org.springframework.util.StringUtils;
 
 /**
@@ -100,7 +99,8 @@ import org.springframework.util.StringUtils;
  * @author David Turanski
  * @author Marius Bogoevici
  */
-public class RabbitMessageChannelBinder extends AbstractBinder<MessageChannel, RabbitConsumerProperties, RabbitProducerProperties> {
+public class RabbitMessageChannelBinder extends AbstractBinder<MessageChannel, RabbitConsumerProperties,
+			RabbitProducerProperties> {
 
 	public static final AnonymousQueue.Base64UrlNamingStrategy ANONYMOUS_GROUP_NAME_GENERATOR
 			= new AnonymousQueue.Base64UrlNamingStrategy("anonymous.");
@@ -125,8 +125,6 @@ public class RabbitMessageChannelBinder extends AbstractBinder<MessageChannel, R
 
 	private final RabbitAdmin rabbitAdmin;
 
-	private final RabbitTemplate rabbitTemplate = new RabbitTemplate();
-
 	private final GenericApplicationContext autoDeclareContext = new GenericApplicationContext();
 
 	private ConnectionFactory connectionFactory;
@@ -156,8 +154,6 @@ public class RabbitMessageChannelBinder extends AbstractBinder<MessageChannel, R
 	public RabbitMessageChannelBinder(ConnectionFactory connectionFactory) {
 		Assert.notNull(connectionFactory, "connectionFactory must not be null");
 		this.connectionFactory = connectionFactory;
-		this.rabbitTemplate.setConnectionFactory(connectionFactory);
-		this.rabbitTemplate.afterPropertiesSet();
 		this.rabbitAdmin = new RabbitAdmin(connectionFactory);
 		this.autoDeclareContext.refresh();
 		this.rabbitAdmin.setApplicationContext(this.autoDeclareContext);
@@ -230,7 +226,8 @@ public class RabbitMessageChannelBinder extends AbstractBinder<MessageChannel, R
 	}
 
 	@Override
-	public Binding<MessageChannel> doBindConsumer(String name, String group, MessageChannel inputChannel, RabbitConsumerProperties properties) {
+	public Binding<MessageChannel> doBindConsumer(String name, String group, MessageChannel inputChannel,
+			RabbitConsumerProperties properties) {
 		boolean anonymousConsumer = !StringUtils.hasText(group);
 		String baseQueueName = anonymousConsumer ? groupedName(name, ANONYMOUS_GROUP_NAME_GENERATOR.generateName())
 				: groupedName(name, group);
@@ -256,7 +253,8 @@ public class RabbitMessageChannelBinder extends AbstractBinder<MessageChannel, R
 				queueName += partitionSuffix;
 			}
 			if (durable) {
-				queue = new Queue(queueName, true, false, false, queueArgs(queueName, properties.getPrefix(), properties.isAutoBindDlq()));
+				queue = new Queue(queueName, true, false, false,
+						queueArgs(queueName, properties.getPrefix(), properties.isAutoBindDlq()));
 			}
 			else {
 				queue = new Queue(queueName, false, false, true);
@@ -290,71 +288,63 @@ public class RabbitMessageChannelBinder extends AbstractBinder<MessageChannel, R
 
 	private Binding<MessageChannel> doRegisterConsumer(final String name, String group, MessageChannel moduleInputChannel, Queue queue,
 													   final RabbitConsumerProperties properties) {
-		DefaultBinding<MessageChannel> consumerBinding = null;
-		// TODO https://github.com/spring-cloud/spring-cloud-stream/issues/401
-		ClassLoader originalClassloader = Thread.currentThread().getContextClassLoader();
-		try {
-			ClassUtils.overrideThreadContextClassLoader(SimpleMessageListenerContainer.class.getClassLoader());
-			SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer(
-					this.connectionFactory);
-			listenerContainer.setAcknowledgeMode(properties.getAcknowledgeMode());
-			listenerContainer.setChannelTransacted(properties.isTransacted());
-			listenerContainer.setDefaultRequeueRejected(properties.isRequeueRejected());
+		DefaultBinding<MessageChannel> consumerBinding;
+		SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer(
+				this.connectionFactory);
+		listenerContainer.setAcknowledgeMode(properties.getAcknowledgeMode());
+		listenerContainer.setChannelTransacted(properties.isTransacted());
+		listenerContainer.setDefaultRequeueRejected(properties.isRequeueRejected());
 
-			int concurrency = properties.getConcurrency();
-			concurrency = concurrency > 0 ? concurrency : 1;
-			listenerContainer.setConcurrentConsumers(concurrency);
-			int maxConcurrency = properties.getMaxConcurrency();
-			if (maxConcurrency > concurrency) {
-				listenerContainer.setMaxConcurrentConsumers(maxConcurrency);
-			}
+		int concurrency = properties.getConcurrency();
+		concurrency = concurrency > 0 ? concurrency : 1;
+		listenerContainer.setConcurrentConsumers(concurrency);
+		int maxConcurrency = properties.getMaxConcurrency();
+		if (maxConcurrency > concurrency) {
+			listenerContainer.setMaxConcurrentConsumers(maxConcurrency);
+		}
 
-			listenerContainer.setPrefetchCount(properties.getPrefetch());
-			listenerContainer.setTxSize(properties.getTxSize());
-			listenerContainer.setTaskExecutor(new SimpleAsyncTaskExecutor(queue.getName() + "-"));
-			listenerContainer.setQueues(queue);
-			int maxAttempts = properties.getMaxAttempts();
-			if (maxAttempts > 1 || properties.isRepublishToDlq()) {
-				RetryOperationsInterceptor retryInterceptor = RetryInterceptorBuilder.stateless()
-						.maxAttempts(maxAttempts)
-						.backOffOptions(properties.getBackOffInitialInterval(),
-								properties.getBackOffMultiplier(),
-								properties.getBackOffMaxInterval())
-						.recoverer(determineRecoverer(name, properties.getPrefix(), properties.isRepublishToDlq()))
-						.build();
-				listenerContainer.setAdviceChain(new Advice[] { retryInterceptor });
+		listenerContainer.setPrefetchCount(properties.getPrefetch());
+		listenerContainer.setTxSize(properties.getTxSize());
+		listenerContainer.setTaskExecutor(new SimpleAsyncTaskExecutor(queue.getName() + "-"));
+		listenerContainer.setQueues(queue);
+		int maxAttempts = properties.getMaxAttempts();
+		if (maxAttempts > 1 || properties.isRepublishToDlq()) {
+			RetryOperationsInterceptor retryInterceptor = RetryInterceptorBuilder.stateless()
+					.maxAttempts(maxAttempts)
+					.backOffOptions(properties.getBackOffInitialInterval(),
+							properties.getBackOffMultiplier(),
+							properties.getBackOffMaxInterval())
+					.recoverer(determineRecoverer(name, properties.getPrefix(), properties.isRepublishToDlq()))
+					.build();
+			listenerContainer.setAdviceChain(new Advice[] { retryInterceptor });
+		}
+		listenerContainer.setAfterReceivePostProcessors(this.decompressingPostProcessor);
+		listenerContainer.setMessagePropertiesConverter(RabbitMessageChannelBinder.inboundMessagePropertiesConverter);
+		listenerContainer.afterPropertiesSet();
+		AmqpInboundChannelAdapter adapter = new AmqpInboundChannelAdapter(listenerContainer);
+		adapter.setBeanFactory(this.getBeanFactory());
+		DirectChannel bridgeToModuleChannel = new DirectChannel();
+		bridgeToModuleChannel.setBeanFactory(this.getBeanFactory());
+		bridgeToModuleChannel.setBeanName(name + ".bridge");
+		adapter.setOutputChannel(bridgeToModuleChannel);
+		adapter.setBeanName("inbound." + name);
+		DefaultAmqpHeaderMapper mapper = new DefaultAmqpHeaderMapper();
+		mapper.setRequestHeaderNames(properties.getRequestHeaderPatterns());
+		mapper.setReplyHeaderNames(properties.getReplyHeaderPatterns());
+		adapter.setHeaderMapper(mapper);
+		adapter.afterPropertiesSet();
+		consumerBinding = new DefaultBinding<MessageChannel>(name, group, moduleInputChannel, adapter) {
+			@Override
+			protected void afterUnbind() {
+				cleanAutoDeclareContext(properties.getPrefix(), name);
 			}
-			listenerContainer.setAfterReceivePostProcessors(this.decompressingPostProcessor);
-			listenerContainer.setMessagePropertiesConverter(RabbitMessageChannelBinder.inboundMessagePropertiesConverter);
-			listenerContainer.afterPropertiesSet();
-			AmqpInboundChannelAdapter adapter = new AmqpInboundChannelAdapter(listenerContainer);
-			adapter.setBeanFactory(this.getBeanFactory());
-			DirectChannel bridgeToModuleChannel = new DirectChannel();
-			bridgeToModuleChannel.setBeanFactory(this.getBeanFactory());
-			bridgeToModuleChannel.setBeanName(name + ".bridge");
-			adapter.setOutputChannel(bridgeToModuleChannel);
-			adapter.setBeanName("inbound." + name);
-			DefaultAmqpHeaderMapper mapper = new DefaultAmqpHeaderMapper();
-			mapper.setRequestHeaderNames(properties.getRequestHeaderPatterns());
-			mapper.setReplyHeaderNames(properties.getReplyHeaderPatterns());
-			adapter.setHeaderMapper(mapper);
-			adapter.afterPropertiesSet();
-			consumerBinding = new DefaultBinding<MessageChannel>(name, group, moduleInputChannel, adapter) {
-				@Override
-				protected void afterUnbind() {
-					cleanAutoDeclareContext(properties.getPrefix(), name);
-				}
-			};
-			ReceivingHandler convertingBridge = new ReceivingHandler();
-			convertingBridge.setOutputChannel(moduleInputChannel);
-			convertingBridge.setBeanName(name + ".convert.bridge");
-			convertingBridge.afterPropertiesSet();
-			bridgeToModuleChannel.subscribe(convertingBridge);
-			adapter.start();
-		}
-		finally {
-			Thread.currentThread().setContextClassLoader(originalClassloader);
-		}
+		};
+		ReceivingHandler convertingBridge = new ReceivingHandler();
+		convertingBridge.setOutputChannel(moduleInputChannel);
+		convertingBridge.setBeanName(name + ".convert.bridge");
+		convertingBridge.afterPropertiesSet();
+		bridgeToModuleChannel.subscribe(convertingBridge);
+		adapter.start();
 		return consumerBinding;
 	}
 
@@ -426,11 +416,12 @@ public class RabbitMessageChannelBinder extends AbstractBinder<MessageChannel, R
 		String exchangeName = applyPrefix(producerProperties.getPrefix(), name);
 		TopicExchange exchange = new TopicExchange(exchangeName);
 		declareExchange(exchangeName, exchange);
-		AmqpOutboundEndpoint endpoint = this.buildOutboundEndpoint(name, producerProperties, determineRabbitTemplate(producerProperties));
+		AmqpOutboundEndpoint endpoint = this.buildOutboundEndpoint(name, producerProperties,
+				buildRabbitTemplate(producerProperties));
 		return doRegisterProducer(name, outputChannel, endpoint, producerProperties);
 	}
 
-	private RabbitTemplate determineRabbitTemplate(RabbitProducerProperties properties) {
+	private RabbitTemplate buildRabbitTemplate(RabbitProducerProperties properties) {
 		RabbitTemplate rabbitTemplate = null;
 		if (properties.isBatchingEnabled()) {
 			BatchingStrategy batchingStrategy = new SimpleBatchingStrategy(
@@ -440,18 +431,16 @@ public class RabbitMessageChannelBinder extends AbstractBinder<MessageChannel, R
 			rabbitTemplate = new BatchingRabbitTemplate(batchingStrategy,
 					getApplicationContext().getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME,
 							TaskScheduler.class));
-			rabbitTemplate.setConnectionFactory(this.connectionFactory);
 		}
+		else {
+			rabbitTemplate = new RabbitTemplate();
+		}
+		rabbitTemplate.setConnectionFactory(this.connectionFactory);
 		if (properties.isCompress()) {
-			if (rabbitTemplate == null) {
-				rabbitTemplate = new RabbitTemplate(this.connectionFactory);
-			}
 			rabbitTemplate.setBeforePublishPostProcessors(this.compressingPostProcessor);
-			rabbitTemplate.afterPropertiesSet();
-		}
-		if (rabbitTemplate == null) {
-			rabbitTemplate = this.rabbitTemplate;
 		}
+		rabbitTemplate.setChannelTransacted(properties.isTransacted());
+		rabbitTemplate.afterPropertiesSet();
 		return rabbitTemplate;
 	}
 
diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitProducerProperties.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitProducerProperties.java
index 6bdbe9a0c..bcd29b4ff 100644
--- a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitProducerProperties.java
+++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitProducerProperties.java
@@ -21,6 +21,7 @@ import org.springframework.cloud.stream.binder.ProducerProperties;
 
 /**
  * @author Marius Bogoevici
+ * @author Gary Russell
  */
 public class RabbitProducerProperties extends ProducerProperties {
 
@@ -40,6 +41,8 @@ public class RabbitProducerProperties extends ProducerProperties {
 
 	private int batchTimeout = 5000;
 
+	private boolean transacted;
+
 	private MessageDeliveryMode deliveryMode = MessageDeliveryMode.PERSISTENT;
 
 	private String[] replyHeaderPatterns = new String[] {"STANDARD_REPLY_HEADERS", "*"};
@@ -124,4 +127,12 @@ public class RabbitProducerProperties extends ProducerProperties {
 		this.batchTimeout = batchTimeout;
 	}
 
+	public boolean isTransacted() {
+		return this.transacted;
+	}
+
+	public void setTransacted(boolean transacted) {
+		this.transacted = transacted;
+	}
+
 }
diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java
index 1b2cb9c0c..3c7d0b581 100644
--- a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java
+++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java
@@ -200,6 +200,7 @@ public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBin
 		assertEquals(2, requestHeaders.size());
 		producerBinding.unbind();
 		assertFalse(endpoint.isRunning());
+		assertFalse(TestUtils.getPropertyValue(endpoint, "handler.delegate.amqpTemplate.transactional", Boolean.class));
 
 		RabbitProducerProperties properties = new RabbitProducerProperties();
 		properties.setPrefix("foo.");
@@ -210,6 +211,7 @@ public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBin
 		properties.setPartitionSelectorExpression(spelExpressionParser.parseExpression("0"));
 		properties.setPartitionSelectorClass(TestPartitionSelectorClass.class);
 		properties.setPartitionCount(1);
+		properties.setTransacted(true);
 
 		producerBinding = binder.bindProducer("props.0", new DirectChannel(), properties);
 		endpoint = extractEndpoint(producerBinding);
@@ -220,6 +222,7 @@ public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBin
 		mode = TestUtils.getPropertyValue(endpoint, "handler.delegate.defaultDeliveryMode",
 				MessageDeliveryMode.class);
 		assertEquals(MessageDeliveryMode.NON_PERSISTENT, mode);
+		assertTrue(TestUtils.getPropertyValue(endpoint, "handler.delegate.amqpTemplate.transactional", Boolean.class));
 		verifyFooRequestProducer(endpoint);
 
 		producerBinding.unbind();