GH-3326: TCP: Support Unsolicited Server Messages

Resolves https://github.com/spring-projects/spring-integration/issues/3326

- OB Gateway - send unsolicited messages and late replies to a channel
- Support multiple `TcpSender` s

* Add channel variant to spec; code polishing

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

@@ -26,6 +26,7 @@ import org.springframework.integration.expression.FunctionExpression;
 import org.springframework.integration.ip.tcp.TcpOutboundGateway;
 import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory;
 import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
 
 /**
  * A {@link MessageHandlerSpec} for {@link TcpOutboundGateway}s.
@@ -114,6 +115,28 @@ public class TcpOutboundGatewaySpec extends MessageHandlerSpec<TcpOutboundGatewa
 		return this;
 	}
 
+	/**
+	 * Set the unsolicited message channel name.
+	 * @param channelName the name.
+	 * @return the spec.
+	 * @since 5.4
+	 */
+	public TcpOutboundGatewaySpec unsolictedMessageChannelName(String channelName) {
+		this.target.setUnsolicitedMessageChannelName(channelName);
+		return this;
+	}
+
+	/**
+	 * Set the unsolicited message channel.
+	 * @param channel the channel.
+	 * @return the spec.
+	 * @since 5.4
+	 */
+	public TcpOutboundGatewaySpec unsolictedMessageChannelName(MessageChannel channel) {
+		this.target.setUnsolicitedMessageChannel(channel);
+		return this;
+	}
+
 	@Override
 	public Map<Object, String> getComponentsToRegister() {
 		return this.connectionFactory != null

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

@@ -91,6 +91,17 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler
 
 	private boolean closeStreamAfterSend;
 
+	private String unsolicitedMessageChannelName;
+
+	private MessageChannel unsolicitedMessageChannel;
+
+	public void setConnectionFactory(AbstractClientConnectionFactory connectionFactory) {
+		this.connectionFactory = connectionFactory;
+		connectionFactory.registerListener(this);
+		connectionFactory.registerSender(this);
+		this.isSingleUse = connectionFactory.isSingleUse();
+	}
+
 	/**
 	 * @param requestTimeout the requestTimeout to set
 	 */
@@ -118,14 +129,53 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler
 		this.evaluationContextSet = true;
 	}
 
