diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java index 05aab83..57b7af7 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java @@ -57,6 +57,9 @@ import org.springframework.integration.splitter.AbstractMessageSplitter; import org.springframework.integration.splitter.DefaultMessageSplitter; import org.springframework.integration.splitter.ExpressionEvaluatingSplitter; import org.springframework.integration.splitter.MethodInvokingSplitter; +import org.springframework.integration.store.MessageStore; +import org.springframework.integration.transformer.ClaimCheckInTransformer; +import org.springframework.integration.transformer.ClaimCheckOutTransformer; import org.springframework.integration.transformer.ContentEnricher; import org.springframework.integration.transformer.ExpressionEvaluatingTransformer; import org.springframework.integration.transformer.GenericTransformer; @@ -308,6 +311,29 @@ public final class IntegrationFlowBuilder { return this.transform(headerFilter, endpointConfigurer); } + public IntegrationFlowBuilder claimCheckIn(MessageStore messageStore) { + return this.claimCheckIn(messageStore, null); + } + + public IntegrationFlowBuilder claimCheckIn(MessageStore messageStore, EndpointConfigurer> endpointConfigurer) { + return this.transform(new ClaimCheckInTransformer(messageStore), endpointConfigurer); + } + + public IntegrationFlowBuilder claimCheckOut(MessageStore messageStore) { + return this.claimCheckOut(messageStore, false); + } + + public IntegrationFlowBuilder claimCheckOut(MessageStore messageStore, boolean removeMessage) { + return this.claimCheckOut(messageStore, removeMessage, null); + } + + public IntegrationFlowBuilder claimCheckOut(MessageStore messageStore, boolean removeMessage, + EndpointConfigurer> endpointConfigurer) { + ClaimCheckOutTransformer claimCheckOutTransformer = new ClaimCheckOutTransformer(messageStore); + claimCheckOutTransformer.setRemoveMessage(removeMessage); + return this.transform(claimCheckOutTransformer, endpointConfigurer); + } + public IntegrationFlowBuilder resequence() { return this.resequence((EndpointConfigurer>) null); } diff --git a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java index 5480f9d..80a1735 100644 --- a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java +++ b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java @@ -20,6 +20,7 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; @@ -85,8 +86,10 @@ import org.springframework.integration.handler.advice.ExpressionEvaluatingReques import org.springframework.integration.router.MethodInvokingRouter; import org.springframework.integration.scheduling.PollerMetadata; import org.springframework.integration.splitter.DefaultMessageSplitter; +import org.springframework.integration.store.MessageStore; import org.springframework.integration.store.SimpleMessageStore; import org.springframework.integration.support.MessageBuilder; +import org.springframework.integration.support.MutableMessageBuilder; import org.springframework.integration.transformer.PayloadDeserializingTransformer; import org.springframework.integration.transformer.PayloadSerializingTransformer; import org.springframework.integration.xml.transformer.support.XPathExpressionEvaluatingHeaderValueMessageProcessor; @@ -233,6 +236,13 @@ public class IntegrationFlowTests { @Qualifier("defaultOutputChannel") private QueueChannel defaultOutputChannel; + @Autowired + private MessageStore messageStore; + + @Autowired + @Qualifier("claimCheckInput") + private MessageChannel claimCheckInput; + @Test public void testPollingFlow() { assertThat(this.beanFactory.getBean("integerChannel"), Matchers.instanceOf(FixedSubscriberChannel.class)); @@ -629,6 +639,22 @@ public class IntegrationFlowTests { } + @Test + public void testClaimCheck() { + QueueChannel replyChannel = new QueueChannel(); + + Message message = MutableMessageBuilder.withPayload("foo").setReplyChannel(replyChannel).build(); + + this.claimCheckInput.send(message); + + Message receive = replyChannel.receive(2000); + assertNotNull(receive); + assertSame(message, receive); + + assertEquals(1, this.messageStore.getMessageCount()); + assertSame(message, this.messageStore.getMessage(message.getHeaders().getId())); + } + @MessagingGateway(defaultRequestChannel = "controlBus") private static interface ControlBusGateway { @@ -780,6 +806,19 @@ public class IntegrationFlowTests { .get(); } + @Bean + public MessageStore messageStore() { + return new SimpleMessageStore(); + } + + @Bean + public IntegrationFlow claimCheckFlow() { + return IntegrationFlows.from("claimCheckInput") + .claimCheckIn(this.messageStore()) + .claimCheckOut(this.messageStore()) + .get(); + } + @Bean(name = "foo-channel") public QueueChannel fooChannel() { return new QueueChannel();