INT-4166: ThreadAffinityClientConnectionFactory
JIRA: https://jira.spring.io/browse/INT-4166 Binds connections to threads. Polishing - PR Comments * Fix `ip.adoc` typo for `[[tcp-affinity-cf]]`
This commit is contained in:
Gary Russell
committed by
Artem Bilan
Artem Bilan
parent
799dcaae9f
commit
b84a334379
@@ -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";
|
||||
|
||||
@@ -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<TcpThreadConnection> 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<MessageChannel> 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<String> 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() {
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
@@ -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 <<connection-factories>> for more information.
|
||||
See <<tcp-connection-factories>> for more information.
|
||||
|
||||
[[x4.2-tcp-gw-rto]]
|
||||
===== TCP Gateway Remote Timeout
|
||||
|
||||
@@ -127,7 +127,7 @@ See <<udp-adapters>> 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 <<connection-factories>> for more information.
|
||||
See <<tcp-connection-factories>> for more information.
|
||||
|
||||
===== TCP Message Mapper
|
||||
|
||||
|
||||
@@ -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 <<connection-factories>> now emit `TcpDeserializationExceptionEvent` s when problems are encountered decoding the data stream.
|
||||
In addition, since _version 4.0_ the standard deserializers discussed in <<tcp-connection-factories>> 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 <<appevent-inbound>>, 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 <<connection-factories>>, including the default `ByteArrayCrLfSerializer`, do not support such payloads, unless the transformed payload is a `String` or `byte[]`,
|
||||
The `ByteArray*Serializer` options mentioned in <<tcp-connection-factories>>, including the default `ByteArrayCrLfSerializer`, do not support such payloads, unless the transformed payload is a `String` or `byte[]`,
|
||||
|
||||
[NOTE]
|
||||
=====
|
||||
|
||||
@@ -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 <<connection-factories>>.
|
||||
For more information, see <<tcp-connection-factories>>.
|
||||
|
||||
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.
|
||||
|
||||
|
||||
@@ -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 <<redis>> for more information.
|
||||
|
||||
==== TCP Changes
|
||||
|
||||
A new `ThreadAffinityClientConnectionFactory` is provided that binds TCP connections to threads.
|
||||
See <<tcp-affinity-cf>> for more information.
|
||||
|
||||
Reference in New Issue
Block a user