-	@Override
-	protected void doInit() {
-		super.doInit();
-		if (!this.evaluationContextSet) {
-			this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(getBeanFactory());
-		}
-		Assert.state(!this.closeStreamAfterSend || this.isSingleUse,
-				"Single use connection needed with closeStreamAfterSend");
+	/**
+	 * Specify the Spring Integration reply channel. If this property is not
+	 * set the gateway will check for a 'replyChannel' header on the request.
+	 * @param replyChannel The reply channel.
+	 */
+	public void setReplyChannel(MessageChannel replyChannel) {
+		setOutputChannel(replyChannel);
+	}
+
+	/**
+	 * Specify the Spring Integration reply channel name. If this property is not
+	 * set the gateway will check for a 'replyChannel' header on the request.
+	 * @param replyChannel The reply channel.
+	 * @since 5.0
+	 */
+	public void setReplyChannelName(String replyChannel) {
+		setOutputChannelName(replyChannel);
+	}
+
+	/**
+	 * Set the channel name for unsolicited incoming messages, or late replies.
+	 * @param unsolicitedMessageChannelName the channel name.
+	 * @since 5.4
+	 */
+	public void setUnsolicitedMessageChannelName(String unsolicitedMessageChannelName) {
+		this.unsolicitedMessageChannelName = unsolicitedMessageChannelName;
+	}
+
+	/**
+	 * Set the channel for unsolicited incoming messages, or late replies.
+	 * @param unsolicitedMessageChannel the channel.
+	 * @since 5.4
+	 */
+	public void setUnsolicitedMessageChannel(MessageChannel unsolicitedMessageChannel) {
+		this.unsolicitedMessageChannel = unsolicitedMessageChannel;
+	}
+
+	/**
+	 * Set to true to close the connection ouput stream after sending without
+	 * closing the connection. Use to signal EOF to the server, such as when using
+	 * a {@link org.springframework.integration.ip.tcp.serializer.ByteArrayRawSerializer}.
+	 * Requires a single-use connection factory.
+	 * @param closeStreamAfterSend true to close.
+	 * @since 5.2
+	 */
+	public void setCloseStreamAfterSend(boolean closeStreamAfterSend) {
+		this.closeStreamAfterSend = closeStreamAfterSend;
 	}
 
 	/**
@@ -140,6 +190,21 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler
 		this.secondChanceDelay = secondChanceDelay;
 	}
 
+	@Override
+	public String getComponentType() {
+		return "ip:tcp-outbound-gateway";
+	}
+
+	@Override
+	protected void doInit() {
+		super.doInit();
+		if (!this.evaluationContextSet) {
+			this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(getBeanFactory());
+		}
+		Assert.state(!this.closeStreamAfterSend || this.isSingleUse,
+				"Single use connection needed with closeStreamAfterSend");
+	}
+
 	@Override
 	protected Object handleRequestMessage(Message<?> requestMessage) {
 		Assert.notNull(this.connectionFactory, this.getClass().getName() +
@@ -260,6 +325,9 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler
 	public boolean onMessage(Message<?> message) {
 		String connectionId = message.getHeaders().get(IpHeaders.CONNECTION_ID, String.class);
 		if (connectionId == null) {
+			if (unsolicitedSupported(message)) {
+				return false;
+			}
 			logger.error("Cannot correlate response - no connection id");
 			publishNoConnectionEvent(message, null, "Cannot correlate response - no connection id");
 			return false;
@@ -277,6 +345,9 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler
 				return false;
 			}
 			else {
+				if (unsolicitedSupported(message)) {
+					return false;
+				}
 				String errorMessage = "Cannot correlate response - no pending reply for " + connectionId;
 				logger.error(errorMessage);
 				publishNoConnectionEvent(message, connectionId, errorMessage);
@@ -293,6 +364,24 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler
 		return false;
 	}
 
+	private boolean unsolicitedSupported(Message<?> message) {
+		String channelName = this.unsolicitedMessageChannelName;
+		if (channelName != null) {
+			this.unsolicitedMessageChannel = getChannelResolver().resolveDestination(channelName);
+			this.unsolicitedMessageChannelName = null;
+		}
+		if (this.unsolicitedMessageChannel != null) {
+			try {
+				this.messagingTemplate.send(this.unsolicitedMessageChannel, message);
+			}
+			catch (Exception e) {
+				logger.error("Failed to send unsolicited message " + message, e);
+			}
+			return true;
+		}
+		return false;
+	}
+
 	private void publishNoConnectionEvent(Message<?> message, String connectionId, String errorMessage) {
 		ApplicationEventPublisher applicationEventPublisher = this.connectionFactory.getApplicationEventPublisher();
 		if (applicationEventPublisher != null) {
@@ -301,13 +390,6 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler
 		}
 	}
 
-	public void setConnectionFactory(AbstractClientConnectionFactory connectionFactory) {
-		this.connectionFactory = connectionFactory;
-		connectionFactory.registerListener(this);
-		connectionFactory.registerSender(this);
-		this.isSingleUse = connectionFactory.isSingleUse();
-	}
-
 	@Override
 	public void addNewConnection(TcpConnection connection) {
 		// do nothing - no asynchronous multiplexing supported
@@ -318,42 +400,6 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler
 		// do nothing - no asynchronous multiplexing supported
 	}
 
-	/**
-	 * Specify the Spring Integration reply channel. If this property is not
-	 * set the gateway will check for a 'replyChannel' header on the request.
-	 * @param replyChannel The reply channel.
-	 */
-	public void setReplyChannel(MessageChannel replyChannel) {
-		this.setOutputChannel(replyChannel);
-	}
-
-	/**
-	 * Specify the Spring Integration reply channel name. If this property is not
-	 * set the gateway will check for a 'replyChannel' header on the request.
-	 * @param replyChannel The reply channel.
-	 * @since 5.0
-	 */
-	public void setReplyChannelName(String replyChannel) {
-		this.setOutputChannelName(replyChannel);
-	}
-
-	/**
-	 * Set to true to close the connection ouput stream after sending without
-	 * closing the connection. Use to signal EOF to the server, such as when using
-	 * a {@link org.springframework.integration.ip.tcp.serializer.ByteArrayRawSerializer}.
-	 * Requires a single-use connection factory.
-	 * @param closeStreamAfterSend true to close.
-	 * @since 5.2
-	 */
-	public void setCloseStreamAfterSend(boolean closeStreamAfterSend) {
-		this.closeStreamAfterSend = closeStreamAfterSend;
-	}
-
-	@Override
-	public String getComponentType() {
-		return "ip:tcp-outbound-gateway";
-	}
-
 	@Override
 	public void start() {
 		this.connectionFactory.start();

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

@@ -218,10 +218,7 @@ public abstract class AbstractClientConnectionFactory extends AbstractConnection
 				connection.registerListener(listener);
 			}
 		}
-		TcpSender sender = getSender();
-		if (sender != null) {
-			connection.registerSender(sender);
-		}
+		connection.registerSenders(getSenders());
 		connection.setMapper(getMapper());
 		connection.setDeserializer(getDeserializer());
 		connection.setSerializer(getSerializer());

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

