From b8d986167c1c242461f23968bda1d70173f620e2 Mon Sep 17 00:00:00 2001
From: Artem Bilan <abilan@pivotal.io>
Date: Mon, 10 Nov 2014 18:36:36 +0200
Subject: [PATCH] DSL: O-Gs: `requiresReply="true"` by default

Since many `outbound-gateway`s are designed to always expect replies and XML configuration
specifies `requires-reply="true"` for them by default, the DSL should follow with that.
---
 .../dsl/amqp/AmqpOutboundEndpointSpec.java    |  3 ++
 .../file/FileWritingMessageHandlerSpec.java   | 17 +++---
 .../file/RemoteFileOutboundGatewaySpec.java   |  1 +
 .../dsl/jms/JmsOutboundGatewaySpec.java       |  1 +
 .../integration/dsl/test/amqp/AmqpTests.java  | 52 +++++++++++++++----
 5 files changed, 56 insertions(+), 18 deletions(-)

diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/AmqpOutboundEndpointSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/AmqpOutboundEndpointSpec.java
index f1844b3..ffd2872 100644
--- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/AmqpOutboundEndpointSpec.java
+++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/AmqpOutboundEndpointSpec.java
@@ -40,6 +40,9 @@ public class AmqpOutboundEndpointSpec extends MessageHandlerSpec<AmqpOutboundEnd
 		this.endpoint = new AmqpOutboundEndpoint(amqpTemplate);
 		this.expectReply = expectReply;
 		this.endpoint.setExpectReply(expectReply);
+		if (expectReply) {
+			this.endpoint.setRequiresReply(true);
+		}
 		this.endpoint.setHeaderMapper(this.headerMapper);
 	}
 
diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/FileWritingMessageHandlerSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/FileWritingMessageHandlerSpec.java
index fdb4f22..5213f70 100644
--- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/FileWritingMessageHandlerSpec.java
+++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/FileWritingMessageHandlerSpec.java
@@ -55,28 +55,31 @@ public class FileWritingMessageHandlerSpec
 	}
 
 	FileWritingMessageHandlerSpec expectReply(boolean expectReply) {
-		target.setExpectReply(expectReply);
+		this.target.setExpectReply(expectReply);
+		if (expectReply) {
+			this.target.setRequiresReply(true);
+		}
 		return _this();
 	}
 
 	public FileWritingMessageHandlerSpec autoCreateDirectory(boolean autoCreateDirectory) {
-		target.setAutoCreateDirectory(autoCreateDirectory);
+		this.target.setAutoCreateDirectory(autoCreateDirectory);
 		return _this();
 	}
 
 	public FileWritingMessageHandlerSpec temporaryFileSuffix(String temporaryFileSuffix) {
-		target.setTemporaryFileSuffix(temporaryFileSuffix);
+		this.target.setTemporaryFileSuffix(temporaryFileSuffix);
 		return _this();
 	}
 
 	public FileWritingMessageHandlerSpec fileExistsMode(FileExistsMode fileExistsMode) {
-		target.setFileExistsMode(fileExistsMode);
+		this.target.setFileExistsMode(fileExistsMode);
 		return _this();
 	}
 
 	public FileWritingMessageHandlerSpec fileNameGenerator(FileNameGenerator fileNameGenerator) {
 		this.fileNameGenerator = fileNameGenerator;
-		target.setFileNameGenerator(fileNameGenerator);
+		this.target.setFileNameGenerator(fileNameGenerator);
 		return _this();
 	}
 
@@ -90,12 +93,12 @@ public class FileWritingMessageHandlerSpec
 
 
 	public FileWritingMessageHandlerSpec deleteSourceFiles(boolean deleteSourceFiles) {
-		target.setDeleteSourceFiles(deleteSourceFiles);
+		this.target.setDeleteSourceFiles(deleteSourceFiles);
 		return _this();
 	}
 
 	public FileWritingMessageHandlerSpec charset(String charset) {
-		target.setCharset(charset);
+		this.target.setCharset(charset);
 		return _this();
 	}
 
diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/RemoteFileOutboundGatewaySpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/RemoteFileOutboundGatewaySpec.java
index d03ea8d..0d05859 100644
--- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/RemoteFileOutboundGatewaySpec.java
+++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/RemoteFileOutboundGatewaySpec.java
@@ -40,6 +40,7 @@ public abstract class RemoteFileOutboundGatewaySpec<F, S extends RemoteFileOutbo
 
 	protected RemoteFileOutboundGatewaySpec(AbstractRemoteFileOutboundGateway<F> outboundGateway) {
 		this.target = outboundGateway;
+		this.target.setRequiresReply(true);
 	}
 
 	public S options(String options) {
diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsOutboundGatewaySpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsOutboundGatewaySpec.java
index 699ef5c..1030349 100644
--- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsOutboundGatewaySpec.java
+++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsOutboundGatewaySpec.java
@@ -41,6 +41,7 @@ public class JmsOutboundGatewaySpec extends MessageHandlerSpec<JmsOutboundGatewa
 	JmsOutboundGatewaySpec(ConnectionFactory connectionFactory) {
 		this.target = new JmsOutboundGateway();
 		this.target.setConnectionFactory(connectionFactory);
+		this.target.setRequiresReply(true);
 	}
 
 	public JmsOutboundGatewaySpec extractRequestPayload(boolean extractPayload) {
diff --git a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/amqp/AmqpTests.java b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/amqp/AmqpTests.java
index 678bce2..cc08ff4 100644
--- a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/amqp/AmqpTests.java
+++ b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/amqp/AmqpTests.java
@@ -16,8 +16,11 @@
 
 package org.springframework.integration.dsl.test.amqp;
 
+import static org.hamcrest.Matchers.instanceOf;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.fail;
 
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -26,19 +29,23 @@ import org.springframework.amqp.core.AmqpTemplate;
 import org.springframework.amqp.core.AnonymousQueue;
 import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.integration.dsl.MessageProducers;
 import org.springframework.integration.dsl.amqp.Amqp;
+import org.springframework.integration.handler.ReplyRequiredException;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.PollableChannel;
+import org.springframework.messaging.support.GenericMessage;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -58,20 +65,23 @@ public class AmqpTests {
 	@Qualifier("queue")
 	private Queue amqpQueue;
 
+	@Autowired
+	@Qualifier("amqpOutboundInput")
+	private MessageChannel amqpOutboundInput;
+	@Autowired
+	@Qualifier("amqpReplyChannel.channel")
+	private PollableChannel amqpReplyChannel;
+
+	@Autowired
+	@Qualifier("amqpOutboundGatewayFlow.input")
+	private MessageChannel amqpOutboundGatewayFlowInput;
+
 	@Test
 	public void testAmqpInboundGatewayFlow() throws Exception {
 		Object result = this.amqpTemplate.convertSendAndReceive(this.amqpQueue.getName(), "world");
 		assertEquals("HELLO WORLD", result);
 	}
 
-	@Autowired
-	@Qualifier("amqpOutboundInput")
-	private MessageChannel amqpOutboundInput;
-
-	@Autowired
-	@Qualifier("amqpReplyChannel.channel")
-	private PollableChannel amqpReplyChannel;
-
 	@Test
 	public void testAmqpOutboundFlow() throws Exception {
 		this.amqpOutboundInput.send(MessageBuilder.withPayload("hello through the amqp")
@@ -92,6 +102,17 @@ public class AmqpTests {
 		assertEquals("HELLO THROUGH THE AMQP", receive.getPayload());
 	}
 
+	@Test
+	public void testAmqpOutboundGatewayRequiresReply() {
+		try {
+			this.amqpOutboundGatewayFlowInput.send(new GenericMessage<>("foo"));
+			fail("ReplyRequiredException expected");
+		}
+		catch (Exception e) {
+			assertThat(e, instanceOf(ReplyRequiredException.class));
+		}
+	}
+
 	@Configuration
 	@EnableAutoConfiguration
 	public static class ContextConfiguration {
@@ -99,8 +120,12 @@ public class AmqpTests {
 		@Autowired
 		private ConnectionFactory rabbitConnectionFactory;
 
-		@Autowired
-		private AmqpTemplate amqpTemplate;
+		@Bean
+		public RabbitTemplate rabbitTemplate() {
+			RabbitTemplate rabbitTemplate = new RabbitTemplate(this.rabbitConnectionFactory);
+			rabbitTemplate.setReplyTimeout(1000);
+			return rabbitTemplate;
+		}
 
 		@Bean
 		public Queue queue() {
@@ -118,7 +143,7 @@ public class AmqpTests {
 		@Bean
 		public IntegrationFlow amqpOutboundFlow() {
 			return IntegrationFlows.from(Amqp.channel("amqpOutboundInput", this.rabbitConnectionFactory))
-					.handle(Amqp.outboundAdapter(this.amqpTemplate).routingKeyExpression("headers.routingKey"))
+					.handle(Amqp.outboundAdapter(rabbitTemplate()).routingKeyExpression("headers.routingKey"))
 					.get();
 		}
 
@@ -142,6 +167,11 @@ public class AmqpTests {
 					.get();
 		}
 
+		@Bean
+		public IntegrationFlow amqpOutboundGatewayFlow() {
+			return f -> f.handleWithAdapter(a -> a.amqpGateway(rabbitTemplate()).routingKey("bar"));
+		}
+
 	}
 
 }