Fix NPEs in DSL Specs (#8597)

* Fix NPEs in DSL Specs

The `BaseWsInboundGatewaySpec` and `TailAdapterSpec` don't override super methods
and when we call them we fail with NPE since `target` was not populated.

* Fix `BaseWsInboundGatewaySpec` and its inheritors to populate the `target`
from their ctors.
* Remove redundant methods from `BaseWsInboundGatewaySpec` hierarchy
* Propagate `AbstractWebServiceInboundGateway` properties directly to the target
from the `BaseWsInboundGatewaySpec` inheritors
* Override `MessageProducerSpec` methods in the `TailAdapterSpec`
to populate respective option into the `FileTailInboundChannelAdapterFactoryBean`
* Expose more missed options for producer endpoint in the `FileTailInboundChannelAdapterFactoryBean`

**Cherry-pick to `6.0.x` & `5.5.x`**

* * `acceptIfNotNull()` for `errorMessageStrategy` in the `FileTailInboundChannelAdapterFactoryBean`

* * Fix `Unmarshaller` population logic in the `MarshallingWsInboundGatewaySpec`

Author: artembilan

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

@@ -35,7 +35,7 @@
 public abstract class MessagingGatewaySpec<S extends MessagingGatewaySpec<S, G>, G extends MessagingGatewaySupport>
 		extends IntegrationComponentSpec<S, G> {
 
-	public MessagingGatewaySpec(@Nullable G gateway) {
+	public MessagingGatewaySpec(G gateway) {
 		this.target = gateway;
 	}
 

spring-integration-file/src/main/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterFactoryBean.java

@@ -29,6 +29,7 @@
 import org.springframework.integration.file.tail.ApacheCommonsFileTailingMessageProducer;
 import org.springframework.integration.file.tail.FileTailingMessageProducerSupport;
 import org.springframework.integration.file.tail.OSDelegatingFileTailingMessageProducer;
+import org.springframework.integration.support.ErrorMessageStrategy;
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.scheduling.TaskScheduler;
@@ -84,6 +85,12 @@ public class FileTailInboundChannelAdapterFactoryBean extends AbstractFactoryBea
 
 	private ApplicationEventPublisher applicationEventPublisher;
 
+	private long sendTimeout;
+
+	private boolean shouldTrack;
+
+	private ErrorMessageStrategy errorMessageStrategy;
+
 	public void setNativeOptions(String nativeOptions) {
 		if (StringUtils.hasText(nativeOptions)) {
 			this.nativeOptions = nativeOptions;
@@ -166,6 +173,18 @@ public void setPhase(int phase) {
 		this.phase = phase;
 	}
 
+	public void setSendTimeout(long sendTimeout) {
+		this.sendTimeout = sendTimeout;
+	}
+
+	public void setShouldTrack(boolean shouldTrack) {
+		this.shouldTrack = shouldTrack;
+	}
+
+	public void setErrorMessageStrategy(ErrorMessageStrategy errorMessageStrategy) {
+		this.errorMessageStrategy = errorMessageStrategy;
+	}
+
 	@Override
 	public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
 		this.applicationEventPublisher = applicationEventPublisher;
@@ -242,6 +261,8 @@ protected FileTailingMessageProducerSupport createInstance() {
 		adapter.setOutputChannel(this.outputChannel);
 		adapter.setErrorChannel(this.errorChannel);
 		adapter.setBeanName(this.beanName);
+		adapter.setSendTimeout(this.sendTimeout);
+		adapter.setShouldTrack(this.shouldTrack);
 		BeanFactory beanFactory = getBeanFactory();
 		JavaUtils.INSTANCE
 				.acceptIfNotNull(this.taskExecutor, adapter::setTaskExecutor)
@@ -253,6 +274,7 @@ protected FileTailingMessageProducerSupport createInstance() {
 				.acceptIfNotNull(this.applicationEventPublisher, adapter::setApplicationEventPublisher)
 				.acceptIfNotNull(this.outputChannelName, adapter::setOutputChannelName)
 				.acceptIfNotNull(this.errorChannelName, adapter::setErrorChannelName)
+				.acceptIfNotNull(this.errorMessageStrategy, adapter::setErrorMessageStrategy)
 				.acceptIfNotNull(beanFactory, adapter::setBeanFactory);
 		adapter.afterPropertiesSet();
 		this.tailAdapter = adapter;

spring-integration-file/src/main/java/org/springframework/integration/file/dsl/TailAdapterSpec.java

@@ -23,6 +23,7 @@
 import org.springframework.integration.dsl.MessageProducerSpec;
 import org.springframework.integration.file.config.FileTailInboundChannelAdapterFactoryBean;
 import org.springframework.integration.file.tail.FileTailingMessageProducerSupport;
+import org.springframework.integration.support.ErrorMessageStrategy;
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.scheduling.TaskScheduler;
@@ -40,12 +41,6 @@ public class TailAdapterSpec extends MessageProducerSpec<TailAdapterSpec, FileTa
 
 	private final FileTailInboundChannelAdapterFactoryBean factoryBean = new FileTailInboundChannelAdapterFactoryBean();
 
-	@Nullable
-	private MessageChannel outputChannel;
-
-	@Nullable
-	private MessageChannel errorChannel;
-
 	protected TailAdapterSpec() {
 		super(null);
 		this.factoryBean.setBeanFactory(new DefaultListableBeanFactory());
@@ -148,7 +143,7 @@ public TailAdapterSpec end(boolean end) {
 	/**
 	 * If {@code true}, close and reopen the file between reading chunks.
 	 * Default {@code false}.
-	 * @param reopen the reopen.
+	 * @param reopen the 'reopen' option.
 	 * @return the spec.
 	 * @see org.springframework.integration.file.tail.ApacheCommonsFileTailingMessageProducer#setReopen(boolean)
 	 */
@@ -177,13 +172,43 @@ public TailAdapterSpec autoStartup(boolean autoStartup) {
 
 	@Override
 	public TailAdapterSpec outputChannel(MessageChannel outputChannel) {
-		this.outputChannel = outputChannel;
+		this.factoryBean.setOutputChannel(outputChannel);
 		return _this();
 	}
 
 	@Override
 	public TailAdapterSpec errorChannel(MessageChannel errorChannel) {
-		this.errorChannel = errorChannel;
+		this.factoryBean.setErrorChannel(errorChannel);
+		return _this();
+	}
+
+	@Override
+	public TailAdapterSpec outputChannel(String outputChannel) {
+		this.factoryBean.setOutputChannelName(outputChannel);
+		return _this();
+	}
+
+	@Override
+	public TailAdapterSpec errorChannel(String errorChannel) {
+		this.factoryBean.setErrorChannelName(errorChannel);
+		return _this();
+	}
+
+	@Override
+	public TailAdapterSpec sendTimeout(long sendTimeout) {
+		this.factoryBean.setSendTimeout(sendTimeout);
+		return _this();
+	}
+
+	@Override
+	public TailAdapterSpec shouldTrack(boolean shouldTrack) {
+		this.factoryBean.setShouldTrack(shouldTrack);
+		return _this();
+	}
+
+	@Override
+	public TailAdapterSpec errorMessageStrategy(ErrorMessageStrategy errorMessageStrategy) {
+		this.factoryBean.setErrorMessageStrategy(errorMessageStrategy);
 		return _this();
 	}
 
@@ -199,10 +224,6 @@ protected FileTailingMessageProducerSupport doGet() {
 		}
 		Assert.notNull(tailingMessageProducerSupport,
 				"The 'FileTailInboundChannelAdapterFactoryBean' must not produce null");
-		if (this.errorChannel != null) {
-			tailingMessageProducerSupport.setErrorChannel(this.errorChannel);
-		}
-		tailingMessageProducerSupport.setOutputChannel(this.outputChannel);
 		return tailingMessageProducerSupport;
 	}
 

spring-integration-file/src/test/java/org/springframework/integration/file/dsl/FileTests.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.
@@ -65,6 +65,7 @@
 import org.springframework.integration.file.support.FileExistsMode;
 import org.springframework.integration.file.support.FileUtils;
 import org.springframework.integration.file.tail.ApacheCommonsFileTailingMessageProducer;
+import org.springframework.integration.support.DefaultErrorMessageStrategy;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.integration.test.util.TestUtils;
 import org.springframework.messaging.Message;
@@ -355,9 +356,12 @@ public IntegrationFlow tailFlow() {
 			return IntegrationFlow
 					.from(Files.tailAdapter(new File(tmpDir, "TailTest"))
 							.delay(500)
+							.errorMessageStrategy(new DefaultErrorMessageStrategy())
 							.end(false)
 							.id("tailer")
-							.autoStartup(false))
+							.autoStartup(false)
+							.shouldTrack(true)
+							.sendTimeout(30_000))
 					.transform("hello "::concat)
 					.channel(MessageChannels.queue("tailChannel"))
 					.get();

spring-integration-ws/src/main/java/org/springframework/integration/ws/dsl/BaseWsInboundGatewaySpec.java

@@ -27,6 +27,8 @@
  * @param <E> the target {@link AbstractWebServiceInboundGateway} implementation type.
  *
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 5.3
  *
  */
@@ -35,10 +37,10 @@ public abstract class BaseWsInboundGatewaySpec<
 		extends MessagingGatewaySpec<S, E> {
 
 	/**
-	 * Construct an instance.
+	 * Construct an instance based on the provided {@link AbstractWebServiceInboundGateway}.
 	 */
-	protected BaseWsInboundGatewaySpec() {
-		super(null);
+	protected BaseWsInboundGatewaySpec(E gateway) {
+		super(gateway);
 	}
 
 	/**
@@ -51,15 +53,4 @@ public S headerMapper(SoapHeaderMapper headerMapper) {
 		return _this();
 	}
 
-	@Override
-	protected E doGet() {
-		return assemble(create());
-	}
-
-	protected abstract E create();
-
-	protected E assemble(E gateway) {
-		return gateway;
-	}
-
 }

spring-integration-ws/src/main/java/org/springframework/integration/ws/dsl/MarshallingWsInboundGatewaySpec.java

@@ -24,45 +24,41 @@
  * The spec for a {@link MarshallingWebServiceInboundGateway}.
  *
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 5.3
  *
  */
-public class MarshallingWsInboundGatewaySpec extends BaseWsInboundGatewaySpec<MarshallingWsInboundGatewaySpec,
-		MarshallingWebServiceInboundGateway> {
+public class MarshallingWsInboundGatewaySpec
+		extends BaseWsInboundGatewaySpec<MarshallingWsInboundGatewaySpec, MarshallingWebServiceInboundGateway> {
 
-	protected Marshaller gatewayMarshaller; // NOSONAR
 
-	protected Unmarshaller gatewayUnmarshaller; // NOSONAR
+	protected MarshallingWsInboundGatewaySpec() {
+		super(new MarshallingWebServiceInboundGateway());
+	}
 
 	/**
 	 * Specify a marshaller to use.
 	 * @param marshaller the marshaller.
 	 * @return the spec.
 	 */
 	public MarshallingWsInboundGatewaySpec marshaller(Marshaller marshaller) {
-		this.gatewayMarshaller = marshaller;
+		this.target.setMarshaller(marshaller);
+		if (marshaller instanceof Unmarshaller unmarshaller) {
+			return unmarshaller(unmarshaller);
+		}
 		return this;
 	}
 
 	/**
-	 * Specify an unmarshaller to use. Required if the {@link #gatewayMarshaller} is not also
+	 * Specify an unmarshaller to use. Required if the {@link #marshaller} is not also
 	 * an {@link Unmarshaller}.
 	 * @param unmarshaller the unmarshaller.
 	 * @return the spec.
 	 */
 	public MarshallingWsInboundGatewaySpec unmarshaller(Unmarshaller unmarshaller) {
-		this.gatewayUnmarshaller = unmarshaller;
+		this.target.setUnmarshaller(unmarshaller);
 		return this;
 	}
 
-	@Override
-	protected MarshallingWebServiceInboundGateway create() {
-		if (this.gatewayUnmarshaller != null) {
-			return new MarshallingWebServiceInboundGateway(this.gatewayMarshaller, this.gatewayUnmarshaller);
-		}
-		else {
-			return new MarshallingWebServiceInboundGateway(this.gatewayMarshaller);
-		}
-	}
-
 }

spring-integration-ws/src/main/java/org/springframework/integration/ws/dsl/SimpleWsInboundGatewaySpec.java

@@ -22,30 +22,27 @@
  * The spec for a {@link SimpleWebServiceInboundGateway}.
  *
  * @author Gary Russell
- * @since 5.3
+ * @author Artem Bilan
  *
+ * @since 5.3
  */
-public class SimpleWsInboundGatewaySpec extends BaseWsInboundGatewaySpec<SimpleWsInboundGatewaySpec,
-		SimpleWebServiceInboundGateway> {
+public class SimpleWsInboundGatewaySpec
+		extends BaseWsInboundGatewaySpec<SimpleWsInboundGatewaySpec, SimpleWebServiceInboundGateway> {
 
-	protected boolean extractPayload = true; // NOSONAR
+	protected SimpleWsInboundGatewaySpec() {
+		super(new SimpleWebServiceInboundGateway());
+	}
 
 	/**
 	 * Specify true to extract the payloadSource from the request or use
 	 * the entire request as the payload; default true.
+	 *
 	 * @param extract true to extract.
 	 * @return the spec.
 	 */
 	public SimpleWsInboundGatewaySpec extractPayload(boolean extract) {
-		this.extractPayload = extract;
+		this.target.setExtractPayload(extract);
 		return this;
 	}
 
-	@Override
-	protected SimpleWebServiceInboundGateway create() {
-		SimpleWebServiceInboundGateway gateway = new SimpleWebServiceInboundGateway();
-		gateway.setExtractPayload(this.extractPayload);
-		return gateway;
-	}
-
 }

spring-integration-ws/src/test/java/org/springframework/integration/ws/dsl/WsDslTests.java

@@ -24,6 +24,7 @@
 import org.springframework.expression.Expression;
 import org.springframework.expression.common.LiteralExpression;
 import org.springframework.integration.test.util.TestUtils;
+import org.springframework.integration.ws.DefaultSoapHeaderMapper;
 import org.springframework.integration.ws.MarshallingWebServiceInboundGateway;
 import org.springframework.integration.ws.MarshallingWebServiceOutboundGateway;
 import org.springframework.integration.ws.SimpleWebServiceInboundGateway;
@@ -71,10 +72,16 @@ void marshallingInbound() {
 
 	@Test
 	void simpleInbound() {
-		SimpleWebServiceInboundGateway gateway = Ws.simpleInboundGateway()
-				.extractPayload(false)
-				.getObject();
+		DefaultSoapHeaderMapper testHeaderMapper = new DefaultSoapHeaderMapper();
+		SimpleWebServiceInboundGateway gateway =
+				Ws.simpleInboundGateway()
+						.extractPayload(false)
+						.headerMapper(testHeaderMapper)
+						.errorChannel("myErrorChannel")
+						.getObject();
 		assertThat(TestUtils.getPropertyValue(gateway, "extractPayload", Boolean.class)).isFalse();
+		assertThat(TestUtils.getPropertyValue(gateway, "headerMapper")).isSameAs(testHeaderMapper);
+		assertThat(TestUtils.getPropertyValue(gateway, "errorChannelName")).isEqualTo("myErrorChannel");
 	}
 
 	@Test