@@ -79,14 +79,14 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
 
 	private final BlockingQueue<PendingIO> delayedReads = new LinkedBlockingQueue<>();
 
+	private final List<TcpSender> senders = Collections.synchronizedList(new ArrayList<>());
+
 	private String host;
 
 	private int port;
 
 	private TcpListener listener;
 
-	private TcpSender sender;
-
 	private int soTimeout = -1;
 
 	private int soSendBufferSize;
@@ -326,11 +326,20 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
 	}
 
 	/**
-	 * @return the sender
+	 * @return the first sender, if present.
 	 */
 	@Nullable
 	public TcpSender getSender() {
-		return this.sender;
+		return this.senders.size() > 0 ? this.senders.get(0) : null;
+	}
+
+	/**
+	 * Return the list of senders.
+	 * @return the senders.
+	 * @since 5.4
+	 */
+	public List<TcpSender> getSenders() {
+		return Collections.unmodifiableList(this.senders);
 	}
 
 	/**
@@ -372,8 +381,16 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
 	 * @param senderToRegister The sender
 	 */
 	public void registerSender(TcpSender senderToRegister) {
-		Assert.isNull(this.sender, this.getClass().getName() + " may only be used by one outbound adapter");
-		this.sender = senderToRegister;
+		this.senders.add(senderToRegister);
+	}
+
+	/**
+	 * Unregister a TcpSender.
+	 * @param sender the sender.
+	 * @return true if the sender was registered.
+	 */
+	public boolean unregisterSender(TcpSender sender) {
+		return this.senders.remove(sender);
 	}
 
 	/**
@@ -600,7 +617,7 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
 				if (this.listener == null) {
 					connection.registerListener(wrapper);
 				}
-				if (this.sender == null) {
+				if (this.senders.size() == 0) {
 					connection.registerSender(wrapper);
 				}
 				connection = wrapper;

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

@@ -127,7 +127,7 @@ public abstract class AbstractServerConnectionFactory extends AbstractConnection
 		if (listener != null) {
 			connection.registerListener(listener);
 		}
-		connection.registerSender(getSender());
+		connection.registerSenders(getSenders());
 		connection.setMapper(getMapper());
 		connection.setDeserializer(getDeserializer());
 		connection.setSerializer(getSerializer());

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

@@ -18,7 +18,9 @@ package org.springframework.integration.ip.tcp.connection;
 
 import java.net.InetAddress;
 import java.net.Socket;
+import java.util.ArrayList;
 import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.CountDownLatch;
@@ -68,6 +70,8 @@ public abstract class TcpConnectionSupport implements TcpConnection {
 
 	private final SocketInfo socketInfo;
 
+	private final List<TcpSender> senders = Collections.synchronizedList(new ArrayList<>());
+
 	@SuppressWarnings("rawtypes")
 	private Deserializer deserializer;
 
@@ -80,8 +84,6 @@ public abstract class TcpConnectionSupport implements TcpConnection {
 
 	private volatile TcpListener testListener;
 
-	private TcpSender sender;
-
 	private String connectionId;
 
 	private String hostName = "unknown";
@@ -162,8 +164,8 @@ public abstract class TcpConnectionSupport implements TcpConnection {
 	 */
 	@Override
 	public void close() {
-		if (this.sender != null) {
-			this.sender.removeDeadConnection(this);
+		for (TcpSender sender : this.senders) {
+			sender.removeDeadConnection(this);
 		}
 		// close() may be called multiple times; only publish once
 		if (!this.closePublished.getAndSet(true)) {
@@ -297,11 +299,25 @@ public abstract class TcpConnectionSupport implements TcpConnection {
 	 * Registers a sender. Used on server side connections so a
 	 * sender can determine which connection to send a reply
 	 * to.
-	 * @param sender the sender.
+	 * @param senderToRegister the sender.
 	 */
-	public void registerSender(@Nullable TcpSender sender) {
-		this.sender = sender;
-		if (sender != null) {
+	public void registerSender(@Nullable TcpSender senderToRegister) {
+		if (senderToRegister != null) {
+			this.senders.add(senderToRegister);
+			senderToRegister.addNewConnection(this);
+		}
+	}
+
+	/**
+	 * Registers the senders. Used on server side connections so a
+	 * sender can determine which connection to send a reply
+	 * to.
+	 * @param sendersToRegister the sender.
+	 * @since 5.4
+	 */
+	public void registerSenders(List<TcpSender> sendersToRegister) {
+		this.senders.addAll(sendersToRegister);
+		for (TcpSender sender : sendersToRegister) {
 			sender.addNewConnection(this);
 		}
 	}
@@ -336,10 +352,20 @@ public abstract class TcpConnectionSupport implements TcpConnection {
 	}
 
 	/**
-	 * @return the sender
+	 * @return the first sender, if present.
 	 */
+	@Nullable
 	public TcpSender getSender() {
-		return this.sender;
+		return this.senders.size() > 0 ? this.senders.get(0) : null;
+	}
+
+	/**
+	 * Return the list of senders.
+	 * @return the senders.
+	 * @since 5.4
+	 */
+	public List<TcpSender> getSenders() {
+		return Collections.unmodifiableList(this.senders);
 	}
 
 	@Override

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

@@ -250,6 +250,11 @@ public class ThreadAffinityClientConnectionFactory extends AbstractClientConnect
 		return this.connectionFactory.getSender();
 	}
 
+	@Override
+	public List<TcpSender> getSenders() {
+		return this.connectionFactory.getSenders();
+	}
+
 	@Override
 	public Serializer<?> getSerializer() {
 		return this.connectionFactory.getSerializer();

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

@@ -18,6 +18,7 @@ package org.springframework.integration.ip.dsl;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
+import java.util.Collections;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -43,6 +44,7 @@ import org.springframework.integration.dsl.MessageChannels;
 import org.springframework.integration.dsl.Transformers;
 import org.springframework.integration.dsl.context.IntegrationFlowContext;
 import org.springframework.integration.dsl.context.IntegrationFlowContext.IntegrationFlowRegistration;
+import org.springframework.integration.ip.IpHeaders;
 import org.springframework.integration.ip.tcp.TcpOutboundGateway;
 import org.springframework.integration.ip.tcp.TcpReceivingChannelAdapter;
 import org.springframework.integration.ip.tcp.TcpSendingMessageHandler;
@@ -110,6 +112,13 @@ public class IpIntegrationTests {
 	@Autowired
 	private AtomicBoolean adviceCalled;
 
+	@Autowired
+	@Qualifier("unsolicitedServerSide.input")
+	private MessageChannel unsolicitedServerSide;
+
+	@Autowired
+	private QueueChannel unsolicited;
+
 	@Test
 	void testTcpAdapters() {
 		ApplicationEventPublisher publisher = e -> { };
@@ -149,6 +158,11 @@ public class IpIntegrationTests {
 		assertThat(messagingTemplate.convertSendAndReceive("foo", String.class)).isEqualTo("FOO");
 		assertThat(messagingTemplate.convertSendAndReceive("junk", String.class)).isEqualTo("error:non-convertible");
 		assertThat(this.adviceCalled.get()).isTrue();
+
+		GenericMessage<String> unsol = new GenericMessage<>("foo",
+				Collections.singletonMap(IpHeaders.CONNECTION_ID, this.config.connectionId));
+		this.unsolicitedServerSide.send(unsol);
+		assertThat(this.unsolicited.receive(10_000).getPayload()).isEqualTo("foo".getBytes());
 	}
 
 	@Test
@@ -227,6 +241,8 @@ public class IpIntegrationTests {
 
 		private volatile int serverPort;
 
+		private volatile String connectionId;
+
 		@Bean
 		public AbstractServerConnectionFactory server1() {
 			return Tcp.netServer(0)
@@ -242,12 +258,23 @@ public class IpIntegrationTests {
 							.replyTimeout(1)
 							.errorOnTimeout(true)
 							.errorChannel("inTcpGatewayErrorFlow.input"))
+					.handle(this, "captureId")
 					.transform(Transformers.objectToString())
 					.<String>filter((payload) -> !"junk".equals(payload))
 					.<String, String>transform(String::toUpperCase)
 					.get();
 		}
 
+		public Message<?> captureId(Message<?> msg) {
+			this.connectionId = msg.getHeaders().get(IpHeaders.CONNECTION_ID, String.class);
+			return msg;
+		}
+
+		@Bean
+		public IntegrationFlow unsolicitedServerSide() {
+			return f -> f.handle(Tcp.outboundAdapter(server1()));
+		}
+
 		@Bean
 		public IntegrationFlow inTcpGatewayErrorFlow() {
 			return (flow) -> flow
@@ -299,9 +326,15 @@ public class IpIntegrationTests {
 		public TcpOutboundGateway tcpOut() {
 			return Tcp.outboundGateway(client1())
 					.remoteTimeout(m -> 5000)
+					.unsolictedMessageChannelName("unsolicited")
 					.get();
 		}
 
+		@Bean
+		public QueueChannel unsolicited() {
+			return new QueueChannel();
+		}
+
 		@Bean
 		public AbstractClientConnectionFactory client2() {
 			return Tcp.netClient("localhost", server1().getPort())