Support for request sub-flow in content enricher

Fixed issues as per code review and added a test for terminating sub flow

Simplified implementation of enricher sub flow and simplified tests

Polishing

spring-integration-core/src/main/java/org/springframework/integration/dsl/EnricherSpec.java

@@ -21,6 +21,7 @@ import java.util.Map;
 import java.util.function.Function;
 
 import org.springframework.expression.Expression;
+import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.config.ConsumerEndpointFactoryBean;
 import org.springframework.integration.expression.FunctionExpression;
 import org.springframework.integration.expression.ValueExpression;
@@ -40,6 +41,7 @@ import reactor.util.function.Tuple2;
  *
  * @author Artem Bilan
  * @author Tim Ysewyn
+ * @author Ian Bondoc
  *
  * @since 5.0
  */
@@ -136,6 +138,23 @@ public class EnricherSpec extends ConsumerEndpointSpec<EnricherSpec, ContentEnri
 		return _this();
 	}
 
+	/**
+	 * The request sub-flow.
+	 * @param subFlow the subFlowDefinition
+	 * @return the enricher spec
+	 */
+	public EnricherSpec requestSubFlow(IntegrationFlow subFlow) {
+		Assert.notNull(subFlow, "'subFlow' must not be null");
+
+		DirectChannel requestChannel = new DirectChannel();
+		IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(requestChannel);
+		subFlow.configure(flowBuilder);
+
+		this.componentsToRegister.add(flowBuilder.get());
+
+		return requestChannel(requestChannel);
+	}
+
 	/**
 	 * @param shouldClonePayload the shouldClonePayload.
 	 * @return the enricher spec.

spring-integration-core/src/test/java/org/springframework/integration/dsl/transformers/TransformerTests.java

@@ -18,6 +18,7 @@ package org.springframework.integration.dsl.transformers;
 
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.instanceOf;
+import static org.hamcrest.Matchers.is;
 import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
@@ -49,6 +50,7 @@ import org.springframework.integration.config.EnableIntegration;
 import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.integration.dsl.Transformers;
+import org.springframework.integration.dsl.channel.MessageChannels;
 import org.springframework.integration.handler.advice.IdempotentReceiverInterceptor;
 import org.springframework.integration.selector.MetadataStoreSelector;
 import org.springframework.integration.support.MessageBuilder;
@@ -63,6 +65,7 @@ import org.springframework.test.context.junit4.SpringRunner;
 
 /**
  * @author Artem Bilan
+ * @author Ian Bondoc
  *
  * @since 5.0
  */
@@ -136,6 +139,39 @@ public class TransformerTests {
 		assertNull(result.getDate());
 	}
 
+	@Autowired
+	@Qualifier("replyProducingSubFlowEnricher.input")
+	private MessageChannel replyProducingSubFlowEnricherInput;
+
+	@Autowired
+	@Qualifier("terminatingSubFlowEnricher.input")
+	private MessageChannel terminatingSubFlowEnricherInput;
+
+	@Autowired
+	@Qualifier("subFlowTestReplyChannel")
+	private PollableChannel subFlowTestReplyChannel;
+
+	@Test
+	public void testSubFlowContentEnricher() {
+		this.replyProducingSubFlowEnricherInput.send(MessageBuilder.withPayload(new TestPojo("Bar")).build());
+		Message<?> receive = this.subFlowTestReplyChannel.receive(5000);
+		assertNotNull(receive);
+		assertEquals("Foo Bar (Reply Producing)", receive.getHeaders().get("foo"));
+		Object payload = receive.getPayload();
+		assertThat(payload, instanceOf(TestPojo.class));
+		TestPojo result = (TestPojo) payload;
+		assertThat(result.getName(), is("Foo Bar (Reply Producing)"));
+
+		this.terminatingSubFlowEnricherInput.send(MessageBuilder.withPayload(new TestPojo("Bar")).build());
+		receive = this.subFlowTestReplyChannel.receive(5000);
+		assertNotNull(receive);
+		assertEquals("Foo Bar (Terminating)", receive.getHeaders().get("foo"));
+		payload = receive.getPayload();
+		assertThat(payload, instanceOf(TestPojo.class));
+		result = (TestPojo) payload;
+		assertThat(result.getName(), is("Foo Bar (Terminating)"));
+	}
+
 	@Autowired
 	@Qualifier("encodingFlow.input")
 	private MessageChannel encodingFlowInput;
@@ -287,6 +323,43 @@ public class TransformerTests {
 			return idempotentReceiverInterceptor;
 		}
 
+		@Bean
+		public PollableChannel subFlowTestReplyChannel() {
+			return new QueueChannel();
+		}
+
+		@Bean
+		public IntegrationFlow replyProducingSubFlowEnricher(SomeService someService) {
+			return f -> f
+					.enrich(e -> e.<TestPojo>requestPayload(p -> p.getPayload().getName())
+							.requestSubFlow(sf -> sf
+									.<String>handle((p, h) -> someService.someServiceMethod(p)))
+							.<String>headerFunction("foo", Message::getPayload)
+							.propertyFunction("name", Message::getPayload))
+					.channel("subFlowTestReplyChannel");
+		}
+
+		@Bean
+		public MessageChannel enricherReplyChannel() {
+			return MessageChannels.direct().get();
+		}
+
+		@Bean
+		public IntegrationFlow terminatingSubFlowEnricher(SomeService someService) {
+			return f -> f
+					.enrich(e -> e.<TestPojo>requestPayload(p -> p.getPayload().getName())
+							.requestSubFlow(sf -> sf
+									.<String>handle(someService::aTerminatingServiceMethod))
+							.replyChannel("enricherReplyChannel")
+							.<String>headerFunction("foo", Message::getPayload)
+							.propertyFunction("name", Message::getPayload))
+					.channel("subFlowTestReplyChannel");
+		}
+
+		@Bean
+		public SomeService someService() {
+			return new SomeService();
+		}
 
 	}
 
@@ -354,4 +427,21 @@ public class TransformerTests {
 
 	}
 
+	public static class SomeService {
+
+		@Autowired
+		@Qualifier("enricherReplyChannel")
+		public MessageChannel enricherReplyChannel;
+
+		public String someServiceMethod(String value) {
+			return "Foo ".concat(value).concat(" (Reply Producing)");
+		}
+
+		public void aTerminatingServiceMethod(Message<?> message) {
+			String payload = "Foo ".concat(message.getPayload().toString()).concat(" (Terminating)");
+			enricherReplyChannel.send(MessageBuilder.withPayload(payload).copyHeaders(message.getHeaders()).build());
+		}
+
+	}
+
 }