DSL: Add LambdaMessageProcessor

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java

@@ -36,6 +36,7 @@ import org.springframework.integration.dsl.support.BeanNameMessageProcessor;
 import org.springframework.integration.dsl.support.ComponentConfigurer;
 import org.springframework.integration.dsl.support.EndpointConfigurer;
 import org.springframework.integration.dsl.support.FixedSubscriberChannelPrototype;
+import org.springframework.integration.dsl.support.GenericHandler;
 import org.springframework.integration.dsl.support.GenericRouter;
 import org.springframework.integration.dsl.support.GenericSplitter;
 import org.springframework.integration.dsl.support.MessageChannelReference;
@@ -136,14 +137,25 @@ public final class IntegrationFlowBuilder {
 	}
 
 	public <S, T> IntegrationFlowBuilder transform(GenericTransformer<S, T> genericTransformer) {
-		return this.transform(genericTransformer, null);
+		return this.transform(null, genericTransformer);
+	}
+
+	public <P, T> IntegrationFlowBuilder transform(Class<P> payloadType, GenericTransformer<P, T> genericTransformer) {
+		return this.transform(payloadType, genericTransformer, null);
 	}
 
 	public <S, T> IntegrationFlowBuilder transform(GenericTransformer<S, T> genericTransformer,
 			EndpointConfigurer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
+		return this.transform(null, genericTransformer, endpointConfigurer);
+	}
+
+	public <P, T> IntegrationFlowBuilder transform(Class<P> payloadType, GenericTransformer<P, T> genericTransformer,
+			EndpointConfigurer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
 		Assert.notNull(genericTransformer);
-		Transformer transformer = genericTransformer instanceof Transformer
-				? (Transformer) genericTransformer : new MethodInvokingTransformer(genericTransformer);
+		Transformer transformer = genericTransformer instanceof Transformer ? (Transformer) genericTransformer :
+				(isLambda(genericTransformer)
+						? new MethodInvokingTransformer(new LambdaMessageProcessor(genericTransformer, payloadType))
+						: new MethodInvokingTransformer(genericTransformer));
 		return this.handle(new MessageTransformingHandler(transformer), endpointConfigurer);
 	}
 
@@ -153,14 +165,25 @@ public final class IntegrationFlowBuilder {
 	}
 
 	public <S> IntegrationFlowBuilder filter(GenericSelector<S> genericSelector) {
-		return this.filter(genericSelector, null);
+		return this.filter(null, genericSelector);
 	}
 
-	public <S> IntegrationFlowBuilder filter(GenericSelector<S> genericSelector,
+	public <P> IntegrationFlowBuilder filter(Class<P> payloadType, GenericSelector<P> genericSelector) {
+		return this.filter(payloadType, genericSelector, null);
+	}
+
+	public <P> IntegrationFlowBuilder filter(GenericSelector<P> genericSelector,
+			EndpointConfigurer<FilterEndpointSpec> endpointConfigurer) {
+		return filter(null, genericSelector, endpointConfigurer);
+	}
+
+	public <P> IntegrationFlowBuilder filter(Class<P> payloadType, GenericSelector<P> genericSelector,
 			EndpointConfigurer<FilterEndpointSpec> endpointConfigurer) {
 		Assert.notNull(genericSelector);
-		MessageSelector selector = genericSelector instanceof MessageSelector
-				? (MessageSelector) genericSelector : new MethodInvokingSelector(genericSelector);
+		MessageSelector selector = genericSelector instanceof MessageSelector ? (MessageSelector) genericSelector :
+				(isLambda(genericSelector)
+						? new MethodInvokingSelector(new LambdaMessageProcessor(genericSelector, payloadType))
+						: new MethodInvokingSelector(genericSelector));
 		return this.register(new FilterEndpointSpec(new MessageFilter(selector)), endpointConfigurer);
 	}
 
@@ -178,6 +201,32 @@ public final class IntegrationFlowBuilder {
 				endpointConfigurer);
 	}
 
