Spring/spring-integration-extensions
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

DSL: O-Gs: requiresReply="true" by default

Browse Source 
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.
This commit is contained in:
Artem Bilan
2014-11-10 18:36:36 +02:00
parent 57caf41fb2
commit b8d986167c
5 changed files with 56 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/AmqpOutboundEndpointSpec.java
View File

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

17
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/FileWritingMessageHandlerSpec.java
View File

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

1
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/RemoteFileOutboundGatewaySpec.java
View File

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

1
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsOutboundGatewaySpec.java
View File

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

52
spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/amqp/AmqpTests.java
View File

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