GH-3155: Add support for Java DSL extensions (#3167)

* GH-3155: Add support for Java DSL extensions

Fixes https://github.com/spring-projects/spring-integration/issues/3155

Provide an `IntegrationFlowExtension` for possible custom EI-operators
in the target project use-cases.

* * Move `IntegrationFlowExtension` tests ot its own test class
* Make all the `IntegrationComponentSpec` ctors as `protected` for possible custom extensions
* Make some `BaseIntegrationFlowDefinition` methods and properties as `protected` to get them
access from the `IntegrationFlowExtension` implementations
* Document the feature

* * Fix language and typos in docs

* * Add `protected` to one more `GatewayEndpointSpec` ctor
* Add JavaDocs to `GatewayEndpointSpec` methods

* * Add `protected` to one more `JmsPollableMessageChannelSpec` ctor

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2020 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.
@@ -34,7 +34,7 @@ public class AbstractRouterSpec<S extends AbstractRouterSpec<S, R>, R extends Ab
 
 	private boolean defaultToParentFlow;
 
-	AbstractRouterSpec(R router) {
+	protected AbstractRouterSpec(R router) {
 		super(router);
 	}
 
@@ -100,7 +100,7 @@ public class AbstractRouterSpec<S extends AbstractRouterSpec<S, R>, R extends Ab
 		return _this();
 	}
 
-	boolean isDefaultToParentFlow() {
+	protected boolean isDefaultToParentFlow() {
 		return this.defaultToParentFlow;
 	}
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2020 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.
@@ -39,7 +39,7 @@ public class AggregatorSpec extends CorrelationHandlerSpec<AggregatorSpec, Aggre
 
 	private Function<MessageGroup, Map<String, Object>> headersFunction;
 
-	AggregatorSpec() {
+	protected AggregatorSpec() {
 		super(new AggregatingMessageHandler(new DefaultAggregatingMessageGroupProcessor()));
 	}
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2020 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.
@@ -52,7 +52,7 @@ public class BarrierSpec extends ConsumerEndpointSpec<BarrierSpec, BarrierMessag
 
 	private boolean async;
 
-	BarrierSpec(long timeout) {
+	protected BarrierSpec(long timeout) {
 		super(null);
 		this.timeout = timeout;
 	}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019 the original author or authors.
+ * Copyright 2019-2020 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.
@@ -121,10 +121,10 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
 
 	private static final String MESSAGE_PROCESSOR_SPEC_MUST_NOT_BE_NULL = "'messageProcessorSpec' must not be null";
 
-	private static final SpelExpressionParser PARSER = new SpelExpressionParser();
-
 	private static final Set<MessageProducer> REFERENCED_REPLY_PRODUCERS = new HashSet<>();
 
+	protected static final SpelExpressionParser PARSER = new SpelExpressionParser(); //NOSONAR - final
+
 	protected final Map<Object, String> integrationComponents = new LinkedHashMap<>(); //NOSONAR - final
 
 	private MessageChannel currentMessageChannel;
@@ -380,7 +380,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
 		return wireTap(wireTapChannel, wireTapConfigurer);
 	}
 
-	private MessageChannel obtainInputChannelFromFlow(IntegrationFlow flow) {
+	protected MessageChannel obtainInputChannelFromFlow(IntegrationFlow flow) {
 		Assert.notNull(flow, "'flow' must not be null");
 		MessageChannel messageChannel = flow.getInputChannel();
 		if (messageChannel == null) {
@@ -1222,6 +1222,18 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
 		return enrichHeaders(headers.get(), endpointConfigurer);
 	}
 
+	/**
+	 * Accept a {@link Map} of values to be used for the
+	 * {@link Message} header enrichment.
+	 * {@code values} can apply an {@link Expression}
+	 * to be evaluated against a request {@link Message}.
+	 * @param headers the Map of headers to enrich.
+	 * @return the current {@link IntegrationFlowDefinition}.
+	 */
+	public B enrichHeaders(Map<String, Object> headers) {
+		return enrichHeaders(headers, null);
+	}
+
 	/**
 	 * Accept a {@link Map} of values to be used for the
 	 * {@link Message} header enrichment.
@@ -1908,7 +1920,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
 		return route(new RouterSpec<>(new MethodInvokingRouter(processor)), routerConfigurer);
 	}
 
-	private <R extends AbstractMessageRouter, S extends AbstractRouterSpec<S, R>> B route(S routerSpec,
+	protected <R extends AbstractMessageRouter, S extends AbstractRouterSpec<? super S, R>> B route(S routerSpec,
 			Consumer<S> routerConfigurer) {
 
 		if (routerConfigurer != null) {
@@ -2825,6 +2837,17 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
 				.addComponent(downstream);
 	}
 
+	/**
+	 * Add a {@value IntegrationContextUtils#NULL_CHANNEL_BEAN_NAME} bean into this flow
+	 * definition as a terminal operator.
+	 * @return The {@link IntegrationFlow} instance based on this definition.
+	 * @since 5.1
+	 */
+	public IntegrationFlow nullChannel() {
+		return channel(IntegrationContextUtils.NULL_CHANNEL_BEAN_NAME)
+				.get();
+	}
+
 	/**
 	 * Represent an Integration Flow as a Reactive Streams {@link Publisher} bean.
 	 * @param <T> the expected {@code payload} type
@@ -2858,19 +2881,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
 		return new PublisherIntegrationFlow<>(components, publisher);
 	}
 
-	/**
-	 * Add a {@value IntegrationContextUtils#NULL_CHANNEL_BEAN_NAME} bean into this flow
-	 * definition as a terminal operator.
-	 * @return The {@link IntegrationFlow} instance based on this definition.
-	 * @since 5.1
-	 */
-	public IntegrationFlow nullChannel() {
-		return channel(IntegrationContextUtils.NULL_CHANNEL_BEAN_NAME)
-				.get();
-	}
-
-	@SuppressWarnings(UNCHECKED)
-	private <S extends ConsumerEndpointSpec<S, ? extends MessageHandler>> B register(S endpointSpec,
+	protected <S extends ConsumerEndpointSpec<? super S, ? extends MessageHandler>> B register(S endpointSpec,
 			Consumer<S> endpointConfigurer) {
 
 		if (endpointConfigurer != null) {
@@ -2906,7 +2917,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
 		return addComponent(endpointSpec).currentComponent(factoryBeanTuple2.getT2());
 	}
 
-	private B registerOutputChannelIfCan(MessageChannel outputChannel) {
+	protected B registerOutputChannelIfCan(MessageChannel outputChannel) {
 		if (!(outputChannel instanceof FixedSubscriberChannelPrototype)) {
 			addComponent(outputChannel, null);
 			Object currComponent = getCurrentComponent();
@@ -2947,7 +2958,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
 		return _this();
 	}
 
-	private boolean isOutputChannelRequired() {
+	protected boolean isOutputChannelRequired() {
 		Object currentElement = getCurrentComponent();
 		if (currentElement != null) {
 			if (AopUtils.isAopProxy(currentElement)) {
@@ -3006,7 +3017,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
 		return this.integrationFlow;
 	}
 
-	private void checkReuse(MessageProducer replyHandler) {
+	protected void checkReuse(MessageProducer replyHandler) {
 		Assert.isTrue(!REFERENCED_REPLY_PRODUCERS.contains(replyHandler),
 				"A reply MessageProducer may only be referenced once ("
 						+ replyHandler
@@ -3014,19 +3025,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
 		REFERENCED_REPLY_PRODUCERS.add(replyHandler);
 	}
 
-	/**
-	 * Accept a {@link Map} of values to be used for the
-	 * {@link Message} header enrichment.
-	 * {@code values} can apply an {@link Expression}
-	 * to be evaluated against a request {@link Message}.
-	 * @param headers the Map of headers to enrich.
-	 * @return the current {@link IntegrationFlowDefinition}.
-	 */
-	public B enrichHeaders(Map<String, Object> headers) {
-		return enrichHeaders(headers, null);
-	}
-
-	private static Object extractProxyTarget(Object target) {
+	protected static Object extractProxyTarget(Object target) {
 		if (!(target instanceof Advised)) {
 			return target;
 		}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2020 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.
@@ -42,11 +42,11 @@ import org.springframework.util.Assert;
  *
  * @since 5.0
  */
-public final class DelayerEndpointSpec extends ConsumerEndpointSpec<DelayerEndpointSpec, DelayHandler> {
+public class DelayerEndpointSpec extends ConsumerEndpointSpec<DelayerEndpointSpec, DelayHandler> {
 
 	private final List<Advice> delayedAdvice = new LinkedList<>();
 
-	DelayerEndpointSpec(DelayHandler delayHandler) {
+	protected DelayerEndpointSpec(DelayHandler delayHandler) {
 		super(delayHandler);
 		Assert.notNull(delayHandler, "'delayHandler' must not be null.");
 		this.handler.setDelayedAdviceChain(this.delayedAdvice);

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2020 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.
@@ -37,7 +37,4 @@ public class DirectChannelSpec extends LoadBalancingChannelSpec<DirectChannelSpe
 		return super.doGet();
 	}
 
-	DirectChannelSpec() {
-	}
-
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2020 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.
@@ -47,11 +47,11 @@ import reactor.util.function.Tuple2;
  */
 public class EnricherSpec extends ConsumerEndpointSpec<EnricherSpec, ContentEnricher> {
 
-	private final Map<String, Expression> propertyExpressions = new HashMap<>();
+	protected final Map<String, Expression> propertyExpressions = new HashMap<>(); // NOSONAR - final
 
-	private final Map<String, HeaderValueMessageProcessor<?>> headerExpressions = new HashMap<>();
+	protected final Map<String, HeaderValueMessageProcessor<?>> headerExpressions = new HashMap<>(); // NOSONAR - final
 
-	EnricherSpec() {
+	protected EnricherSpec() {
 		super(new ContentEnricher());
 	}
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2020 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.
@@ -29,7 +29,7 @@ public class ExecutorChannelSpec extends LoadBalancingChannelSpec<ExecutorChanne
 
 	private final Executor executor;
 
-	ExecutorChannelSpec(Executor executor) {
+	protected ExecutorChannelSpec(Executor executor) {
 		this.executor = executor;
 	}
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2020 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.
@@ -26,9 +26,9 @@ import org.springframework.messaging.MessageChannel;
  *
  * @since 5.0
  */
-public final class FilterEndpointSpec extends ConsumerEndpointSpec<FilterEndpointSpec, MessageFilter> {
+public class FilterEndpointSpec extends ConsumerEndpointSpec<FilterEndpointSpec, MessageFilter> {
 
-	FilterEndpointSpec(MessageFilter messageFilter) {
+	protected FilterEndpointSpec(MessageFilter messageFilter) {
 		super(messageFilter);
 	}
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2019 the original author or authors.
+ * Copyright 2017-2020 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.
@@ -26,7 +26,7 @@ import org.springframework.integration.channel.FluxMessageChannel;
  */
 public class FluxMessageChannelSpec extends MessageChannelSpec<FluxMessageChannelSpec, FluxMessageChannel> {
 
-	FluxMessageChannelSpec() {
+	protected FluxMessageChannelSpec() {
 		this.channel = new FluxMessageChannel();
 	}
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2020 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.
@@ -26,43 +26,73 @@ import org.springframework.messaging.MessageChannel;
  *
  * @since 5.0
  */
-public final class GatewayEndpointSpec extends ConsumerEndpointSpec<GatewayEndpointSpec, GatewayMessageHandler> {
+public class GatewayEndpointSpec extends ConsumerEndpointSpec<GatewayEndpointSpec, GatewayMessageHandler> {
 
-	GatewayEndpointSpec(MessageChannel requestChannel) {
+	protected GatewayEndpointSpec(MessageChannel requestChannel) {
 		super(new GatewayMessageHandler());
 		this.handler.setRequestChannel(requestChannel);
 	}
 
-	GatewayEndpointSpec(String requestChannel) {
+	protected GatewayEndpointSpec(String requestChannel) {
 		super(new GatewayMessageHandler());
 		this.handler.setRequestChannelName(requestChannel);
 	}
 
+	/**
+	 * Set a reply channel.
+	 * @param replyChannel the reply channel
+	 * @return the spec
+	 */
 	public GatewayEndpointSpec replyChannel(MessageChannel replyChannel) {
 		this.handler.setReplyChannel(replyChannel);
 		return this;
 	}
 
+	/**
+	 * Set a reply channel.
+	 * @param replyChannel the reply channel
+	 * @return the spec
+	 */
 	public GatewayEndpointSpec replyChannel(String replyChannel) {
 		this.handler.setReplyChannelName(replyChannel);
 		return this;
 	}
 
+	/**
+	 * Set an error channel.
+	 * @param errorChannel the error channel
+	 * @return the spec
+	 */
 	public GatewayEndpointSpec errorChannel(MessageChannel errorChannel) {
 		this.handler.setErrorChannel(errorChannel);
 		return this;
 	}
 
+	/**
+	 * Set an error channel.
+	 * @param errorChannel the error channel
+	 * @return the spec
+	 */
 	public GatewayEndpointSpec errorChannel(String errorChannel) {
 		this.handler.setErrorChannelName(errorChannel);
 		return this;
 	}
 
+	/**
+	 * Set a request timeout.
+	 * @param requestTimeout the request timeout
+	 * @return the spec
+	 */
 	public GatewayEndpointSpec requestTimeout(Long requestTimeout) {
 		this.handler.setRequestTimeout(requestTimeout);
 		return this;
 	}
 
+	/**
+	 * Set a reply timeout.
+	 * @param replyTimeout the reply timeout
+	 * @return the spec
+	 */
 	public GatewayEndpointSpec replyTimeout(Long replyTimeout) {
 		this.handler.setReplyTimeout(replyTimeout);
 		return this;

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019 the original author or authors.
+ * Copyright 2019-2020 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.
@@ -45,19 +45,19 @@ import org.springframework.messaging.MessageChannel;
  */
 public class GatewayProxySpec {
 
-	private static final SpelExpressionParser PARSER = new SpelExpressionParser();
+	protected  static final SpelExpressionParser PARSER = new SpelExpressionParser(); // NOSONAR - final
 
-	private final MessageChannel gatewayRequestChannel = new DirectChannel();
+	protected final MessageChannel gatewayRequestChannel = new DirectChannel(); // NOSONAR - final
 
-	private final GatewayProxyFactoryBean gatewayProxyFactoryBean;
+	protected final GatewayProxyFactoryBean gatewayProxyFactoryBean; // NOSONAR - final
 
-	private final GatewayMethodMetadata gatewayMethodMetadata = new GatewayMethodMetadata();
+	protected final GatewayMethodMetadata gatewayMethodMetadata = new GatewayMethodMetadata(); // NOSONAR - final
 
-	private final Map<String, Expression> headerExpressions = new HashMap<>();
+	protected final Map<String, Expression> headerExpressions = new HashMap<>(); // NOSONAR - final
 
 	private boolean populateGatewayMethodMetadata;
 
-	GatewayProxySpec(Class<?> serviceInterface) {
+	protected GatewayProxySpec(Class<?> serviceInterface) {
 		this.gatewayProxyFactoryBean = new AnnotationGatewayProxyFactoryBean(serviceInterface);
 		this.gatewayProxyFactoryBean.setDefaultRequestChannel(this.gatewayRequestChannel);
 	}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2020 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.
@@ -27,10 +27,10 @@ import org.springframework.messaging.MessageHandler;
  *
  * @since 5.0
  */
-public final class GenericEndpointSpec<H extends MessageHandler>
+public class GenericEndpointSpec<H extends MessageHandler>
 		extends ConsumerEndpointSpec<GenericEndpointSpec<H>, H> {
 
-	GenericEndpointSpec(H messageHandler) {
+	protected GenericEndpointSpec(H messageHandler) {
 		super(messageHandler);
 	}
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2020 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.
@@ -58,11 +58,11 @@ public class HeaderEnricherSpec extends ConsumerEndpointSpec<HeaderEnricherSpec,
 
 	private static final String HEADERS_MUST_NOT_BE_NULL = "'headers' must not be null";
 
-	private final Map<String, HeaderValueMessageProcessor<?>> headerToAdd = new HashMap<>();
+	protected final Map<String, HeaderValueMessageProcessor<?>> headerToAdd = new HashMap<>(); // NOSONAR - final
 
-	private final HeaderEnricher headerEnricher = new HeaderEnricher(this.headerToAdd);
+	protected final HeaderEnricher headerEnricher = new HeaderEnricher(this.headerToAdd); // NOSONAR - final
 
-	HeaderEnricherSpec() {
+	protected HeaderEnricherSpec() {
 		super(null);
 		this.handler = new MessageTransformingHandler(this.headerEnricher);
 	}

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

@@ -0,0 +1,98 @@
+/*
+ * Copyright 2020 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
+ *
+ *      https://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.util.Map;
+
+import org.springframework.beans.factory.BeanNameAware;
+import org.springframework.integration.channel.DirectChannel;
+
+/**
+ * An {@link IntegrationFlowDefinition} extension for custom Java DSL operators
+ * and reusable solutions.
+ * For supporting method flow chain an implementation of this class has to return
+ * an extension class from new methods, e.g.:
+ * <pre class="code">
+ * {@code
+ * 	public class MyIntegrationFlowDefinition
+ * 			extends IntegrationFlowExtension<MyIntegrationFlowDefinition> {
+ *
+ * 		public MyIntegrationFlowDefinition upperCaseAfterSplit() {
+ * 			return split()
+ * 					.transform("payload.toUpperCase()");
+ *      }
+ * }
+ * }
+ * </pre>
+ * This way it will be used in the target configuration as natural DSL definition:
+ * <pre class="code">
+ * {@code
+ *  &#064;Bean
+ *  public IntegrationFlow myFlowDefinition() {
+ * 		return
+ * 				new MyIntegrationFlowDefinition()
+ * 			            .log()
+ * 						.upperCaseAfterSplit()
+ * 						.aggregate()
+ * 						.get();
+ *  }
+ * }
+ * </pre>
+ * This {@link IntegrationFlowExtension} can also be used for overriding
+ * existing operators with extensions to any {@link IntegrationComponentSpec} extensions,
+ * e.g. adding new options for target component configuration.
+ *
+ * @param <B> the {@link IntegrationFlowDefinition} implementation type.
+ *
+ * @author Artem Bilan
+ *
+ * @since 5.3
+ */
+public abstract class IntegrationFlowExtension<B extends IntegrationFlowExtension<B>>
+		extends IntegrationFlowDefinition<B> {
+
+	private final DirectChannel inputChannel = new DirectChannel();
+
+	protected IntegrationFlowExtension() {
+		channel(this.inputChannel);
+	}
+
+	@Override
+	public StandardIntegrationFlow get() {
+		StandardIntegrationFlow targetIntegrationFlow = super.get();
+		return new StandardIntegrationFlowExtension(targetIntegrationFlow.getIntegrationComponents(),
+				this.inputChannel);
+	}
+
+	private static class StandardIntegrationFlowExtension extends StandardIntegrationFlow
+			implements BeanNameAware {
+
+		private final DirectChannel inputChannel;
+
+		StandardIntegrationFlowExtension(Map<Object, String> integrationComponents, DirectChannel inputChannel) {
+			super(integrationComponents);
+			this.inputChannel = inputChannel;
+		}
+
+		@Override
+		public void setBeanName(String name) {
+			this.inputChannel.setBeanName(name + ".input");
+		}
+
+	}
+
+}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2020 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.
@@ -37,7 +37,7 @@ public class PriorityChannelSpec extends MessageChannelSpec<PriorityChannelSpec,
 
 	private MessageGroupQueue messageGroupQueue;
 
-	PriorityChannelSpec() {
+	protected PriorityChannelSpec() {
 	}
 
 	public PriorityChannelSpec capacity(int capacity) {

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2020 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.
@@ -36,10 +36,10 @@ public class PublishSubscribeSpec extends PublishSubscribeChannelSpec<PublishSub
 
 	private int order;
 
-	PublishSubscribeSpec() {
+	protected PublishSubscribeSpec() {
 	}
 
-	PublishSubscribeSpec(@Nullable Executor executor) {
+	protected PublishSubscribeSpec(@Nullable Executor executor) {
 		super(executor);
 	}
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2020 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.
@@ -35,14 +35,14 @@ public class QueueChannelSpec extends MessageChannelSpec<QueueChannelSpec, Queue
 
 	protected Integer capacity; // NOSONAR
 
-	QueueChannelSpec() {
+	protected QueueChannelSpec() {
 	}
 
-	QueueChannelSpec(Queue<Message<?>> queue) {
+	protected QueueChannelSpec(Queue<Message<?>> queue) {
 		this.queue = queue;
 	}
 
-	QueueChannelSpec(Integer capacity) {
+	protected QueueChannelSpec(Integer capacity) {
 		this.capacity = capacity;
 	}
 
@@ -71,7 +71,7 @@ public class QueueChannelSpec extends MessageChannelSpec<QueueChannelSpec, Queue
 
 		private Lock storeLock;
 
-		MessageStoreSpec(ChannelMessageStore messageGroupStore, Object groupId) {
+		protected MessageStoreSpec(ChannelMessageStore messageGroupStore, Object groupId) {
 			this.messageGroupStore = messageGroupStore;
 			this.groupId = groupId;
 		}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2020 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.
@@ -36,7 +36,7 @@ import org.springframework.util.StringUtils;
  */
 public class RecipientListRouterSpec extends AbstractRouterSpec<RecipientListRouterSpec, RecipientListRouter> {
 
-	RecipientListRouterSpec() {
+	protected RecipientListRouterSpec() {
 		super(new RecipientListRouter());
 	}
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2020 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.
@@ -25,7 +25,7 @@ import org.springframework.integration.channel.RendezvousChannel;
  */
 public class RendezvousChannelSpec extends MessageChannelSpec<RendezvousChannelSpec, RendezvousChannel> {
 
-	RendezvousChannelSpec() {
+	protected RendezvousChannelSpec() {
 		this.channel = new RendezvousChannel();
 	}
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2020 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.
@@ -26,7 +26,7 @@ import org.springframework.integration.aggregator.ResequencingMessageHandler;
  */
 public class ResequencerSpec extends CorrelationHandlerSpec<ResequencerSpec, ResequencingMessageHandler> {
 
-	ResequencerSpec() {
+	protected ResequencerSpec() {
 		super(new ResequencingMessageHandler(new ResequencingMessageGroupProcessor()));
 	}
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2020 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.
@@ -52,7 +52,7 @@ public final class RouterSpec<K, R extends AbstractMappingMessageRouter>
 
 	private boolean mappingProviderRegistered;
 
-	RouterSpec(R router) {
+	protected RouterSpec(R router) {
 		super(router);
 		this.mappingProvider = new RouterMappingProvider(this.handler);
 	}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2020 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.
@@ -31,7 +31,7 @@ import org.springframework.messaging.MessageChannel;
  */
 public class ScatterGatherSpec extends ConsumerEndpointSpec<ScatterGatherSpec, ScatterGatherHandler> {
 
-	ScatterGatherSpec(ScatterGatherHandler messageHandler) {
+	protected ScatterGatherSpec(ScatterGatherHandler messageHandler) {
 		super(messageHandler);
 	}
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2020 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.
@@ -25,10 +25,10 @@ import org.springframework.integration.scheduling.PollerMetadata;
  *
  * @since 5.0
  */
-public final class SourcePollingChannelAdapterSpec extends
+public class SourcePollingChannelAdapterSpec extends
 		EndpointSpec<SourcePollingChannelAdapterSpec, SourcePollingChannelAdapterFactoryBean, MessageSource<?>> {
 
-	SourcePollingChannelAdapterSpec(MessageSource<?> messageSource) {
+	protected SourcePollingChannelAdapterSpec(MessageSource<?> messageSource) {
 		super(messageSource);
 		this.endpointFactoryBean.setSource(messageSource);
 	}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2020 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.
@@ -29,10 +29,10 @@ import org.springframework.messaging.MessageChannel;
  *
  * @since 5.0
  */
-public final class SplitterEndpointSpec<S extends AbstractMessageSplitter>
+public class SplitterEndpointSpec<S extends AbstractMessageSplitter>
 		extends ConsumerEndpointSpec<SplitterEndpointSpec<S>, S> {
 
-	SplitterEndpointSpec(S splitter) {
+	protected SplitterEndpointSpec(S splitter) {
 		super(splitter);
 	}
 

spring-integration-core/src/test/java/org/springframework/integration/dsl/extensions/IntegrationFlowExtensionTests.java

@@ -0,0 +1,117 @@
+/*
+ * Copyright 2020 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
+ *
+ *      https://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.extensions;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.Arrays;
+import java.util.function.Consumer;
+import java.util.stream.Collectors;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.config.EnableIntegration;
+import org.springframework.integration.dsl.AggregatorSpec;
+import org.springframework.integration.dsl.IntegrationFlow;
+import org.springframework.integration.dsl.IntegrationFlowExtension;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.SubscribableChannel;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+
+/**
+ * @author Artem Bilan
+ *
+ * @since 5.3
+ */
+@SpringJUnitConfig
+@DirtiesContext
+public class IntegrationFlowExtensionTests {
+
+	@Autowired
+	@Qualifier("customFlowDefinition.input")
+	SubscribableChannel customFlowDefinitionInput;
+
+	@Test
+	public void testCustomFlowDefinition() {
+		QueueChannel replyChannel = new QueueChannel();
+		Message<?> testMessage =
+				MessageBuilder.withPayload(Arrays.asList("one", "two", "three"))
+						.setReplyChannel(replyChannel)
+						.build();
+		this.customFlowDefinitionInput.send(testMessage);
+
+		Message<?> replyMessage = replyChannel.receive(10_000);
+
+		assertThat(replyMessage)
+				.isNotNull()
+				.extracting(Message::getPayload)
+				.isEqualTo("ONE, TWO, THREE");
+	}
+
+	@Configuration
+	@EnableIntegration
+	public static class ContextConfiguration {
+
+		@Bean
+		public IntegrationFlow customFlowDefinition() {
+			return
+					new CustomIntegrationFlowDefinition()
+							.log()
+							.upperCaseAfterSplit()
+							.channel("innerChannel")
+							.customAggregate(customAggregatorSpec ->
+									customAggregatorSpec.expireGroupsUponCompletion(true))
+							.logAndReply();
+		}
+
+	}
+
+	public static class CustomIntegrationFlowDefinition
+			extends IntegrationFlowExtension<CustomIntegrationFlowDefinition> {
+
+		public CustomIntegrationFlowDefinition upperCaseAfterSplit() {
+			return split()
+					.transform("payload.toUpperCase()");
+		}
+
+		public CustomIntegrationFlowDefinition customAggregate(Consumer<CustomAggregatorSpec> aggregator) {
+			return register(new CustomAggregatorSpec(), aggregator);
+		}
+
+	}
+
+	public static class CustomAggregatorSpec extends AggregatorSpec {
+
+		CustomAggregatorSpec() {
+			outputProcessor((group) ->
+					group.getMessages()
+							.stream()
+							.map(Message::getPayload)
+							.map(String.class::cast)
+							.collect(Collectors.joining(", ")));
+		}
+
+	}
+
+}