+	public <P> IntegrationFlowBuilder handle(GenericHandler<P> handler) {
+		return this.handle(null, handler);
+	}
+
+	public <P> IntegrationFlowBuilder handle(GenericHandler<P> handler,
+			EndpointConfigurer<GenericEndpointSpec<ServiceActivatingHandler>> endpointConfigurer) {
+		return this.handle(null, handler, endpointConfigurer);
+	}
+
+
+	public <P> IntegrationFlowBuilder handle(Class<P> payloadType, GenericHandler<P> handler) {
+		return this.handle(payloadType, handler, null);
+	}
+
+	public <P> IntegrationFlowBuilder handle(Class<P> payloadType, GenericHandler<P> handler,
+			EndpointConfigurer<GenericEndpointSpec<ServiceActivatingHandler>> endpointConfigurer) {
+		ServiceActivatingHandler serviceActivatingHandler = null;
+		if (isLambda(handler)) {
+			serviceActivatingHandler = new ServiceActivatingHandler(new LambdaMessageProcessor(handler, payloadType));
+		}
+		else {
+			serviceActivatingHandler = new ServiceActivatingHandler(handler);
+		}
+		return this.handle(serviceActivatingHandler, endpointConfigurer);
+	}
+
 	public <H extends MessageHandler> IntegrationFlowBuilder handle(H messageHandler,
 			EndpointConfigurer<GenericEndpointSpec<H>> endpointConfigurer) {
 		Assert.notNull(messageHandler);
@@ -243,19 +292,10 @@ public final class IntegrationFlowBuilder {
 		return this.addComponent(headerEnricher).transform(headerEnricher, endpointConfigurer);
 	}
 
-	public IntegrationFlowBuilder split() {
-		return this.split((EndpointConfigurer<SplitterEndpointSpec<DefaultMessageSplitter>>) null);
-	}
-
-	public
-	IntegrationFlowBuilder split(EndpointConfigurer<SplitterEndpointSpec<DefaultMessageSplitter>> endpointConfigurer) {
+	public IntegrationFlowBuilder split(EndpointConfigurer<SplitterEndpointSpec<DefaultMessageSplitter>> endpointConfigurer) {
 		return this.split(new DefaultMessageSplitter(), endpointConfigurer);
 	}
 
-	public IntegrationFlowBuilder split(String expression) {
-		return this.split(expression, (EndpointConfigurer<SplitterEndpointSpec<ExpressionEvaluatingSplitter>>) null);
-	}
-
 	public IntegrationFlowBuilder split(String expression,
 			EndpointConfigurer<SplitterEndpointSpec<ExpressionEvaluatingSplitter>> endpointConfigurer) {
 		return this.split(new ExpressionEvaluatingSplitter(PARSER.parseExpression(expression)), endpointConfigurer);
@@ -271,17 +311,21 @@ public final class IntegrationFlowBuilder {
 				endpointConfigurer);
 	}
 
-	public IntegrationFlowBuilder split(AbstractMessageSplitter splitter) {
-		return this.split(splitter, null);
-	}
-
-	public <T> IntegrationFlowBuilder split(GenericSplitter<T> splitter) {
-		return this.split(splitter, null);
+	public <P> IntegrationFlowBuilder split(Class<P> payloadType, GenericSplitter<P> splitter) {
+		return this.split(payloadType, splitter, null);
 	}
 
 	public <T> IntegrationFlowBuilder split(GenericSplitter<T> splitter,
 			EndpointConfigurer<SplitterEndpointSpec<MethodInvokingSplitter>> endpointConfigurer) {
-		return this.split(new MethodInvokingSplitter(splitter, "split"), endpointConfigurer);
+		return split(null, splitter, endpointConfigurer);
+	}
+
+	public <P> IntegrationFlowBuilder split(Class<P> payloadType, GenericSplitter<P> splitter,
+			EndpointConfigurer<SplitterEndpointSpec<MethodInvokingSplitter>> endpointConfigurer) {
+		MethodInvokingSplitter split = isLambda(splitter)
+				? new MethodInvokingSplitter(new LambdaMessageProcessor(splitter, payloadType))
+				: new MethodInvokingSplitter(splitter, "split");
+		return this.split(split, endpointConfigurer);
 	}
 
 	public <S extends AbstractMessageSplitter> IntegrationFlowBuilder split(S splitter,
@@ -348,8 +392,7 @@ public final class IntegrationFlowBuilder {
 		return this.resequence((EndpointConfigurer<GenericEndpointSpec<ResequencingMessageHandler>>) null);
 	}
 
-	public
-	IntegrationFlowBuilder resequence(EndpointConfigurer<GenericEndpointSpec<ResequencingMessageHandler>> endpointConfigurer) {
+	public IntegrationFlowBuilder resequence(EndpointConfigurer<GenericEndpointSpec<ResequencingMessageHandler>> endpointConfigurer) {
 		return this.resequence(new ResequencingMessageHandler(new ResequencingMessageGroupProcessor()),
 				endpointConfigurer);
 	}
@@ -375,20 +418,11 @@ public final class IntegrationFlowBuilder {
 		return this.handle(resequencer, endpointConfigurer);
 	}
 
-	public IntegrationFlowBuilder aggregate() {
-		return this.aggregate((EndpointConfigurer<GenericEndpointSpec<AggregatingMessageHandler>>) null);
-	}
-
-	public
-	IntegrationFlowBuilder aggregate(EndpointConfigurer<GenericEndpointSpec<AggregatingMessageHandler>> endpointConfigurer) {
+	public IntegrationFlowBuilder aggregate(EndpointConfigurer<GenericEndpointSpec<AggregatingMessageHandler>> endpointConfigurer) {
 		return this.aggregate(new AggregatingMessageHandler(new DefaultAggregatingMessageGroupProcessor()),
 				endpointConfigurer);
 	}
 
-	public IntegrationFlowBuilder aggregate(ComponentConfigurer<AggregatorSpec> aggregatorConfigurer) {
-		return this.aggregate(aggregatorConfigurer, null);
-	}
-
 	public IntegrationFlowBuilder aggregate(ComponentConfigurer<AggregatorSpec> aggregatorConfigurer,
 			EndpointConfigurer<GenericEndpointSpec<AggregatingMessageHandler>> endpointConfigurer) {
 		Assert.notNull(aggregatorConfigurer);
@@ -397,10 +431,6 @@ public final class IntegrationFlowBuilder {
 		return this.aggregate(spec.get(), endpointConfigurer);
 	}
 
-	public IntegrationFlowBuilder aggregate(AggregatingMessageHandler aggregator) {
-		return this.aggregate(aggregator, null);
-	}
-
 	public IntegrationFlowBuilder aggregate(AggregatingMessageHandler aggregator,
 			EndpointConfigurer<GenericEndpointSpec<AggregatingMessageHandler>> endpointConfigurer) {
 		return this.handle(aggregator, endpointConfigurer);
@@ -441,23 +471,36 @@ public final class IntegrationFlowBuilder {
 	}
 
 	public <S, T> IntegrationFlowBuilder route(GenericRouter<S, T> router) {
-		return this.route(router, (ComponentConfigurer<RouterSpec<MethodInvokingRouter>>) null);
+		return this.route(null, router);
 	}
 
 	public <S, T> IntegrationFlowBuilder route(GenericRouter<S, T> router,
 			ComponentConfigurer<RouterSpec<MethodInvokingRouter>> routerConfigurer) {
-		return this.route(router, routerConfigurer, null);
+		return this.route(null, router, routerConfigurer);
+	}
+
+	public <P, T> IntegrationFlowBuilder route(Class<P> payloadType, GenericRouter<P, T> router) {
+		return this.route(payloadType, router, null, null);
+	}
+
+	public <P, T> IntegrationFlowBuilder route(Class<P> payloadType, GenericRouter<P, T> router,
+			ComponentConfigurer<RouterSpec<MethodInvokingRouter>> routerConfigurer) {
+		return this.route(payloadType, router, routerConfigurer, null);
 	}
 
 	public <S, T> IntegrationFlowBuilder route(GenericRouter<S, T> router,
 			ComponentConfigurer<RouterSpec<MethodInvokingRouter>> routerConfigurer,
 			EndpointConfigurer<GenericEndpointSpec<MethodInvokingRouter>> endpointConfigurer) {
-		return this.route(new MethodInvokingRouter(router), routerConfigurer, endpointConfigurer);
+		return route(null, router, routerConfigurer, endpointConfigurer);
 	}
 
-	public <R extends AbstractMappingMessageRouter> IntegrationFlowBuilder route(R router,
-			ComponentConfigurer<RouterSpec<R>> routerConfigurer) {
-		return this.route(router, routerConfigurer, null);
+	public <P, T> IntegrationFlowBuilder route(Class<P> payloadType, GenericRouter<P, T> router,
+			ComponentConfigurer<RouterSpec<MethodInvokingRouter>> routerConfigurer,
+			EndpointConfigurer<GenericEndpointSpec<MethodInvokingRouter>> endpointConfigurer) {
+		MethodInvokingRouter methodInvokingRouter = isLambda(router)
+				? new MethodInvokingRouter(new LambdaMessageProcessor(router, payloadType))
+				: new MethodInvokingRouter(router);
+		return this.route(methodInvokingRouter, routerConfigurer, endpointConfigurer);
 	}
 
 	public <R extends AbstractMappingMessageRouter> IntegrationFlowBuilder route(R router,
@@ -613,4 +656,9 @@ public final class IntegrationFlowBuilder {
 		return this.flow;
 	}
 
+	private static boolean isLambda(Object o) {
+		Class<?> aClass = o.getClass();
+		return aClass.isSynthetic() && !aClass.isAnonymousClass() && aClass.getDeclaredMethods().length == 1;
+	}
+
 }

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/LambdaMessageProcessor.java

@@ -0,0 +1,111 @@
+/*
+ * Copyright 2014 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.dsl;
+
+import java.lang.reflect.Method;
+import java.util.Map;
+
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.BeanFactoryAware;
+import org.springframework.core.convert.ConversionService;
+import org.springframework.core.convert.TypeDescriptor;
+import org.springframework.core.convert.support.DefaultConversionService;
+import org.springframework.integration.handler.MessageProcessor;
+import org.springframework.integration.support.utils.IntegrationUtils;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHandlingException;
+import org.springframework.util.Assert;
+
+/**
+ * @author Artem Bilan
+ */
+class LambdaMessageProcessor implements MessageProcessor<Object>, BeanFactoryAware {
+
+	private final Object target;
+
+	private final Method method;
+
+	private final TypeDescriptor payloadType;
+
+	private final Class<?>[] parameterTypes;
+
+
+	private ConversionService conversionService;
+
+	public LambdaMessageProcessor(Object target, Class<?> payloadType) {
+		Assert.notNull(target);
+		this.target = target;
+		Method[] declaredMethods = target.getClass().getDeclaredMethods();
+		Assert.isTrue(declaredMethods.length == 1, "LambdaMessageProcessor is applicable for inline or lambda" +
+				"classes with single method - functional interfaces implementations.");
+		this.method = declaredMethods[0];
+		this.method.setAccessible(true);
+		this.parameterTypes = this.method.getParameterTypes();
+		this.payloadType = TypeDescriptor.valueOf(payloadType);
+	}
+
+	@Override
+	public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
+		ConversionService conversionService = IntegrationUtils.getConversionService(beanFactory);
+		if (conversionService == null) {
+			conversionService = new DefaultConversionService();
+		}
+		this.conversionService = conversionService;
+	}
+
+	@Override
+	public Object processMessage(Message<?> message) {
+		Object[] args = new Object[this.parameterTypes.length];
+		for (int i = 0; i < this.parameterTypes.length; i++) {
+			Class<?> parameterType = this.parameterTypes[i];
+			if (Message.class.isAssignableFrom(parameterType)) {
+				args[i] = message;
+			}
+			if (Map.class.isAssignableFrom(parameterType)) {
+				if (message.getPayload() instanceof Map) {
+					args[i] = message.getPayload();
+				}
+				else {
+					args[i] = message.getHeaders();
+				}
+			}
+			else {
+				if (this.payloadType != null) {
+					if (Message.class.isAssignableFrom(this.payloadType.getType())) {
+						args[i] = message;
+					}
+					else {
+						args[i] = this.conversionService.convert(message.getPayload(),
+								TypeDescriptor.forObject(message.getPayload()), this.payloadType);
+					}
+
+				}
+				else {
+					args[i] = message.getPayload();
+				}
+			}
+		}
+
+		try {
+			return this.method.invoke(this.target, args);
+		}
+		catch (Exception e) {
+			throw new MessageHandlingException(message, e);
+		}
+	}
+}

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/GenericHandler.java

@@ -0,0 +1,28 @@
+/*
+ * Copyright 2014 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.dsl.support;
+
+import java.util.Map;
+
+/**
+ * @author Artem Bilan
+ */
+public interface GenericHandler<P> {
+
+	Object handle(P payload, Map<String, Object> headers);
+
+}

spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java

@@ -82,6 +82,7 @@ import org.springframework.integration.channel.FixedSubscriberChannel;
 import org.springframework.integration.channel.PublishSubscribeChannel;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.config.EnableIntegration;
+import org.springframework.integration.config.EnableMessageHistory;
 import org.springframework.integration.config.GlobalChannelInterceptor;
 import org.springframework.integration.core.MessageSource;
 import org.springframework.integration.dsl.IntegrationFlow;
@@ -91,6 +92,7 @@ import org.springframework.integration.dsl.SplitterEndpointSpec;
 import org.springframework.integration.dsl.amqp.Amqp;
 import org.springframework.integration.dsl.channel.DirectChannelSpec;
 import org.springframework.integration.dsl.channel.MessageChannels;
+import org.springframework.integration.dsl.support.GenericHandler;
 import org.springframework.integration.dsl.support.GenericSplitter;
 import org.springframework.integration.dsl.support.Pollers;
 import org.springframework.integration.endpoint.MethodInvokingMessageSource;
@@ -372,9 +374,9 @@ public class IntegrationFlowTests {
 	@Test
 	public void testHandle() {
 		assertNull(this.eventHolder.get());
-		this.flow3Input.send(new GenericMessage<>("foo"));
+		this.flow3Input.send(new GenericMessage<>("2"));
 		assertNotNull(this.eventHolder.get());
-		assertEquals("foo", this.eventHolder.get());
+		assertEquals(4, this.eventHolder.get());
 	}
 
 	@Test
@@ -548,7 +550,7 @@ public class IntegrationFlowTests {
 			assertFalse(receive.getHeaders().containsKey("foo"));
 			assertTrue(receive.getHeaders().containsKey("FOO"));
 			assertEquals("BAR", receive.getHeaders().get("FOO"));
-			assertEquals(new Integer(i + 1), receive.getPayload());
+			assertEquals(i + 1, receive.getPayload());
 		}
 	}
 
@@ -603,29 +605,29 @@ public class IntegrationFlowTests {
 	@Test
 	public void testRouter() {
 
-		int[] payloads = new int[]{1, 2, 3, 4, 5, 6};
+		int[] payloads = new int[] {1, 2, 3, 4, 5, 6};
 
 		for (int payload : payloads) {
-			this.routerInput.send(new GenericMessage<Integer>(payload));
+			this.routerInput.send(new GenericMessage<>(payload));
 		}
 
 		for (int i = 0; i < 3; i++) {
 			Message<?> receive = this.oddChannel.receive(2000);
 			assertNotNull(receive);
-			assertEquals(new Integer(i * 2 + 1), receive.getPayload());
+			assertEquals(i * 2 + 1, receive.getPayload());
 
 			receive = this.evenChannel.receive(2000);
 			assertNotNull(receive);
-			assertEquals(new Integer(i * 2 + 2), receive.getPayload());
+			assertEquals(i * 2 + 2, receive.getPayload());
 		}
 
 	}
 
 	@Test
 	public void testMethodInvokingRouter() {
-		Message<String> fooMessage = new GenericMessage<String>("foo");
-		Message<String> barMessage = new GenericMessage<String>("bar");
-		Message<String> badMessage = new GenericMessage<String>("bad");
+		Message<String> fooMessage = new GenericMessage<>("foo");
+		Message<String> barMessage = new GenericMessage<>("bar");
+		Message<String> badMessage = new GenericMessage<>("bad");
 
 		this.routerMethodInput.send(fooMessage);
 
@@ -684,9 +686,9 @@ public class IntegrationFlowTests {
 
 	@Test
 	public void testMethodInvokingRouter3() {
-		Message<String> fooMessage = new GenericMessage<String>("foo");
-		Message<String> barMessage = new GenericMessage<String>("bar");
-		Message<String> badMessage = new GenericMessage<String>("bad");
+		Message<String> fooMessage = new GenericMessage<>("foo");
+		Message<String> barMessage = new GenericMessage<>("bar");
+		Message<String> badMessage = new GenericMessage<>("bad");
 
 		this.routerMethod3Input.send(fooMessage);
 
@@ -715,9 +717,9 @@ public class IntegrationFlowTests {
 	@Test
 	public void testMultiRouter() {
 
-		Message<String> fooMessage = new GenericMessage<String>("foo");
-		Message<String> barMessage = new GenericMessage<String>("bar");
-		Message<String> badMessage = new GenericMessage<String>("bad");
+		Message<String> fooMessage = new GenericMessage<>("foo");
+		Message<String> barMessage = new GenericMessage<>("bar");
+		Message<String> badMessage = new GenericMessage<>("bad");
 
 		this.routerMultiInput.send(fooMessage);
 		Message<?> result1a = this.fooChannel.receive(2000);
@@ -750,7 +752,7 @@ public class IntegrationFlowTests {
 
 		Message<String> fooMessage = MessageBuilder.withPayload("fooPayload").setHeader("recipient", true).build();
 		Message<String> barMessage = MessageBuilder.withPayload("barPayload").setHeader("recipient", true).build();
-		Message<String> badMessage = new GenericMessage<String>("badPayload");
+		Message<String> badMessage = new GenericMessage<>("badPayload");
 
 		this.recipientListInput.send(fooMessage);
 		Message<?> result1a = this.fooChannel.receive(2000);
@@ -986,7 +988,7 @@ public class IntegrationFlowTests {
 
 		@Bean
 		public MongoDbFactory mongoDbFactory() throws Exception {
-			return new SimpleMongoDbFactory(new MongoClient("localhost",mongoPort), "local");
+			return new SimpleMongoDbFactory(new MongoClient("localhost", mongoPort), "local");
 		}
 
 		@Bean
@@ -999,7 +1001,7 @@ public class IntegrationFlowTests {
 		@Bean
 		public IntegrationFlow priorityFlow(PriorityCapableChannelMessageStore mongoDbChannelMessageStore) {
 			return IntegrationFlows.from(MessageChannels.priority("priorityChannel",
-					mongoDbChannelMessageStore,	"priorityGroup").interceptor())
+					mongoDbChannelMessageStore, "priorityGroup").interceptor())
 					.bridge(s -> s.poller(Pollers.fixedDelay(1000, 2000)))
 					.channel(MessageChannels.queue("priorityReplyChannel"))
 					.get();
@@ -1047,6 +1049,7 @@ public class IntegrationFlowTests {
 		@Bean
 		public IntegrationFlow flow3() {
 			return IntegrationFlows.from("flow3Input")
+					.handle(Integer.class, (p, h) -> p * 2)
 					.handle(new ApplicationEventPublishingMessageHandler())
 					.get();
 		}
@@ -1206,16 +1209,9 @@ public class IntegrationFlowTests {
 					.enrichHeaders(s -> s.header("FOO", "BAR"))
 					.split("testSplitterData", "buildList", c -> c.applySequence(false))
 					.channel(MessageChannels.executor(this.taskExecutor()))
-					.split(new GenericSplitter<Message<List<?>>>() {
-
-						@Override
-						public Collection<?> split(Message<List<?>> target) {
-							return target.getPayload();
-						}
-					}, c -> c.applySequence(false))
+					.split(Message.class, target -> (List<?>) target.getPayload(), c -> c.applySequence(false))
 					.channel(MessageChannels.executor(this.taskExecutor()))
-					.split((SplitterEndpointSpec<DefaultMessageSplitter> s) ->
-							s.applySequence(false).get().getT2().setDelimiters(","))
+					.split(s -> s.applySequence(false).get().getT2().setDelimiters(","))
 					.channel(MessageChannels.executor(this.taskExecutor()))
 					.<String, Integer>transform(Integer::parseInt)
 					.enrichHeaders(s -> s.headerExpression(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, "payload"))
@@ -1227,10 +1223,10 @@ public class IntegrationFlowTests {
 		@Bean
 		public IntegrationFlow splitAggregateFlow() {
 			return IntegrationFlows.fromFixedMessageChannel("splitAggregateInput")
-					.split()
+					.split(null)
 					.channel(MessageChannels.executor(this.taskExecutor()))
 					.resequence()
-					.aggregate()
+					.aggregate(null)
 					.get();
 		}
 
@@ -1261,8 +1257,7 @@ public class IntegrationFlowTests {
 					.<Integer, Boolean>route(p -> p % 2 == 0,
 							m -> m.suffix("Channel")
 									.channelMapping("true", "even")
-									.channelMapping("false", "odd")
-					)
+									.channelMapping("false", "odd"))
 					.get();
 		}
 
@@ -1295,9 +1290,8 @@ public class IntegrationFlowTests {
 		@Bean
 		public IntegrationFlow routeMultiMethodInvocationFlow() {
 			return IntegrationFlows.from("routerMultiInput")
-					.<String, String[]>route(p -> p.equals("foo") || p.equals("bar") ? new String[] {"foo", "bar"} : null,
-							s -> s.suffix("-channel")
-					)
+					.route(String.class, p -> p.equals("foo") || p.equals("bar") ? new String[] {"foo", "bar"} : null,
+							s -> s.suffix("-channel"))
 					.get();
 		}
 
@@ -1324,7 +1318,7 @@ public class IntegrationFlowTests {
 		public IntegrationFlow amqpFlow() {
 			return IntegrationFlows.from(Amqp.inboundGateway(this.rabbitConnectionFactory, queue()).get())
 					.transform("hello "::concat)
-					.transform((String p) -> p.toUpperCase())
+					.transform(String.class, String::toUpperCase)
 					.get();
 		}