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