Add claimCheckIn(), claimCheckOut EIP-methods

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<GenericEndpointSpec<MessageTransformingHandler>> 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<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
+		ClaimCheckOutTransformer claimCheckOutTransformer = new ClaimCheckOutTransformer(messageStore);
+		claimCheckOutTransformer.setRemoveMessage(removeMessage);
+		return this.transform(claimCheckOutTransformer, endpointConfigurer);
+	}
+
 	public IntegrationFlowBuilder resequence() {
 		return this.resequence((EndpointConfigurer<GenericEndpointSpec<ResequencingMessageHandler>>) null);
 	}

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<String> 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();