spring-projectsGH-3107: Add errorOnTimeout for TcpInboundGateway

Fixes spring-projects#3107

The `MessagingGatewaySupport` has an `errorOnTimeout` option to throw
a `MessageTimeoutException` when downstream reply doesn't come back in
time for configured reply timeout

* Expose an `errorOnTimeout` option as a `TcpInboundGateway` ctor
property
* Add new factory methods into a `Tcp` factory for Java DSL
* Ensure a property works as expected in the `IpIntegrationTests`
* Document a new option

Author: artembilan

spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/Tcp.java

@@ -95,16 +95,44 @@ public static TcpClientConnectionFactorySpec netClient(String host, int port) {
 	 * @return the spec.
 	 */
 	public static TcpInboundGatewaySpec inboundGateway(AbstractConnectionFactory connectionFactory) {
-		return new TcpInboundGatewaySpec(connectionFactory);
+		return inboundGateway(connectionFactory, false);
 	}
 
+	/**
+	 * Create an inbound gateway using the supplied connection factory.
+	 * @param connectionFactory the connection factory - must be an existing bean - it
+	 * will not be initialized.
+	 * @param errorOnTimeout true to create the error message on reply timeout.
+	 * @return the spec.
+	 * @since 5.2.2
+	 */
+	public static TcpInboundGatewaySpec inboundGateway(AbstractConnectionFactory connectionFactory,
+			boolean errorOnTimeout) {
+
+		return new TcpInboundGatewaySpec(connectionFactory, errorOnTimeout);
+	}
+
+
 	/**
 	 * Create an inbound gateway using the supplied connection factory.
 	 * @param connectionFactorySpec the connection factory spec.
 	 * @return the spec.
 	 */
 	public static TcpInboundGatewaySpec inboundGateway(AbstractConnectionFactorySpec<?, ?> connectionFactorySpec) {
-		return new TcpInboundGatewaySpec(connectionFactorySpec);
+		return inboundGateway(connectionFactorySpec, false);
+	}
+
+	/**
+	 * Create an inbound gateway using the supplied connection factory.
+	 * @param connectionFactorySpec the connection factory spec.
+	 * @param errorOnTimeout true to create the error message on reply timeout.
+	 * @return the spec.
+	 * @since 5.2.2
+	 */
+	public static TcpInboundGatewaySpec inboundGateway(AbstractConnectionFactorySpec<?, ?> connectionFactorySpec,
+			boolean errorOnTimeout) {
+
+		return new TcpInboundGatewaySpec(connectionFactorySpec, errorOnTimeout);
 	}
 
 	/**

spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/TcpInboundGatewaySpec.java

@@ -42,19 +42,21 @@ public class TcpInboundGatewaySpec extends MessagingGatewaySpec<TcpInboundGatewa
 	/**
 	 * Construct an instance using an existing spring-managed connection factory.
 	 * @param connectionFactoryBean the spring-managed bean.
+	 * @param errorOnTimeout true to create the error message on reply timeout.
 	 */
-	TcpInboundGatewaySpec(AbstractConnectionFactory connectionFactoryBean) {
-		super(new TcpInboundGateway());
+	TcpInboundGatewaySpec(AbstractConnectionFactory connectionFactoryBean, boolean errorOnTimeout) {
+		super(new TcpInboundGateway(errorOnTimeout));
 		this.connectionFactory = null;
 		this.target.setConnectionFactory(connectionFactoryBean);
 	}
 
 	/**
 	 * Construct an instance using a connection factory spec.
 	 * @param connectionFactorySpec the spec.
+	 * @param errorOnTimeout true to create the error message on reply timeout.
 	 */
-	TcpInboundGatewaySpec(AbstractConnectionFactorySpec<?, ?> connectionFactorySpec) {
-		super(new TcpInboundGateway());
+	TcpInboundGatewaySpec(AbstractConnectionFactorySpec<?, ?> connectionFactorySpec, boolean errorOnTimeout) {
+		super(new TcpInboundGateway(errorOnTimeout));
 		this.connectionFactory = connectionFactorySpec.get();
 		this.target.setConnectionFactory(this.connectionFactory);
 	}

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpInboundGateway.java

@@ -83,6 +83,21 @@ public class TcpInboundGateway extends MessagingGatewaySupport implements
 
 	private volatile boolean shuttingDown;
 
+	public TcpInboundGateway() {
+		this(false);
+	}
+
+	/**
+	 * Instantiate based on the provided flag to indicate if an {@link ErrorMessage}
+	 * with a {@link org.springframework.integration.MessageTimeoutException} as a payload
+	 * will be sent into an error channel if a reply is expected but none is received.
+	 * @param errorOnTimeout true to create the error message on reply timeout.
+	 * @since 5.2.2
+	 */
+	public TcpInboundGateway(boolean errorOnTimeout) {
+		super(errorOnTimeout);
+	}
+
 	@Override
 	public boolean onMessage(Message<?> message) {
 		boolean isErrorMessage = message instanceof ErrorMessage;

spring-integration-ip/src/test/java/org/springframework/integration/ip/dsl/IpIntegrationTests.java

@@ -229,9 +229,8 @@ public AbstractServerConnectionFactory server1() {
 		@Bean
 		public IntegrationFlow inTcpGateway() {
 			return IntegrationFlows.from(
-					Tcp.inboundGateway(server1())
+					Tcp.inboundGateway(server1(), true)
 							.replyTimeout(1)
-							.errorOnTimeout(true)
 							.errorChannel("inTcpGatewayErrorFlow.input"))
 					.transform(Transformers.objectToString())
 					.<String>filter((payload) -> !"junk".equals(payload))

src/reference/asciidoc/ip.adoc

@@ -1895,6 +1895,8 @@ Usually, replies arrive on a temporary reply channel added to the inbound messag
 |
 | The time in milliseconds for which the gateway waits for a reply.
 Default: 1000 (1 second).
+Starting with version `5.2.2`, the `TcpInboundGateway` can be created with an `errorOnTimeout` boolean flag to raise a `MessageTimeoutException` when downstream reply doesn't come back during reply timeout.
+Such an exception can be handled on the `errorChannel`, e.g. producing a compensation reply for client.
 
 | `error-channel`
 |