diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java index cf44d5f7d5..16e7df447c 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java @@ -1,5 +1,5 @@ /* - * Copyright 2001-2016 the original author or authors. + * Copyright 2001-2017 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. @@ -241,6 +241,16 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler 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); + } + @Override public String getComponentType() { return "ip:tcp-outbound-gateway"; diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/ThreadAffinityClientConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/ThreadAffinityClientConnectionFactory.java new file mode 100644 index 0000000000..eef7f3bd9a --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/ThreadAffinityClientConnectionFactory.java @@ -0,0 +1,538 @@ +/* + * Copyright 2017 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 + * + * http://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.ip.tcp.connection; + +import java.util.List; +import java.util.concurrent.Executor; + +import javax.net.ssl.SSLSession; + +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.BeanFactory; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ApplicationEventPublisher; +import org.springframework.core.convert.ConversionService; +import org.springframework.core.serializer.Deserializer; +import org.springframework.core.serializer.Serializer; +import org.springframework.expression.Expression; +import org.springframework.integration.support.MessageBuilderFactory; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.core.DestinationResolver; +import org.springframework.util.Assert; + +/** + * A client connection factory that binds a connection to a thread. Close operations + * are ignored; to physically close a connection and release the thread local, invoke + * {@link #releaseConnection()}. + * + * @author Gary Russell + * @since 5.0 + * + */ +public class ThreadAffinityClientConnectionFactory extends AbstractClientConnectionFactory { + + private final AbstractClientConnectionFactory connectionFactory; + + /* + * Not static because we might have several factories with different delegates. + */ + private final ThreadLocal connections = new ThreadLocal<>(); + + public ThreadAffinityClientConnectionFactory(AbstractClientConnectionFactory connectionFactory) { + super("", 0); + Assert.isTrue(connectionFactory.isSingleUse(), + "ConnectionFactory must be single-use to assign a connection per thread"); + this.connectionFactory = connectionFactory; + } + + @Override + public TcpConnectionSupport getConnection() throws Exception { + TcpThreadConnection connection = this.connections.get(); + if (connection == null || !connection.isOpen()) { + TcpConnectionSupport delegate = this.connectionFactory.getConnection(); + connection = new TcpThreadConnection(delegate); + this.connections.set(connection); + } + return connection; + } + + public void releaseConnection() { + TcpThreadConnection connection = this.connections.get(); + if (connection != null) { + this.connections.remove(); + connection.connection.close(); + } + } + + /* + * The following are all delegate methods. + */ + + @Override + public void enableManualListenerRegistration() { + this.connectionFactory.enableManualListenerRegistration(); + } + + @Override + public String getComponentName() { + return this.connectionFactory.getComponentName(); + } + + @Override + public void setComponentName(String componentName) { + this.connectionFactory.setComponentName(componentName); + } + + @Override + public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) { + this.connectionFactory.setApplicationEventPublisher(applicationEventPublisher); + } + + @Override + public String getComponentType() { + return this.connectionFactory.getComponentType(); + } + + @Override + public void setBeanFactory(BeanFactory beanFactory) { + this.connectionFactory.setBeanFactory(beanFactory); + } + + @Override + public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { + this.connectionFactory.setApplicationContext(applicationContext); + } + + @Override + public ApplicationEventPublisher getApplicationEventPublisher() { + return this.connectionFactory.getApplicationEventPublisher(); + } + + @Override + public void setChannelResolver(DestinationResolver channelResolver) { + this.connectionFactory.setChannelResolver(channelResolver); + } + + @Override + public Expression getExpression() { + return this.connectionFactory.getExpression(); + } + + @Override + public void forceClose(TcpConnection connection) { + this.connectionFactory.forceClose(connection); + } + + @Override + public int getSoTimeout() { + return this.connectionFactory.getSoTimeout(); + } + + @Override + public void setSoTimeout(int soTimeout) { + this.connectionFactory.setSoTimeout(soTimeout); + } + + @Override + public int getSoReceiveBufferSize() { + return this.connectionFactory.getSoReceiveBufferSize(); + } + + @Override + public void setSoReceiveBufferSize(int soReceiveBufferSize) { + this.connectionFactory.setSoReceiveBufferSize(soReceiveBufferSize); + } + + @Override + public int getSoSendBufferSize() { + return this.connectionFactory.getSoSendBufferSize(); + } + + @Override + public void setSoSendBufferSize(int soSendBufferSize) { + this.connectionFactory.setSoSendBufferSize(soSendBufferSize); + } + + @Override + public boolean isSoTcpNoDelay() { + return this.connectionFactory.isSoTcpNoDelay(); + } + + @Override + public void setSoTcpNoDelay(boolean soTcpNoDelay) { + this.connectionFactory.setSoTcpNoDelay(soTcpNoDelay); + } + + @Override + public int getSoLinger() { + return this.connectionFactory.getSoLinger(); + } + + @Override + public void setSoLinger(int soLinger) { + this.connectionFactory.setSoLinger(soLinger); + } + + @Override + public boolean isSoKeepAlive() { + return this.connectionFactory.isSoKeepAlive(); + } + + @Override + public void setSoKeepAlive(boolean soKeepAlive) { + this.connectionFactory.setSoKeepAlive(soKeepAlive); + } + + @Override + public ConversionService getConversionService() { + return this.connectionFactory.getConversionService(); + } + + @Override + public int getSoTrafficClass() { + return this.connectionFactory.getSoTrafficClass(); + } + + @Override + public void setSoTrafficClass(int soTrafficClass) { + this.connectionFactory.setSoTrafficClass(soTrafficClass); + } + + @Override + public void setHost(String host) { + this.connectionFactory.setHost(host); + } + + @Override + public String getHost() { + return this.connectionFactory.getHost(); + } + + @Override + public void setPort(int port) { + this.connectionFactory.setPort(port); + } + + @Override + public String getApplicationContextId() { + return this.connectionFactory.getApplicationContextId(); + } + + @Override + public int getPort() { + return this.connectionFactory.getPort(); + } + + @Override + public TcpListener getListener() { + return this.connectionFactory.getListener(); + } + + @Override + public TcpSender getSender() { + return this.connectionFactory.getSender(); + } + + @Override + public Serializer getSerializer() { + return this.connectionFactory.getSerializer(); + } + + @Override + public Deserializer getDeserializer() { + return this.connectionFactory.getDeserializer(); + } + + @Override + public TcpMessageMapper getMapper() { + return this.connectionFactory.getMapper(); + } + + @Override + public void registerListener(TcpListener listener) { + this.connectionFactory.registerListener(listener); + } + + @Override + public void setMessageBuilderFactory(MessageBuilderFactory messageBuilderFactory) { + this.connectionFactory.setMessageBuilderFactory(messageBuilderFactory); + } + + @Override + public void registerSender(TcpSender sender) { + this.connectionFactory.registerSender(sender); + } + + @Override + public void setTaskExecutor(Executor taskExecutor) { + this.connectionFactory.setTaskExecutor(taskExecutor); + } + + @Override + public void setDeserializer(Deserializer deserializer) { + this.connectionFactory.setDeserializer(deserializer); + } + + @Override + public void setSerializer(Serializer serializer) { + this.connectionFactory.setSerializer(serializer); + } + + @Override + public void setMapper(TcpMessageMapper mapper) { + this.connectionFactory.setMapper(mapper); + } + + @Override + public boolean isSingleUse() { + return this.connectionFactory.isSingleUse(); + } + + @Override + public void setSingleUse(boolean singleUse) { + this.connectionFactory.setSingleUse(singleUse); + } + + @Override + public void setLeaveOpen(boolean leaveOpen) { + this.connectionFactory.setLeaveOpen(leaveOpen); + } + + @Override + public void setInterceptorFactoryChain(TcpConnectionInterceptorFactoryChain interceptorFactoryChain) { + this.connectionFactory.setInterceptorFactoryChain(interceptorFactoryChain); + } + + @Override + public void setLookupHost(boolean lookupHost) { + this.connectionFactory.setLookupHost(lookupHost); + } + + @Override + public boolean isLookupHost() { + return this.connectionFactory.isLookupHost(); + } + + @Override + public void setNioHarvestInterval(int nioHarvestInterval) { + this.connectionFactory.setNioHarvestInterval(nioHarvestInterval); + } + + @Override + public void setSslHandshakeTimeout(int sslHandshakeTimeout) { + this.connectionFactory.setSslHandshakeTimeout(sslHandshakeTimeout); + } + + @Override + public void setReadDelay(long readDelay) { + this.connectionFactory.setReadDelay(readDelay); + } + + @Override + public void start() { + this.connectionFactory.start(); + } + + @Override + public void stop() { + this.connectionFactory.stop(); + } + + @Override + public boolean isRunning() { + return this.connectionFactory.isRunning(); + } + + @Override + public void setTcpSocketSupport(TcpSocketSupport tcpSocketSupport) { + this.connectionFactory.setTcpSocketSupport(tcpSocketSupport); + } + + @Override + public List getOpenConnectionIds() { + return this.connectionFactory.getOpenConnectionIds(); + } + + @Override + public boolean closeConnection(String connectionId) { + return this.connectionFactory.closeConnection(connectionId); + } + + @Override + public String toString() { + return getClass().getSimpleName() + ":" + this.connectionFactory.toString(); + } + + /** + * Delegates all calls except {@link #close()} to the wrapped connection. + */ + private static class TcpThreadConnection extends TcpConnectionSupport { + + private final TcpConnectionSupport connection; + + TcpThreadConnection(TcpConnectionSupport connection) { + this.connection = connection; + } + + @Override + public boolean isOpen() { + return this.connection.isOpen(); + } + + @Override + public void send(Message message) throws Exception { + this.connection.send(message); + } + + @Override + public int hashCode() { + return this.connection.hashCode(); + } + + @Override + public Object getPayload() throws Exception { + return this.connection.getPayload(); + } + + @Override + public int getPort() { + return this.connection.getPort(); + } + + @Override + public Object getDeserializerStateKey() { + return this.connection.getDeserializerStateKey(); + } + + @Override + public SSLSession getSslSession() { + return this.connection.getSslSession(); + } + + @Override + public boolean equals(Object obj) { + return this.connection.equals(obj); + } + + @Override + public void close() { + // empty + } + + @Override + public TcpMessageMapper getMapper() { + return this.connection.getMapper(); + } + + @Override + public void setMapper(TcpMessageMapper mapper) { + this.connection.setMapper(mapper); + } + + @Override + public Deserializer getDeserializer() { + return this.connection.getDeserializer(); + } + + @Override + public void setDeserializer(Deserializer deserializer) { + this.connection.setDeserializer(deserializer); + } + + @Override + public Serializer getSerializer() { + return this.connection.getSerializer(); + } + + @Override + public void setSerializer(Serializer serializer) { + this.connection.setSerializer(serializer); + } + + @Override + public void registerListener(TcpListener listener) { + this.connection.registerListener(listener); + } + + @Override + public void enableManualListenerRegistration() { + this.connection.enableManualListenerRegistration(); + } + + @Override + public void registerSender(TcpSender sender) { + this.connection.registerSender(sender); + } + + @Override + public TcpListener getListener() { + return this.connection.getListener(); + } + + @Override + public TcpSender getSender() { + return this.connection.getSender(); + } + + @Override + public boolean isServer() { + return this.connection.isServer(); + } + + @Override + public long incrementAndGetConnectionSequence() { + return this.connection.incrementAndGetConnectionSequence(); + } + + @Override + public String getHostAddress() { + return this.connection.getHostAddress(); + } + + @Override + public String getHostName() { + return this.connection.getHostName(); + } + + @Override + public String getConnectionId() { + return this.connection.getConnectionId(); + } + + @Override + public String toString() { + return "TcpThreadConnection:" + this.connection.getConnectionId(); + } + + @Override + public SocketInfo getSocketInfo() { + return this.connection.getSocketInfo(); + } + + @Override + public void publishEvent(TcpConnectionEvent event) { + this.connection.publishEvent(event); + } + + @Override + public void run() { + } + + } + +} diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ThreadAffinityClientConnectionFactoryTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ThreadAffinityClientConnectionFactoryTests.java new file mode 100644 index 0000000000..393196a201 --- /dev/null +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ThreadAffinityClientConnectionFactoryTests.java @@ -0,0 +1,236 @@ +/* + * Copyright 2017 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 + * + * http://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.ip.tcp.connection; + +import static org.hamcrest.CoreMatchers.containsString; +import static org.hamcrest.CoreMatchers.equalTo; +import static org.hamcrest.CoreMatchers.not; +import static org.hamcrest.Matchers.lessThan; +import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; + +import java.util.Collections; + +import org.junit.Test; + +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.integration.annotation.ServiceActivator; +import org.springframework.integration.annotation.Transformer; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.config.EnableIntegration; +import org.springframework.integration.ip.IpHeaders; +import org.springframework.integration.ip.tcp.TcpInboundGateway; +import org.springframework.integration.ip.tcp.TcpOutboundGateway; +import org.springframework.integration.transformer.ObjectToStringTransformer; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.support.GenericMessage; + +/** + * @author Gary Russell + * @since 5.0 + * + */ +public class ThreadAffinityClientConnectionFactoryTests { + + private static final String PORT = "ThreadAffinityClientConnectionFactoryTests.port"; + + @Test + public void testAffinityNet() throws Exception { + AnnotationConfigApplicationContext server = new AnnotationConfigApplicationContext(ServerNet.class); + TcpNetServerConnectionFactory serverCF = server.getBean(TcpNetServerConnectionFactory.class); + int port = waitForPort(serverCF); + System.setProperty(PORT, String.valueOf(port)); + AnnotationConfigApplicationContext client = new AnnotationConfigApplicationContext(ClientNet.class); + doTest(server, serverCF, client); + } + + @Test + public void testAffinityNio() throws Exception { + AnnotationConfigApplicationContext server = new AnnotationConfigApplicationContext(ServerNio.class); + TcpNioServerConnectionFactory serverCF = server.getBean(TcpNioServerConnectionFactory.class); + int port = waitForPort(serverCF); + System.setProperty(PORT, String.valueOf(port)); + AnnotationConfigApplicationContext client = new AnnotationConfigApplicationContext(ClientNio.class); + doTest(server, serverCF, client); + } + + private int waitForPort(AbstractServerConnectionFactory serverCF) throws InterruptedException { + int port = serverCF.getPort(); + int n = 0; + while (n++ < 200 && port == 0) { + Thread.sleep(100); + port = serverCF.getPort(); + } + assertTrue(n < 200); + return port; + } + + protected void doTest(AnnotationConfigApplicationContext server, AbstractServerConnectionFactory serverCF, + AnnotationConfigApplicationContext client) throws InterruptedException { + MessageChannel channel = client.getBean("out", MessageChannel.class); + QueueChannel replies = new QueueChannel(); + Message message = new GenericMessage<>("foo", + Collections.singletonMap(MessageHeaders.REPLY_CHANNEL, replies)); + channel.send(message); + channel.send(message); + ThreadAffinityClientConnectionFactory clientFactory = client + .getBean(ThreadAffinityClientConnectionFactory.class); + clientFactory.releaseConnection(); + channel.send(message); + channel.send(message); + clientFactory.releaseConnection(); + assertThat(replies.getQueueSize(), equalTo(4)); + Message replyA = replies.receive(0); + Message replyB = replies.receive(0); + Message replyC = replies.receive(0); + Message replyD = replies.receive(0); + assertThat((String) replyA.getPayload(), containsString("ip_connectionId")); + assertThat(replyA.getPayload(), equalTo(replyB.getPayload())); + assertThat(replyC.getPayload(), equalTo(replyD.getPayload())); + assertThat(replyC.getPayload(), not(equalTo(replyA.getPayload()))); + System.getProperties().remove(PORT); + int n = 0; + while (n++ < 200 && serverCF.getOpenConnectionIds().size() > 0) { + Thread.sleep(100); + } + assertThat(n, lessThan(200)); + client.close(); + server.close(); + } + + @Configuration + @EnableIntegration + public static class ServerNet { + + @Bean + public TcpNetServerConnectionFactory sf() { + return new TcpNetServerConnectionFactory(0); + } + + @Bean + public TcpInboundGateway inGate() { + TcpInboundGateway inGate = new TcpInboundGateway(); + inGate.setConnectionFactory(sf()); + inGate.setRequestChannelName("in"); + return inGate; + } + + @ServiceActivator(inputChannel = "in") + public String handle(Message message) { + return IpHeaders.CONNECTION_ID + ":" + (String) message.getHeaders().get(IpHeaders.CONNECTION_ID); + } + + } + + @Configuration + @EnableIntegration + public static class ClientNet { + + @Bean + public TcpNetClientConnectionFactory cf() { + TcpNetClientConnectionFactory cf = new TcpNetClientConnectionFactory("localhost", + Integer.parseInt(System.getProperty(PORT))); + cf.setSingleUse(true); + return cf; + } + + @Bean + public ThreadAffinityClientConnectionFactory tacf() { + return new ThreadAffinityClientConnectionFactory(cf()); + } + + @Bean + @ServiceActivator(inputChannel = "out") + public TcpOutboundGateway outGate() { + TcpOutboundGateway outGate = new TcpOutboundGateway(); + outGate.setConnectionFactory(tacf()); + outGate.setReplyChannelName("toString"); + return outGate; + } + + @Bean + @Transformer(inputChannel = "toString") + public ObjectToStringTransformer otst() { + return new ObjectToStringTransformer(); + } + + } + + @Configuration + @EnableIntegration + public static class ServerNio { + + @Bean + public TcpNioServerConnectionFactory sf() { + return new TcpNioServerConnectionFactory(0); + } + + @Bean + public TcpInboundGateway inGate() { + TcpInboundGateway inGate = new TcpInboundGateway(); + inGate.setConnectionFactory(sf()); + inGate.setRequestChannelName("in"); + return inGate; + } + + @ServiceActivator(inputChannel = "in") + public String handle(Message message) { + return IpHeaders.CONNECTION_ID + ":" + (String) message.getHeaders().get(IpHeaders.CONNECTION_ID); + } + + } + + @Configuration + @EnableIntegration + public static class ClientNio { + + @Bean + public TcpNioClientConnectionFactory cf() { + TcpNioClientConnectionFactory cf = new TcpNioClientConnectionFactory("localhost", + Integer.parseInt(System.getProperty(PORT))); + cf.setSingleUse(true); + return cf; + } + + @Bean + public ThreadAffinityClientConnectionFactory tacf() { + return new ThreadAffinityClientConnectionFactory(cf()); + } + + @Bean + @ServiceActivator(inputChannel = "out") + public TcpOutboundGateway outGate() { + TcpOutboundGateway outGate = new TcpOutboundGateway(); + outGate.setConnectionFactory(tacf()); + outGate.setReplyChannelName("toString"); + return outGate; + } + + @Bean + @Transformer(inputChannel = "toString") + public ObjectToStringTransformer otst() { + return new ObjectToStringTransformer(); + } + + } + + +} diff --git a/src/reference/asciidoc/changes-4.1-4.2.adoc b/src/reference/asciidoc/changes-4.1-4.2.adoc index 7f3b5d1908..4a149bb039 100644 --- a/src/reference/asciidoc/changes-4.1-4.2.adoc +++ b/src/reference/asciidoc/changes-4.1-4.2.adoc @@ -158,7 +158,7 @@ If a TCP server socket factory is configured to listen on a random port, the act be obtained using `getPort()`. `getServerSocketAddress()` is also available. -See <> for more information. +See <> for more information. [[x4.2-tcp-gw-rto]] ===== TCP Gateway Remote Timeout diff --git a/src/reference/asciidoc/changes-4.2-4.3.adoc b/src/reference/asciidoc/changes-4.2-4.3.adoc index 8f058da274..2d2659c73f 100644 --- a/src/reference/asciidoc/changes-4.2-4.3.adoc +++ b/src/reference/asciidoc/changes-4.2-4.3.adoc @@ -127,7 +127,7 @@ See <> for more information. The various deserializers that can't allocate the final buffer until the whole message has been assembled now support pooling of the raw buffer into which the data is received, rather than creating and discarding a buffer for each message. -See <> for more information. +See <> for more information. ===== TCP Message Mapper diff --git a/src/reference/asciidoc/ip.adoc b/src/reference/asciidoc/ip.adoc index 8a17fa780e..c572938515 100644 --- a/src/reference/asciidoc/ip.adoc +++ b/src/reference/asciidoc/ip.adoc @@ -165,7 +165,7 @@ By default, reverse DNS lookups are done on inbound packets to convert IP addres In environments where DNS is not configured, this can cause delays. This default behavior can be overridden by setting the `lookup-host` attribute to "false". -[[connection-factories]] +[[tcp-connection-factories]] === TCP Connection Factories For TCP, the configuration of the underlying connection is provided using a Connection Factory. @@ -364,6 +364,41 @@ Initially, the first factory in the configured list is used; if a connection sub NOTE: When using the failover connection factory, the singleUse property must be consistent between the factory itself and the list of factories it is configured to use. +[[tcp-affinity-cf]] +==== TCP Thread Affinity Connection Factory + +Spring Integration _version 5.0_ introduced this connection factory. +It binds a connection to the calling thread and the same connection is reused each time that thread sends a message. +This continues until the connection is closed (by the server or network) or until the thread calls the `releaseConnection()` method. +The connections themselves are provided by another client factory implementation; which must be configured to provide non-shared (single-use) connections so that each thread gets a connection. + +Example configuration: + +[source, java] +---- +@Bean +public TcpNetClientConnectionFactory cf() { + TcpNetClientConnectionFactory cf = new TcpNetClientConnectionFactory("localhost", + Integer.parseInt(System.getProperty(PORT))); + cf.setSingleUse(true); + return cf; +} + +@Bean +public ThreadAffinityClientConnectionFactory tacf() { + return new ThreadAffinityClientConnectionFactory(cf()); +} + +@Bean +@ServiceActivator(inputChannel = "out") +public TcpOutboundGateway outGate() { + TcpOutboundGateway outGate = new TcpOutboundGateway(); + outGate.setConnectionFactory(tacf()); + outGate.setReplyChannelName("toString"); + return outGate; +} +---- + [[ip-interceptors]] === TCP Connection Interceptors @@ -429,7 +464,7 @@ Beginning with version 3.0, changes to `TcpConnection` s are reported by `TcpCon * `throwable` - the `Throwable` (for `TcpConnectionExceptionEvent` events only) * `source` - the `TcpConnection`; this can be used, for example, to determine the remote IP Address with `getHostAddress()` (cast required) -In addition, since _version 4.0_ the standard deserializers discussed in <> now emit `TcpDeserializationExceptionEvent` s when problems are encountered decoding the data stream. +In addition, since _version 4.0_ the standard deserializers discussed in <> now emit `TcpDeserializationExceptionEvent` s when problems are encountered decoding the data stream. These events contain the exception, the buffer that was in the process of being built, and an offset into the buffer (if available) at the point the exception occurred. Applications can use a normal `ApplicationListener`, or see <>, to capture these events, allowing analysis of the problem. @@ -637,7 +672,7 @@ Such a transformer may transform the original payload to a new object containing Of course, live objects (such as reply channels) from the headers can not be included in the transformed payload. If such a strategy is chosen you will need to ensure the connection factory has an appropriate serializer/deserializer pair to handle such a payload, such as the `DefaultSerializer/Deserializer` which use java serialization, or a custom serializer and deserializer. -The `ByteArray*Serializer` options mentioned in <>, including the default `ByteArrayCrLfSerializer`, do not support such payloads, unless the transformed payload is a `String` or `byte[]`, +The `ByteArray*Serializer` options mentioned in <>, including the default `ByteArrayCrLfSerializer`, do not support such payloads, unless the transformed payload is a `String` or `byte[]`, [NOTE] ===== diff --git a/src/reference/asciidoc/syslog.adoc b/src/reference/asciidoc/syslog.adoc index 9b350d824e..5d9a47073d 100644 --- a/src/reference/asciidoc/syslog.adoc +++ b/src/reference/asciidoc/syslog.adoc @@ -96,7 +96,7 @@ NOTE: When using the `udp-attributes` element, the `port` attribute must be prov A `TCP` adapter that sends messages to channel `fromSyslog`. It also shows how to reference an externally defined connection factory, which can be used for advanced configuration (socket keep alive etc). -For more information, see <>. +For more information, see <>. NOTE: The externally configured `connection-factory` must be of type `server` and, the port is defined there rather than on the `inbound-channel-adapter` element itself. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 1aa6ad7cea..868f9e92f0 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -208,3 +208,8 @@ The `zsetIncrementExpression` can now be configured on the `RedisStoreWritingMes In addition this property has been changed from `true` to `false` since `INCR` option on `ZADD` Redis command is optional. See <> for more information. + +==== TCP Changes + +A new `ThreadAffinityClientConnectionFactory` is provided that binds TCP connections to threads. +See <> for more information.