Introduce HeaderFilterSpec to streamline DSL API

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

@@ -1626,11 +1626,23 @@ public B headerFilter(String... headersToRemove) {
 	 * @param patternMatch the {@code boolean} flag to indicate if {@code headersToRemove}
 	 * should be interpreted as patterns or direct header names.
 	 * @return this {@link BaseIntegrationFlowDefinition}.
+	 * @deprecated since 6.2 in favor of {@link #headerFilter(Consumer)}
 	 */
+	@Deprecated(since = "6.2", forRemoval = true)
 	public B headerFilter(String headersToRemove, boolean patternMatch) {
-		HeaderFilter headerFilter = new HeaderFilter(StringUtils.delimitedListToStringArray(headersToRemove, ",", " "));
-		headerFilter.setPatternMatch(patternMatch);
-		return headerFilter(headerFilter, null);
+		return headerFilter((headerFilterSpec) -> headerFilterSpec
+				.headersToRemove(StringUtils.delimitedListToStringArray(headersToRemove, ",", " "))
+				.patternMatch(patternMatch));
+	}
+
+	/**
+	 * Provide the {@link HeaderFilter} options via fluent API of the {@link HeaderFilterSpec}.
+	 * @param headerFilter the {@link Consumer} to provide header filter and its endpoint options.
+	 * @return this {@link BaseIntegrationFlowDefinition}.
+	 * @since 6.2
+	 */
+	public B headerFilter(Consumer<HeaderFilterSpec> headerFilter) {
+		return register(new HeaderFilterSpec(), headerFilter);
 	}
 
 	/**

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

@@ -0,0 +1,68 @@
+/*
+ * Copyright 2023 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 org.springframework.integration.transformer.HeaderFilter;
+import org.springframework.integration.transformer.MessageTransformingHandler;
+import org.springframework.util.Assert;
+
+/**
+ * A {@link ConsumerEndpointSpec} implementation for the {@link HeaderFilter}.
+ *
+ * @author Artem Bilan
+ *
+ * @since 6.2
+ */
+public class HeaderFilterSpec extends ConsumerEndpointSpec<HeaderFilterSpec, MessageTransformingHandler> {
+
+	private final HeaderFilter headerFilter;
+
+	private final boolean headerFilterExplicitlySet;
+
+	protected HeaderFilterSpec() {
+		this(new HeaderFilter(), false);
+	}
+
+	protected HeaderFilterSpec(HeaderFilter headerFilter) {
+		this(headerFilter, true);
+	}
+
+	private HeaderFilterSpec(HeaderFilter headerFilter, boolean headerFilterExplicitlySet) {
+		super(new MessageTransformingHandler(headerFilter));
+		this.headerFilter = headerFilter;
+		this.componentsToRegister.put(this.headerFilter, null);
+		this.headerFilterExplicitlySet = headerFilterExplicitlySet;
+	}
+
+	public HeaderFilterSpec headersToRemove(String... headersToRemove) {
+		assertHeaderFilterNotExplicitlySet();
+		this.headerFilter.setHeadersToRemove(headersToRemove);
+		return this;
+	}
+
+	public HeaderFilterSpec patternMatch(boolean patternMatch) {
+		assertHeaderFilterNotExplicitlySet();
+		this.headerFilter.setPatternMatch(patternMatch);
+		return this;
+	}
+
+	private void assertHeaderFilterNotExplicitlySet() {
+		Assert.isTrue(!this.headerFilterExplicitlySet,
+				() -> "Cannot override already set header filter: " + this.headerFilter);
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/transformer/HeaderFilter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2023 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.
@@ -40,16 +40,32 @@
  */
 public class HeaderFilter extends IntegrationObjectSupport implements Transformer, IntegrationPattern {
 
-	private final String[] headersToRemove;
+	private String[] headersToRemove;
 
 	private volatile boolean patternMatch = true;
 
 
+	/**
+	 * Create an instance of the class.
+	 * The {@link #setHeadersToRemove} must be called afterwards.
+	 * @since 6.2
+	 */
+	public HeaderFilter() {
+	}
+
 	public HeaderFilter(String... headersToRemove) {
-		Assert.notEmpty(headersToRemove, "At least one header name to remove is required.");
-		this.headersToRemove = Arrays.copyOf(headersToRemove, headersToRemove.length);
+		setHeadersToRemove(headersToRemove);
 	}
 
+	/**
+	 * Set a list of header names (or patterns) to remove from a request message.
+	 * @param headersToRemove the list of header names (or patterns) to remove from a request message.
+	 * @since 6.2
+	 */
+	public final void setHeadersToRemove(String... headersToRemove) {
+		assertHeadersToRemoveNotEmpty(headersToRemove);
+		this.headersToRemove = Arrays.copyOf(headersToRemove, headersToRemove.length);
+	}
 	public void setPatternMatch(boolean patternMatch) {
 		this.patternMatch = patternMatch;
 	}
@@ -66,6 +82,7 @@ public IntegrationPatternType getIntegrationPatternType() {
 
 	@Override
 	protected void onInit() {
+		assertHeadersToRemoveNotEmpty(this.headersToRemove);
 		super.onInit();
 		if (getMessageBuilderFactory() instanceof DefaultMessageBuilderFactory) {
 			for (String header : this.headersToRemove) {
@@ -94,4 +111,8 @@ public Message<?> transform(Message<?> message) {
 		return builder.build();
 	}
 
+	private static void assertHeadersToRemoveNotEmpty(String[] headersToRemove) {
+		Assert.notEmpty(headersToRemove, "At least one header name to remove is required.");
+	}
+
 }

spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinIntegrationFlowDefinition.kt

@@ -22,40 +22,27 @@ import org.springframework.integration.aggregator.AggregatingMessageHandler
 import org.springframework.integration.channel.BroadcastCapableChannel
 import org.springframework.integration.channel.FluxMessageChannel
 import org.springframework.integration.channel.interceptor.WireTap
+import org.springframework.integration.core.GenericHandler
 import org.springframework.integration.core.MessageSelector
 import org.springframework.integration.dsl.support.MessageChannelReference
 import org.springframework.integration.filter.MessageFilter
 import org.springframework.integration.filter.MethodInvokingSelector
-import org.springframework.integration.handler.BridgeHandler
-import org.springframework.integration.handler.DelayHandler
-import org.springframework.integration.core.GenericHandler
-import org.springframework.integration.handler.LoggingHandler
-import org.springframework.integration.handler.MessageProcessor
-import org.springframework.integration.handler.MessageTriggerAction
-import org.springframework.integration.handler.ServiceActivatingHandler
-import org.springframework.integration.router.AbstractMessageRouter
-import org.springframework.integration.router.ErrorMessageExceptionTypeRouter
-import org.springframework.integration.router.ExpressionEvaluatingRouter
-import org.springframework.integration.router.MethodInvokingRouter
-import org.springframework.integration.router.RecipientListRouter
+import org.springframework.integration.handler.*
+import org.springframework.integration.router.*
 import org.springframework.integration.scattergather.ScatterGatherHandler
 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.support.MapBuilder
-import org.springframework.integration.transformer.ClaimCheckInTransformer
-import org.springframework.integration.transformer.ClaimCheckOutTransformer
-import org.springframework.integration.transformer.HeaderFilter
-import org.springframework.integration.transformer.MessageTransformingHandler
-import org.springframework.integration.transformer.MethodInvokingTransformer
-import org.springframework.integration.transformer.Transformer
+import org.springframework.integration.transformer.*
 import org.springframework.messaging.Message
 import org.springframework.messaging.MessageChannel
 import org.springframework.messaging.MessageHandler
 import org.springframework.messaging.MessageHeaders
 import org.springframework.messaging.support.ChannelInterceptor
+import org.springframework.util.StringUtils
 import reactor.core.publisher.Flux
 import java.util.function.Consumer
 
@@ -711,8 +698,23 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ
 	/**
 	 * Provide the [HeaderFilter] to the current [IntegrationFlow].
 	 */
+	@Deprecated("since 6.2",
+			ReplaceWith("""headerFilter { 
+										patternMatch() 
+										headersToRemove() 
+									}"""))
 	fun headerFilter(headersToRemove: String, patternMatch: Boolean = true) {
-		this.delegate.headerFilter(headersToRemove, patternMatch)
+		headerFilter {
+			patternMatch(patternMatch)
+			headersToRemove(*StringUtils.delimitedListToStringArray(headersToRemove, ",", " "))
+		}
+	}
+
+	/**
+	 * Provide the [HeaderFilter] to the current [IntegrationFlow].
+	 */
+	fun headerFilter(endpointConfigurer: HeaderFilterSpec.() -> Unit) {
+		this.delegate.headerFilter(endpointConfigurer)
 	}
 
 	/**

spring-integration-core/src/test/java/org/springframework/integration/dsl/correlation/CorrelationHandlerTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2022 the original author or authors.
+ * Copyright 2016-2023 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.
@@ -60,7 +60,6 @@
 /**
  * @author Artem Bilan
  * @author Gary Russell
- *
  * @since 5.0
  */
 @RunWith(SpringRunner.class)
@@ -241,7 +240,7 @@ public IntegrationFlow splitResequenceFlow(MessageChannel executorChannel, TaskE
 					.enrichHeaders(h ->
 							h.headerFunction(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, Message::getPayload))
 					.resequence(r -> r.releasePartialSequences(true).correlationExpression("'foo'"))
-					.headerFilter("foo", false);
+					.headerFilter(headerFilterSpec -> headerFilterSpec.headersToRemove("foo").patternMatch(false));
 		}
 
 

spring-integration-core/src/test/kotlin/org/springframework/integration/dsl/KotlinDslTests.kt

@@ -182,11 +182,16 @@ class KotlinDslTests {
 	@Test
 	fun `flow from lambda`() {
 		val replyChannel = QueueChannel()
-		val message = MessageBuilder.withPayload("test").setReplyChannel(replyChannel).build()
+		val message = MessageBuilder.withPayload("test")
+				.setHeader("headerToRemove", "no value")
+				.setReplyChannel(replyChannel)
+				.build()
 
 		this.flowLambdaInput.send(message)
 
-		assertThat(replyChannel.receive(10_000)?.payload).isNotNull().isEqualTo("TEST")
+		val receive = replyChannel.receive(10_000)
+		assertThat(receive?.payload).isNotNull().isEqualTo("TEST")
+		assertThat(receive.headers).doesNotContain("headerToRemove", null)
 		assertThat(this.wireTapChannel.receive(10_000)?.payload).isNotNull().isEqualTo("test")
 	}
 
@@ -308,6 +313,10 @@ class KotlinDslTests {
 		fun flowLambda() =
 			integrationFlow {
 				filter<String>({ it === "test" }) { id("filterEndpoint") }
+				headerFilter {
+					patternMatch(false)
+					headersToRemove("notAHeader", "headerToRemove")
+				}
 				wireTap {
 					channel { queue("wireTapChannel") }
 				}

spring-integration-groovy/src/main/groovy/org/springframework/integration/groovy/dsl/GroovyIntegrationFlowDefinition.groovy

@@ -35,6 +35,7 @@ import org.springframework.integration.dsl.FilterEndpointSpec
 import org.springframework.integration.dsl.GatewayEndpointSpec
 import org.springframework.integration.dsl.GenericEndpointSpec
 import org.springframework.integration.dsl.HeaderEnricherSpec
+import org.springframework.integration.dsl.HeaderFilterSpec
 import org.springframework.integration.dsl.IntegrationFlow
 import org.springframework.integration.dsl.IntegrationFlowDefinition
 import org.springframework.integration.dsl.MessageChannelSpec
@@ -818,8 +819,10 @@ class GroovyIntegrationFlowDefinition {
 	 * {@link HeaderFilter}.
 	 * @param headerFilter the {@link HeaderFilter} to use.
 	 * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options.
+	 * @deprecated since 6.2 in favor of {@link #headerFilter(groovy.lang.Closure)}
 	 * @see GenericEndpointSpec
 	 */
+	@Deprecated(since = "6.2", forRemoval = true)
 	GroovyIntegrationFlowDefinition headerFilter(
 			String headersToRemove,
 			boolean patternMatch = true,
@@ -834,6 +837,21 @@ class GroovyIntegrationFlowDefinition {
 		this
 	}
 
+	/**
+	 * Populate {@link HeaderFilter} based on the options from a {@link HeaderFilterSpec}.
+	 * @param endpointConfigurer the {@link Consumer} to provide {@link HeaderFilter} and its endpoint options.
+	 * @see HeaderFilterSpec
+	 * @since 6.2
+	 */
+	GroovyIntegrationFlowDefinition headerFilter(
+			@DelegatesTo(value = HeaderFilterSpec, strategy = Closure.DELEGATE_FIRST)
+			@ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.HeaderFilterSpec')
+					Closure<?> headerFilterConfigurer) {
+
+		this.delegate.headerFilter createConfigurerIfAny(headerFilterConfigurer)
+		this
+	}
+
 	/**
 	 * Populate the {@link MessageTransformingHandler} for the
 	 * {@link org.springframework.integration.transformer.ClaimCheckInTransformer} with provided {@link MessageStore}.

spring-integration-groovy/src/test/groovy/org/springframework/integration/groovy/dsl/test/GroovyDslTests.groovy

@@ -189,11 +189,18 @@ class GroovyDslTests {
 	@Test
 	void 'flow from lambda'() {
 		def replyChannel = new QueueChannel()
-		def message = MessageBuilder.withPayload('test').setReplyChannel(replyChannel).build()
+		def message =
+				MessageBuilder.withPayload('test')
+						.setHeader('headerToRemove', 'no value')
+						.setReplyChannel(replyChannel)
+						.build()
 
 		this.flowLambdaInput.send message
 
-		assert replyChannel.receive(10_000)?.payload == 'TEST'
+		def receive = replyChannel.receive(10_000)
+
+		assert receive?.payload == 'TEST'
+		assert !receive?.headers?.containsKey('headerToRemove')
 		assert this.wireTapChannel.receive(10_000)?.payload == 'test'
 	}
 
@@ -300,6 +307,10 @@ class GroovyDslTests {
 		flowLambda() {
 			integrationFlow {
 				filter String, { it == 'test' }, { id 'filterEndpoint' }
+				headerFilter {
+					patternMatch false
+					headersToRemove "notAHeader", "headerToRemove"
+				}
 				wireTap integrationFlow {
 					channel { queue 'wireTapChannel' }
 				}