Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-4164: TCP Support Pushback and Subclassing

Browse Source 
JIRA: https://jira.spring.io/browse/INT-4164

Previously, it was not possible to create a subclass of `TcpNetConnection`.
This was possible for `TcpNioConnection` via `TcpNioConnectionSupport`.

Add `TcpNetConnectionSupport` and a default implementation.

Create `AbstractTcpConnectionSupport` and add support for all implementation (Net, NIO, NIO with SSL)
to support wrapping the connection's `InputStream` in a `PushBackInputStream`.

checkstyle

Polishing - PR Comments

* Polishing TCP tests a bit:
 - mark some of them with the `LongRunningIntegrationTest` `@Rule`
 - Remove redundant `AFTER_EACH_TEST_METHOD`
 - use Lambda for the `ApplicationEventPublisher` in tests
This commit is contained in:
Gary Russell
2017-05-22 14:48:30 -04:00
committed by Artem Bilan
parent ea89682368
commit 4991d963e8
24 changed files with 929 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-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.
@@ -120,6 +120,8 @@ public abstract class IpAdapterParserUtils {
public static final String NIO_CONNECTION_SUPPORT = "nio-connection-support";
public static final String NET_CONNECTION_SUPPORT = "net-connection-support";
public static final String SOCKET_FACTORY_SUPPORT = "socket-factory-support";
public static final String BACKLOG = "backlog";

26
spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionFactoryFactoryBean.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-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.
@@ -30,6 +30,7 @@ import org.springframework.core.serializer.Serializer;
import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory;
import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory;
import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory;
import org.springframework.integration.ip.tcp.connection.DefaultTcpNetConnectionSupport;
import org.springframework.integration.ip.tcp.connection.DefaultTcpNetSSLSocketFactorySupport;
import org.springframework.integration.ip.tcp.connection.DefaultTcpNetSocketFactorySupport;
import org.springframework.integration.ip.tcp.connection.DefaultTcpNioConnectionSupport;
@@ -38,6 +39,7 @@ import org.springframework.integration.ip.tcp.connection.DefaultTcpSocketSupport
import org.springframework.integration.ip.tcp.connection.TcpConnectionInterceptorFactoryChain;
import org.springframework.integration.ip.tcp.connection.TcpMessageMapper;
import org.springframework.integration.ip.tcp.connection.TcpNetClientConnectionFactory;
import org.springframework.integration.ip.tcp.connection.TcpNetConnectionSupport;
import org.springframework.integration.ip.tcp.connection.TcpNetServerConnectionFactory;
import org.springframework.integration.ip.tcp.connection.TcpNioClientConnectionFactory;
import org.springframework.integration.ip.tcp.connection.TcpNioConnectionSupport;
@@ -118,6 +120,8 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<Abstrac
private volatile TcpNioConnectionSupport nioConnectionSupport;
private volatile TcpNetConnectionSupport netConnectionSupport;
private volatile TcpSocketFactorySupport socketFactorySupport;
private volatile ApplicationEventPublisher applicationEventPublisher;
@@ -178,6 +182,7 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<Abstrac
this.setCommonAttributes(connectionFactory);
this.setServerAttributes(connectionFactory);
connectionFactory.setTcpSocketFactorySupport(this.obtainSocketFactorySupport());
connectionFactory.setTcpNetConnectionSupport(this.obtainNetConnectionSupport());
this.connectionFactory = connectionFactory;
}
else {
@@ -185,6 +190,7 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<Abstrac
this.host, this.port);
this.setCommonAttributes(connectionFactory);
connectionFactory.setTcpSocketFactorySupport(this.obtainSocketFactorySupport());
connectionFactory.setTcpNetConnectionSupport(this.obtainNetConnectionSupport());
this.connectionFactory = connectionFactory;
}
}
@@ -244,6 +250,15 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<Abstrac
}
}
private TcpNetConnectionSupport obtainNetConnectionSupport() {
if (this.netConnectionSupport != null) {
return this.netConnectionSupport;
}
else {
return new DefaultTcpNetConnectionSupport();
}
}
/**
* @param port the port to set
*/
@@ -467,15 +482,16 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<Abstrac
this.socketSupport = tcpSocketSupport;
}
/**
* Rare property - not exposed through namespace
* @param tcpNioSupport The tcpNioSupport to set.
*/
public void setNioConnectionSupport(TcpNioConnectionSupport tcpNioSupport) {
Assert.notNull(tcpNioSupport, "TcpNioConnectionSupport may not be null");
this.nioConnectionSupport = tcpNioSupport;
}
public void setNetConnectionSupport(TcpNetConnectionSupport tcpNetSupport) {
Assert.notNull(tcpNetSupport, "TcpNetConnectionSupport may not be null");
this.netConnectionSupport = tcpNetSupport;
}
public void setSocketFactorySupport(
TcpSocketFactorySupport tcpSocketFactorySupport) {
Assert.notNull(tcpSocketFactorySupport, "TcpSocketFactorySupport may not be null");

4
spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionFactoryParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-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.
@@ -92,6 +92,8 @@ public class TcpConnectionFactoryParser extends AbstractBeanDefinitionParser {
IpAdapterParserUtils.SOCKET_SUPPORT);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
IpAdapterParserUtils.NIO_CONNECTION_SUPPORT);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
IpAdapterParserUtils.NET_CONNECTION_SUPPORT);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
IpAdapterParserUtils.MAPPER);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,

59
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnectionSupport.java Normal file
View File

@@ -0,0 +1,59 @@
/*
* 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.io.PushbackInputStream;
/**
* Base class for TCP Connection Support implementations.
*
* @author Gary Russell
* @since 5.0
*
*/
public abstract class AbstractTcpConnectionSupport {
private boolean pushbackCapable;
private int pushbackBufferSize = 1;
public boolean isPushbackCapable() {
return this.pushbackCapable;
}
/**
* Set to true to cause wrapping of the connection's input stream in a
* {@link PushbackInputStream}, enabling deserializers to "unread" data.
* @param pushbackCapable true to enable.
*/
public void setPushbackCapable(boolean pushbackCapable) {
this.pushbackCapable = pushbackCapable;
}
public int getPushbackBufferSize() {
return this.pushbackBufferSize;
}
/**
* The size of the push back buffer; defaults to 1.
* @param pushbackBufferSize the size.
*/
public void setPushbackBufferSize(int pushbackBufferSize) {
this.pushbackBufferSize = pushbackBufferSize;
}
}

82
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/DefaultTcpNetConnectionSupport.java Normal file
View File

@@ -0,0 +1,82 @@
/*
* 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.io.IOException;
import java.io.InputStream;
import java.io.PushbackInputStream;
import java.net.Socket;
import org.springframework.context.ApplicationEventPublisher;
/**
* Default implementation of {@link TcpNetConnectionSupport}.
* @author Gary Russell
* @since 5.0
*
*/
public class DefaultTcpNetConnectionSupport extends AbstractTcpConnectionSupport implements TcpNetConnectionSupport {
@Override
public TcpNetConnection createNewConnection(Socket socket, boolean server, boolean lookupHost,
ApplicationEventPublisher applicationEventPublisher, String connectionFactoryName) throws Exception {
if (isPushbackCapable()) {
return new PushBackTcpNetConnection(socket, server, lookupHost, applicationEventPublisher,
connectionFactoryName, getPushbackBufferSize());
}
else {
return new TcpNetConnection(socket, server, lookupHost, applicationEventPublisher, connectionFactoryName);
}
}
private static final class PushBackTcpNetConnection extends TcpNetConnection {
private final int pushbackBufferSize;
private final String connectionId;
private volatile PushbackInputStream pushbackStream;
private volatile InputStream wrapped;
PushBackTcpNetConnection(Socket socket, boolean server, boolean lookupHost,
ApplicationEventPublisher applicationEventPublisher, String connectionFactoryName, int bufferSize) {
super(socket, server, lookupHost, applicationEventPublisher, connectionFactoryName);
this.pushbackBufferSize = bufferSize;
this.connectionId = "pushback:" + super.getConnectionId();
}
@Override
protected InputStream inputStream() throws IOException {
InputStream wrapped = super.inputStream();
// It shouldn't be possible for the wrapped stream to change but, just in case...
if (this.pushbackStream == null || wrapped != this.wrapped) {
this.pushbackStream = new PushbackInputStream(wrapped, this.pushbackBufferSize);
this.wrapped = wrapped;
}
return this.pushbackStream;
}
@Override
public String getConnectionId() {
return this.connectionId;
}
}
}

52
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/DefaultTcpNioConnectionSupport.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-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.
@@ -16,6 +16,8 @@
package org.springframework.integration.ip.tcp.connection;
import java.io.InputStream;
import java.io.PushbackInputStream;
import java.nio.channels.SocketChannel;
import org.springframework.context.ApplicationEventPublisher;
@@ -28,11 +30,55 @@ import org.springframework.context.ApplicationEventPublisher;
* @since 2.2
*
*/
public class DefaultTcpNioConnectionSupport implements TcpNioConnectionSupport {
public class DefaultTcpNioConnectionSupport extends AbstractTcpConnectionSupport implements TcpNioConnectionSupport {
@Override
public TcpNioConnection createNewConnection(SocketChannel socketChannel, boolean server, boolean lookupHost,
ApplicationEventPublisher applicationEventPublisher, String connectionFactoryName) throws Exception {
return new TcpNioConnection(socketChannel, server, lookupHost, applicationEventPublisher, connectionFactoryName);
if (isPushbackCapable()) {
return new PushBackTcpNioConnection(socketChannel, server, lookupHost, applicationEventPublisher,
connectionFactoryName, getPushbackBufferSize());
}
else {
return new TcpNioConnection(socketChannel, server, lookupHost, applicationEventPublisher,
connectionFactoryName);
}
}
private static final class PushBackTcpNioConnection extends TcpNioConnection {
private final int pushbackBufferSize;
private final String connectionId;
private volatile PushbackInputStream pushbackStream;
private volatile InputStream wrapped;
PushBackTcpNioConnection(SocketChannel socketChannel, boolean server, boolean lookupHost,
ApplicationEventPublisher applicationEventPublisher, String connectionFactoryName,
int bufferSize) throws Exception {
super(socketChannel, server, lookupHost, applicationEventPublisher, connectionFactoryName);
this.pushbackBufferSize = bufferSize;
this.connectionId = "pushback:" + super.getConnectionId();
}
@Override
protected InputStream inputStream() {
InputStream wrapped = super.inputStream();
// It shouldn't be possible for the wrapped stream to change but, just in case...
if (this.pushbackStream == null || wrapped != this.wrapped) {
this.pushbackStream = new PushbackInputStream(wrapped, this.pushbackBufferSize);
this.wrapped = wrapped;
}
return this.pushbackStream;
}
@Override
public String getConnectionId() {
return this.connectionId;
}
}
}

51
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/DefaultTcpNioSSLConnectionSupport.java
View File

@@ -17,6 +17,8 @@
package org.springframework.integration.ip.tcp.connection;
import java.io.IOException;
import java.io.InputStream;
import java.io.PushbackInputStream;
import java.nio.channels.SocketChannel;
import java.security.GeneralSecurityException;
@@ -33,7 +35,7 @@ import org.springframework.util.Assert;
* @since 2.2
*
*/
public class DefaultTcpNioSSLConnectionSupport implements TcpNioConnectionSupport {
public class DefaultTcpNioSSLConnectionSupport extends AbstractTcpConnectionSupport implements TcpNioConnectionSupport {
private volatile SSLContext sslContext;
@@ -56,8 +58,15 @@ public class DefaultTcpNioSSLConnectionSupport implements TcpNioConnectionSuppor
ApplicationEventPublisher applicationEventPublisher, String connectionFactoryName) throws Exception {
SSLEngine sslEngine = this.sslContext.createSSLEngine();
postProcessSSLEngine(sslEngine);
TcpNioSSLConnection tcpNioSSLConnection = new TcpNioSSLConnection(socketChannel, server, lookupHost,
applicationEventPublisher, connectionFactoryName, sslEngine);
TcpNioSSLConnection tcpNioSSLConnection;
if (isPushbackCapable()) {
tcpNioSSLConnection = new PushBackTcpNioSSLConnection(socketChannel, server, lookupHost,
applicationEventPublisher, connectionFactoryName, sslEngine, getPushbackBufferSize());
}
else {
tcpNioSSLConnection = new TcpNioSSLConnection(socketChannel, server, lookupHost, applicationEventPublisher,
connectionFactoryName, sslEngine);
}
tcpNioSSLConnection.init();
return tcpNioSSLConnection;
}
@@ -71,4 +80,40 @@ public class DefaultTcpNioSSLConnectionSupport implements TcpNioConnectionSuppor
// NOSONAR (empty)
}
private static final class PushBackTcpNioSSLConnection extends TcpNioSSLConnection {
private final int pushbackBufferSize;
private final String connectionId;
private volatile PushbackInputStream pushbackStream;
private volatile InputStream wrapped;
PushBackTcpNioSSLConnection(SocketChannel socketChannel, boolean server, boolean lookupHost,
ApplicationEventPublisher applicationEventPublisher, String connectionFactoryName, SSLEngine sslEngine,
int bufferSize) throws Exception {
super(socketChannel, server, lookupHost, applicationEventPublisher, connectionFactoryName, sslEngine);
this.pushbackBufferSize = bufferSize;
this.connectionId = "pushback:" + super.getConnectionId();
}
@Override
protected InputStream inputStream() {
InputStream wrapped = super.inputStream();
// It shouldn't be possible for the wrapped stream to change but, just in case...
if (this.pushbackStream == null || wrapped != this.wrapped) {
this.pushbackStream = new PushbackInputStream(wrapped, this.pushbackBufferSize);
this.wrapped = wrapped;
}
return this.pushbackStream;
}
@Override
public String getConnectionId() {
return this.connectionId;
}
}
}

20
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetClientConnectionFactory.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-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.
@@ -33,7 +33,9 @@ import org.springframework.util.Assert;
public class TcpNetClientConnectionFactory extends
AbstractClientConnectionFactory {
private volatile TcpSocketFactorySupport tcpSocketFactorySupport = new DefaultTcpNetSocketFactorySupport();
private TcpSocketFactorySupport tcpSocketFactorySupport = new DefaultTcpNetSocketFactorySupport();
private TcpNetConnectionSupport tcpNetConnectionSupport = new DefaultTcpNetConnectionSupport();
/**
* Creates a TcpNetClientConnectionFactory for connections to the host and port.
@@ -48,8 +50,8 @@ public class TcpNetClientConnectionFactory extends
protected TcpConnectionSupport buildNewConnection() throws IOException, SocketException, Exception {
Socket socket = createSocket(this.getHost(), this.getPort());
setSocketAttributes(socket);
TcpConnectionSupport connection = new TcpNetConnection(socket, false, this.isLookupHost(),
this.getApplicationEventPublisher(), this.getComponentName());
TcpConnectionSupport connection = this.tcpNetConnectionSupport.createNewConnection(socket, false, isLookupHost(),
getApplicationEventPublisher(), getComponentName());
connection = wrapConnection(connection);
initializeConnection(connection, socket);
this.getTaskExecutor().execute(connection);
@@ -57,6 +59,16 @@ public class TcpNetClientConnectionFactory extends
return connection;
}
/**
* Set the {@link TcpNetConnectionSupport} to use to create connection objects.
* @param connectionSupport the connection support.
* @since 5.0
*/
public void setTcpNetConnectionSupport(TcpNetConnectionSupport connectionSupport) {
Assert.notNull(connectionSupport, "'connectionSupport' cannot be null");
this.tcpNetConnectionSupport = connectionSupport;
}
@Override
public void start() {
this.setActive(true);

18
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java
View File

@@ -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.
@@ -17,6 +17,8 @@
package org.springframework.integration.ip.tcp.connection;
import java.io.BufferedOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
import java.net.SocketException;
@@ -118,7 +120,7 @@ public class TcpNetConnection extends TcpConnectionSupport implements Scheduling
@Override
public Object getPayload() throws Exception {
return this.getDeserializer().deserialize(this.socket.getInputStream());
return this.getDeserializer().deserialize(inputStream());
}
@Override
@@ -129,7 +131,7 @@ public class TcpNetConnection extends TcpConnectionSupport implements Scheduling
@Override
public Object getDeserializerStateKey() {
try {
return this.socket.getInputStream();
return inputStream();
}
catch (Exception e) {
return null;
@@ -146,6 +148,16 @@ public class TcpNetConnection extends TcpConnectionSupport implements Scheduling
}
}
/**
* Subclasses can override this, for example to wrap the input stream.
* @return the input stream.
* @throws IOException if an exception occurs.
* @since 5.0
*/
protected InputStream inputStream() throws IOException {
return this.socket.getInputStream();
}
/**
* If there is no listener,
* this method exits. When there is a listener, the method runs in a

51
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnectionSupport.java Normal file
View File

@@ -0,0 +1,51 @@
/*
* 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.net.Socket;
import org.springframework.context.ApplicationEventPublisher;
/**
* Used by NET connection factories to instantiate a {@link TcpNetConnection} object.
* @author Gary Russell
* @since 5.0
*
*/
@FunctionalInterface
public interface TcpNetConnectionSupport {
/**
* Create a new {@link TcpNetConnection} object wrapping the {@link Socket}.
* @param socket the Socket.
* @param server true if this connection is a server connection.
* @param lookupHost true if hostname lookup should be performed, otherwise the connection will
* be identified using the ip address.
* @param applicationEventPublisher the publisher to which OPEN, CLOSE and EXCEPTION events will
* be sent; may be null if event publishing is not required.
* @param connectionFactoryName the name of the connection factory creating this connection; used
* during event publishing, may be null, in which case "unknown" will be used.
* @return the TcpNetConnection
* @throws Exception Any exception.
*/
TcpNetConnection createNewConnection(Socket socket,
boolean server, boolean lookupHost,
ApplicationEventPublisher applicationEventPublisher,
String connectionFactoryName) throws Exception;
}

16
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java
View File

@@ -42,6 +42,8 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto
private volatile TcpSocketFactorySupport tcpSocketFactorySupport = new DefaultTcpNetSocketFactorySupport();
private TcpNetConnectionSupport tcpNetConnectionSupport = new DefaultTcpNetConnectionSupport();
/**
* Listens for incoming connections on the port.
* @param port The port.
@@ -75,6 +77,16 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto
}
}
/**
* Set the {@link TcpNetConnectionSupport} to use to create connection objects.
* @param connectionSupport the connection support.
* @since 5.0
*/
public void setTcpNetConnectionSupport(TcpNetConnectionSupport connectionSupport) {
Assert.notNull(connectionSupport, "'connectionSupport' cannot be null");
this.tcpNetConnectionSupport = connectionSupport;
}
/**
* If no listener registers, exits.
* Accepts incoming connections and creates TcpConnections for each new connection.
@@ -137,8 +149,8 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto
logger.debug("Accepted connection from " + socket.getInetAddress().getHostAddress());
}
setSocketAttributes(socket);
TcpConnectionSupport connection = new TcpNetConnection(socket, true, isLookupHost(),
getApplicationEventPublisher(), getComponentName());
TcpConnectionSupport connection = this.tcpNetConnectionSupport.createNewConnection(socket, true,
isLookupHost(), getApplicationEventPublisher(), getComponentName());
connection = wrapConnection(connection);
initializeConnection(connection, socket);
getTaskExecutor().execute(connection);

6
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-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.
@@ -82,10 +82,10 @@ public class TcpNioClientConnectionFactory extends
@Override
protected TcpConnectionSupport buildNewConnection() throws Exception {
SocketChannel socketChannel = SocketChannel.open(new InetSocketAddress(this.getHost(), this.getPort()));
SocketChannel socketChannel = SocketChannel.open(new InetSocketAddress(getHost(), getPort()));
setSocketAttributes(socketChannel.socket());
TcpNioConnection connection = this.tcpNioConnectionSupport.createNewConnection(
socketChannel, false, this.isLookupHost(), this.getApplicationEventPublisher(), this.getComponentName());
socketChannel, false, this.isLookupHost(), this.getApplicationEventPublisher(), getComponentName());
connection.setUsingDirectBuffers(this.usingDirectBuffers);
connection.setTaskExecutor(this.getTaskExecutor());
if (getSslHandshakeTimeout() != null && connection instanceof TcpNioSSLConnection) {

13
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java
View File

@@ -167,7 +167,7 @@ public class TcpNioConnection extends TcpConnectionSupport {
@Override
public Object getPayload() throws Exception {
return this.getDeserializer().deserialize(this.channelInputStream);
return this.getDeserializer().deserialize(inputStream());
}
@Override
@@ -177,7 +177,7 @@ public class TcpNioConnection extends TcpConnectionSupport {
@Override
public Object getDeserializerStateKey() {
return this.channelInputStream;
return inputStream();
}
@Override
@@ -185,6 +185,15 @@ public class TcpNioConnection extends TcpConnectionSupport {
return null;
}
/**
* Subclasses can override this, for example to wrap the input stream.
* @return the input stream.
* @since 5.0
*/
protected InputStream inputStream() {
return this.channelInputStream;
}
/**
* Allocates a ByteBuffer of the requested length using normal or
* direct buffers, depending on the usingDirectBuffers field.

3
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionSupport.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-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.
@@ -28,6 +28,7 @@ import org.springframework.context.ApplicationEventPublisher;
* @since 2.2
*
*/
@FunctionalInterface
public interface TcpNioConnectionSupport {
/**

6
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioSSLConnection.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-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.
@@ -254,12 +254,12 @@ public class TcpNioSSLConnection extends TcpNioConnection {
if (this.decoded == null) {
this.decoded = allocateEncryptionBuffer(2048);
this.encoded = allocateEncryptionBuffer(2048);
this.initilizeEngine();
initilizeEngine();
}
}
private ByteBuffer allocateEncryptionBuffer(int size) {
if (this.isUsingDirectBuffers()) {
if (isUsingDirectBuffers()) {
return ByteBuffer.allocateDirect(size);
}
else {

16
spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-5.0.xsd
View File

@@ -741,7 +741,7 @@
<xsd:annotation>
<xsd:documentation>
A reference to a TcpNioConnectionSupport strategy implementation.
When 'using-nio' is true, this is used to create connections.
When 'using-nio' is true, this is used to create connection objects.
Two default implementations are provided 'DefaultTcpNioConnectionSupport'
and 'DefaultTcpNioSSLConnectionSupport' depending on whether SSL is in
use of not.
@@ -754,6 +754,20 @@
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="net-connection-support" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
A reference to a TcpNetConnectionSupport strategy implementation.
When 'using-nio' is false, this is used to create connection objects.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type
type="org.springframework.integration.ip.tcp.connection.TcpNetConnectionSupport" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="mapper" type="xsd:string">
<xsd:annotation>
<xsd:documentation>

4
spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml
View File

@@ -92,8 +92,12 @@
apply-sequence="true"
ssl-context-support="sslContextSupport"
socket-support="socketSupport"
net-connection-support="netConnectionSupport"
socket-factory-support="socketFactorySupport" />
<bean id="netConnectionSupport"
class="org.springframework.integration.ip.tcp.connection.DefaultTcpNetConnectionSupport" />
<ip:tcp-connection-factory id="secureServerNio"
type="server"
port="#{tcpIpUtils.findAvailableServerSocket(5250)}"

5
spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java
View File

@@ -49,6 +49,7 @@ import org.springframework.integration.ip.tcp.TcpOutboundGateway;
import org.springframework.integration.ip.tcp.TcpReceivingChannelAdapter;
import org.springframework.integration.ip.tcp.TcpSendingMessageHandler;
import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory;
import org.springframework.integration.ip.tcp.connection.DefaultTcpNetConnectionSupport;
import org.springframework.integration.ip.tcp.connection.DefaultTcpNetSSLSocketFactorySupport;
import org.springframework.integration.ip.tcp.connection.DefaultTcpNioSSLConnectionSupport;
import org.springframework.integration.ip.tcp.connection.DefaultTcpSSLContextSupport;
@@ -264,6 +265,9 @@ public class ParserUnitTests {
@Autowired
TcpSocketSupport socketSupport;
@Autowired
DefaultTcpNetConnectionSupport netConnectionSupport;
@Autowired
TcpSSLContextSupport contextSupport;
@@ -657,6 +661,7 @@ public class ParserUnitTests {
assertSame(socketFactorySupport, dfa.getPropertyValue("tcpSocketFactorySupport"));
assertSame(socketSupport, dfa.getPropertyValue("tcpSocketSupport"));
assertEquals(34, TestUtils.getPropertyValue(this.secureServerNio, "sslHandshakeTimeout"));
assertSame(this.netConnectionSupport, dfa.getPropertyValue("tcpNetConnectionSupport"));
}
public static class FooAdvice extends AbstractRequestHandlerAdvice {

6
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java
View File

@@ -83,19 +83,19 @@ import org.springframework.messaging.SubscribableChannel;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.annotation.DirtiesContext.ClassMode;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Gary Russell
* @author Artem Bilan
*
* @since 2.2
*
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext(classMode = ClassMode.AFTER_EACH_TEST_METHOD)
@DirtiesContext
public class CachingClientConnectionFactoryTests {
@Autowired
@@ -742,7 +742,7 @@ public class CachingClientConnectionFactoryTests {
if (!(message instanceof ErrorMessage)) {
if (count.decrementAndGet() < 1) {
try {
Thread.sleep(1000);
Thread.sleep(100);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();

445
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/PushbackTcpTests.java Normal file
View File

@@ -0,0 +1,445 @@
/*
* 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.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import java.io.IOException;
import java.io.InputStream;
import java.io.PushbackInputStream;
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.core.serializer.Deserializer;
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.ip.tcp.serializer.ByteArrayCrLfSerializer;
import org.springframework.integration.ip.tcp.serializer.ByteArrayStxEtxSerializer;
import org.springframework.integration.ip.tcp.serializer.SoftEndOfStreamException;
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 PushbackTcpTests {
private static final String PORT = "PushbackTcpTests.port";
@Test
public void testPushbackNet() 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);
MessageChannel channel1 = client.getBean("out1", MessageChannel.class); // crlf
MessageChannel channel2 = client.getBean("out2", MessageChannel.class); // stxetx
QueueChannel replies = new QueueChannel();
Message<?> message = new GenericMessage<>("foo",
Collections.singletonMap(MessageHeaders.REPLY_CHANNEL, replies));
channel1.send(message);
channel2.send(message);
assertThat(replies.getQueueSize(), equalTo(2));
Message<?> replyA = replies.receive(0);
Message<?> replyB = replies.receive(0);
assertThat((String) replyA.getPayload(), containsString("ip_connectionId:pushback:"));
assertThat(replyB.getPayload(), not(equalTo(replyA.getPayload())));
CompositeDeserializer deserializer = server.getBean(CompositeDeserializer.class);
assertTrue(deserializer.receivedCrLf);
assertTrue(deserializer.receivedStxEtx);
System.getProperties().remove(PORT);
client.close();
server.close();
}
@Test
public void testPushbackNio() 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);
MessageChannel channel1 = client.getBean("out1", MessageChannel.class); // crlf
MessageChannel channel2 = client.getBean("out2", MessageChannel.class); // stxetx
QueueChannel replies = new QueueChannel();
Message<?> message = new GenericMessage<>("foo",
Collections.singletonMap(MessageHeaders.REPLY_CHANNEL, replies));
channel1.send(message);
channel2.send(message);
assertThat(replies.getQueueSize(), equalTo(2));
Message<?> replyA = replies.receive(0);
Message<?> replyB = replies.receive(0);
assertThat((String) replyA.getPayload(), containsString("ip_connectionId:pushback:"));
assertThat(replyB.getPayload(), not(equalTo(replyA.getPayload())));
CompositeDeserializer deserializer = server.getBean(CompositeDeserializer.class);
assertTrue(deserializer.receivedCrLf);
assertTrue(deserializer.receivedStxEtx);
System.getProperties().remove(PORT);
client.close();
server.close();
}
@Test
public void testPushbackNioSSL() throws Exception {
AnnotationConfigApplicationContext server = new AnnotationConfigApplicationContext(ServerNioSSL.class,
SSLConfig.class);
TcpNioServerConnectionFactory serverCF = server.getBean(TcpNioServerConnectionFactory.class);
int port = waitForPort(serverCF);
System.setProperty(PORT, String.valueOf(port));
AnnotationConfigApplicationContext client = new AnnotationConfigApplicationContext(ClientNioSSL.class,
SSLConfig.class);
MessageChannel channel1 = client.getBean("out1", MessageChannel.class); // crlf
MessageChannel channel2 = client.getBean("out2", MessageChannel.class); // stxetx
QueueChannel replies = new QueueChannel();
Message<?> message = new GenericMessage<>("foo",
Collections.singletonMap(MessageHeaders.REPLY_CHANNEL, replies));
channel1.send(message);
channel2.send(message);
assertThat(replies.getQueueSize(), equalTo(2));
Message<?> replyA = replies.receive(0);
Message<?> replyB = replies.receive(0);
assertThat((String) replyA.getPayload(), containsString("ip_connectionId:pushback:"));
assertThat(replyB.getPayload(), not(equalTo(replyA.getPayload())));
CompositeDeserializer deserializer = server.getBean(CompositeDeserializer.class);
assertTrue(deserializer.receivedCrLf);
assertTrue(deserializer.receivedStxEtx);
System.getProperties().remove(PORT);
client.close();
server.close();
}
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;
}
@Configuration
@EnableIntegration
public static class ServerNet {
@Bean
public TcpNetServerConnectionFactory sf() {
TcpNetServerConnectionFactory server = new TcpNetServerConnectionFactory(0);
server.setDeserializer(deserializer());
DefaultTcpNetConnectionSupport connectionSupport = new DefaultTcpNetConnectionSupport();
connectionSupport.setPushbackCapable(true);
server.setTcpNetConnectionSupport(connectionSupport);
return server;
}
@Bean
public CompositeDeserializer deserializer() {
return new CompositeDeserializer();
}
@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 cf1() {
TcpNetClientConnectionFactory cf = new TcpNetClientConnectionFactory("localhost",
Integer.parseInt(System.getProperty(PORT)));
cf.setSingleUse(true);
return cf;
}
@Bean
@ServiceActivator(inputChannel = "out1")
public TcpOutboundGateway outGate1() {
TcpOutboundGateway outGate = new TcpOutboundGateway();
outGate.setConnectionFactory(cf1());
outGate.setReplyChannelName("toString");
return outGate;
}
@Bean
public TcpNetClientConnectionFactory cf2() {
TcpNetClientConnectionFactory cf = new TcpNetClientConnectionFactory("localhost",
Integer.parseInt(System.getProperty(PORT)));
cf.setSerializer(new ByteArrayStxEtxSerializer());
return cf;
}
@Bean
@ServiceActivator(inputChannel = "out2")
public TcpOutboundGateway outGate2() {
TcpOutboundGateway outGate = new TcpOutboundGateway();
outGate.setConnectionFactory(cf2());
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() {
TcpNioServerConnectionFactory server = new TcpNioServerConnectionFactory(0);
server.setDeserializer(deserializer());
DefaultTcpNioConnectionSupport connectionSupport = new DefaultTcpNioConnectionSupport();
connectionSupport.setPushbackCapable(true);
server.setTcpNioConnectionSupport(connectionSupport);
return server;
}
@Bean
public CompositeDeserializer deserializer() {
return new CompositeDeserializer();
}
@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 cf1() {
TcpNioClientConnectionFactory cf = new TcpNioClientConnectionFactory("localhost",
Integer.parseInt(System.getProperty(PORT)));
cf.setSingleUse(true);
return cf;
}
@Bean
@ServiceActivator(inputChannel = "out1")
public TcpOutboundGateway outGate1() {
TcpOutboundGateway outGate = new TcpOutboundGateway();
outGate.setConnectionFactory(cf1());
outGate.setRemoteTimeout(50000);
outGate.setReplyChannelName("toString");
return outGate;
}
@Bean
public TcpNioClientConnectionFactory cf2() {
TcpNioClientConnectionFactory cf = new TcpNioClientConnectionFactory("localhost",
Integer.parseInt(System.getProperty(PORT)));
cf.setSerializer(new ByteArrayStxEtxSerializer());
return cf;
}
@Bean
@ServiceActivator(inputChannel = "out2")
public TcpOutboundGateway outGate2() {
TcpOutboundGateway outGate = new TcpOutboundGateway();
outGate.setConnectionFactory(cf2());
outGate.setReplyChannelName("toString");
return outGate;
}
@Bean
@Transformer(inputChannel = "toString")
public ObjectToStringTransformer otst() {
return new ObjectToStringTransformer();
}
}
@Configuration
@EnableIntegration
public static class ServerNioSSL {
@Bean
public TcpNioServerConnectionFactory sf(DefaultTcpNioSSLConnectionSupport sslNioSupport) {
TcpNioServerConnectionFactory server = new TcpNioServerConnectionFactory(0);
server.setDeserializer(deserializer());
server.setTcpNioConnectionSupport(sslNioSupport);
return server;
}
@Bean
public CompositeDeserializer deserializer() {
return new CompositeDeserializer();
}
@Bean
public TcpInboundGateway inGate(TcpNioServerConnectionFactory sf) {
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 ClientNioSSL {
@Bean
public TcpNioClientConnectionFactory cf1(DefaultTcpNioSSLConnectionSupport sslNioSupport) {
TcpNioClientConnectionFactory cf = new TcpNioClientConnectionFactory("localhost",
Integer.parseInt(System.getProperty(PORT)));
cf.setTcpNioConnectionSupport(sslNioSupport);
return cf;
}
@Bean
@ServiceActivator(inputChannel = "out1")
public TcpOutboundGateway outGate1(DefaultTcpNioSSLConnectionSupport sslNioSupport) {
TcpOutboundGateway outGate = new TcpOutboundGateway();
outGate.setConnectionFactory(cf1(sslNioSupport));
outGate.setReplyChannelName("toString");
return outGate;
}
@Bean
public TcpNioClientConnectionFactory cf2(DefaultTcpNioSSLConnectionSupport sslNioSupport) {
TcpNioClientConnectionFactory cf = new TcpNioClientConnectionFactory("localhost",
Integer.parseInt(System.getProperty(PORT)));
cf.setSerializer(new ByteArrayStxEtxSerializer());
cf.setTcpNioConnectionSupport(sslNioSupport);
return cf;
}
@Bean
@ServiceActivator(inputChannel = "out2")
public TcpOutboundGateway outGate2(DefaultTcpNioSSLConnectionSupport sslNioSupport) {
TcpOutboundGateway outGate = new TcpOutboundGateway();
outGate.setConnectionFactory(cf2(sslNioSupport));
outGate.setReplyChannelName("toString");
return outGate;
}
@Bean
@Transformer(inputChannel = "toString")
public ObjectToStringTransformer otst() {
return new ObjectToStringTransformer();
}
}
@Configuration
public static class SSLConfig {
@Bean
public DefaultTcpNioSSLConnectionSupport connectionSupport() {
DefaultTcpSSLContextSupport sslContextSupport = new DefaultTcpSSLContextSupport("test.ks",
"test.truststore.ks", "secret", "secret");
sslContextSupport.setProtocol("SSL");
DefaultTcpNioSSLConnectionSupport tcpNioConnectionSupport =
new DefaultTcpNioSSLConnectionSupport(sslContextSupport);
tcpNioConnectionSupport.setPushbackCapable(true);
return tcpNioConnectionSupport;
}
}
private static class CompositeDeserializer implements Deserializer<byte[]> {
private final ByteArrayStxEtxSerializer stxEtx = new ByteArrayStxEtxSerializer();
private final ByteArrayCrLfSerializer crlf = new ByteArrayCrLfSerializer();
private volatile boolean receivedStxEtx;
private volatile boolean receivedCrLf;
CompositeDeserializer() {
super();
}
@Override
public byte[] deserialize(InputStream inputStream) throws IOException {
PushbackInputStream pbis = (PushbackInputStream) inputStream;
int first = pbis.read();
if (first < 0) {
throw new SoftEndOfStreamException();
}
pbis.unread(first);
if (first == ByteArrayStxEtxSerializer.STX) {
this.receivedStxEtx = true;
return this.stxEtx.deserialize(pbis);
}
else {
this.receivedCrLf = true;
return this.crlf.deserialize(pbis);
}
}
}
}

31
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNetConnectionTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-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.
@@ -36,8 +36,6 @@ import org.junit.Test;
import org.mockito.Mockito;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.integration.ip.tcp.connection.TcpNioConnection.ChannelInputStream;
import org.springframework.integration.ip.tcp.serializer.ByteArrayStxEtxSerializer;
import org.springframework.integration.ip.tcp.serializer.MapJsonSerializer;
@@ -49,31 +47,20 @@ import org.springframework.messaging.support.ErrorMessage;
/**
* @author Gary Russell
* @author Artem Bilan
*
* @since 2.2.2
*
*/
public class TcpNetConnectionTests {
private final ApplicationEventPublisher nullPublisher = new ApplicationEventPublisher() {
@Override
public void publishEvent(ApplicationEvent event) {
}
@Override
public void publishEvent(Object event) {
}
};
@Test
public void testErrorLog() throws Exception {
Socket socket = mock(Socket.class);
InputStream stream = mock(InputStream.class);
when(socket.getInputStream()).thenReturn(stream);
when(stream.read()).thenReturn((int) 'x');
TcpNetConnection connection = new TcpNetConnection(socket, true, false, nullPublisher, null);
TcpNetConnection connection = new TcpNetConnection(socket, true, false, e -> { }, null);
connection.setDeserializer(new ByteArrayStxEtxSerializer());
final AtomicReference<Object> log = new AtomicReference<Object>();
Log logger = mock(Log.class);
@@ -88,8 +75,8 @@ public class TcpNetConnectionTests {
connection.run();
assertNotNull(log.get());
assertEquals("Read exception " +
connection.getConnectionId() +
" MessageMappingException:Expected STX to begin message",
connection.getConnectionId() +
" MessageMappingException:Expected STX to begin message",
log.get());
}
@@ -98,7 +85,7 @@ public class TcpNetConnectionTests {
SocketChannel socketChannel = mock(SocketChannel.class);
Socket socket = mock(Socket.class);
when(socketChannel.socket()).thenReturn(socket);
TcpNioConnection connection = new TcpNioConnection(socketChannel, true, false, nullPublisher, null);
TcpNioConnection connection = new TcpNioConnection(socketChannel, true, false, e -> { }, null);
ChannelInputStream inputStream =
TestUtils.getPropertyValue(connection, "channelInputStream", ChannelInputStream.class);
inputStream.write(ByteBuffer.wrap(new byte[] { (byte) 0x80 }));
@@ -111,7 +98,7 @@ public class TcpNetConnectionTests {
PipedInputStream pipe = new PipedInputStream();
when(inSocket.getInputStream()).thenReturn(pipe);
TcpConnectionSupport inboundConnection = new TcpNetConnection(inSocket, true, false, nullPublisher, null);
TcpConnectionSupport inboundConnection = new TcpNetConnection(inSocket, true, false, e -> { }, null);
inboundConnection.setDeserializer(new MapJsonSerializer());
MapMessageConverter inConverter = new MapMessageConverter();
MessageConvertingTcpMessageMapper inMapper = new MessageConvertingTcpMessageMapper(inConverter);
@@ -119,7 +106,7 @@ public class TcpNetConnectionTests {
ByteArrayOutputStream baos = new ByteArrayOutputStream();
Socket outSocket = mock(Socket.class);
TcpNetConnection outboundConnection = new TcpNetConnection(outSocket, true, false, nullPublisher, null);
TcpNetConnection outboundConnection = new TcpNetConnection(outSocket, true, false, e -> { }, null);
when(outSocket.getOutputStream()).thenReturn(baos);
MapMessageConverter outConverter = new MapMessageConverter();

7
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java
View File

@@ -33,6 +33,7 @@ import java.util.concurrent.atomic.AtomicReference;
import javax.net.SocketFactory;
import org.junit.Rule;
import org.junit.Test;
import org.springframework.integration.ip.tcp.serializer.AbstractByteArraySerializer;
@@ -41,6 +42,7 @@ import org.springframework.integration.ip.tcp.serializer.ByteArrayLengthHeaderSe
import org.springframework.integration.ip.tcp.serializer.ByteArrayStxEtxSerializer;
import org.springframework.integration.ip.util.SocketTestUtils;
import org.springframework.integration.ip.util.TestingUtilities;
import org.springframework.integration.test.support.LongRunningIntegrationTest;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.ErrorMessage;
@@ -52,6 +54,9 @@ import org.springframework.messaging.support.ErrorMessage;
*/
public class TcpNioConnectionReadTests {
@Rule
public LongRunningIntegrationTest longRunningIntegrationTest = new LongRunningIntegrationTest();
private final CountDownLatch latch = new CountDownLatch(1);
private AbstractServerConnectionFactory getConnectionFactory(
@@ -111,7 +116,7 @@ public class TcpNioConnectionReadTests {
AbstractServerConnectionFactory scf = getConnectionFactory(serializer, message -> {
responses.add(message);
try {
Thread.sleep(1000);
Thread.sleep(10);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();

61
src/reference/asciidoc/ip.adoc
View File

@@ -931,7 +931,7 @@ The keystore file names (first two constructor arguments) use the Spring `Resour
Starting with _version 4.3.6_, when using NIO, you can specify an `ssl-handshake-timeout` (seconds) on the connection factory.
This timeout (default 30) is used during SSL handshake when waiting for data; if the timeout is exceeded, the process is aborted and the socket closed.
[[advanced-techniques]]
[[tcp-advanced-techniques]]
=== Advanced Techniques
==== Strategy Interfaces
@@ -942,8 +942,10 @@ However, a number of strategy interfaces are provided to allow customization and
* `TcpSSLContextSupport`
* `TcpSocketFactorySupport`
* `TcpSocketSupport`
* `TcpNetConnectionSupport`
* `TcpNioConnectionSupport`
.TcpSSLContextSupport
[source,java]
----
public interface TcpSSLContextSupport {
@@ -957,6 +959,7 @@ Implementations of this interface are responsible for creating an SSLContext.
The implementation provided by the framework is the `DefaultTcpSSLContextSupport` described above.
If you require different behavior, implement this interface and provide the connection factory with a reference to a bean of your class' implementation.
.TcpSocketFactorySupport
[source,java]
----
public interface TcpSocketFactorySupport {
@@ -975,6 +978,7 @@ The second implementation is `DefaultTcpNetSSLSocketFactorySupport`; this is use
NOTE: This interface only applies if `using-nio` is "false"; socket factories are not used by NIO.
.TcpSocketSupport
[source,java]
----
public interface TcpSocketSupport {
@@ -993,6 +997,56 @@ The sole implementation provided by the framework is the `DefaultTcpSocketSuppor
To supply your own implementation of `TcpSocketFactorySupport` or `TcpSocketSupport`, provide the connection factory with references to beans of your custom type using the `socket-factory-support` and `socket-support` attributes, respectively.
.TcpNetConnectionSupport
[source, java]
----
public interface TcpNetConnectionSupport {
TcpNetConnection createNewConnection(Socket socket,
boolean server, boolean lookupHost,
ApplicationEventPublisher applicationEventPublisher,
String connectionFactoryName) throws Exception;
}
----
This interface is invoked to create `TcpNetConnection` objects (or objects from subclasses).
The framework provides a single implementation `DefatulTcpNetConnectionSupport` which creates simple `TcpNetConnection` objects by default.
It has two properties `pushbackCapable` and `pushbackBufferSize`; when push back is enabled, the implementation returns a subclass that wraps the connection's `InputStream` in a `PushbackInputStream`.
Aligned with the `PushbackInputStream` default, the buffer size defaults to 1.
This enables deserializers to "unread" (push back) bytes into the stream.
The following is a trivial example of how it might be used in a delegating deserializer which "peeks" at the first byte to determine which deserializer to invoke:
[source, java]
----
public class CompositeDeserializer implements Deserializer<byte[]> {
private final ByteArrayStxEtxSerializer stxEtx = new ByteArrayStxEtxSerializer();
private final ByteArrayCrLfSerializer crlf = new ByteArrayCrLfSerializer();
@Override
public byte[] deserialize(InputStream inputStream) throws IOException {
PushbackInputStream pbis = (PushbackInputStream) inputStream;
int first = pbis.read();
if (first < 0) {
throw new SoftEndOfStreamException();
}
pbis.unread(first);
if (first == ByteArrayStxEtxSerializer.STX) {
this.receivedStxEtx = true;
return this.stxEtx.deserialize(pbis);
}
else {
this.receivedCrLf = true;
return this.crlf.deserialize(pbis);
}
}
}
----
.TcpNioConnectionSupport
[source, java]
----
public interface TcpNioConnectionSupport {
@@ -1005,9 +1059,10 @@ public interface TcpNioConnectionSupport {
}
----
This interface is invoked to create `TcpNioConnection` objects (or subclasses).
This interface is invoked to create `TcpNioConnection` objects (or objects from subclasses).
Two implementations are provided `DefaultTcpNioSSLConnectionSupport` and `DefaultTcpNioConnectionSupport` which are used depending on whether SSL is in use or not.
A common use case would be to subclass `DefaultTcpNioSSLConnectionSupport` and override `postProcessSSLEngine`; see the example below.
As with the `DefatulTcpNetConnectionSupport`, these implementations also support push back.
==== Example: Enabling SSL Client Authentication
@@ -1221,7 +1276,7 @@ Defaults to true.
| Y
| Y
|
| See <<advanced-techniques>>
| See <<tcp-advanced-techniques>>
| read-delay
| Y
| Y

4
src/reference/asciidoc/whats-new.adoc
View File

@@ -215,5 +215,9 @@ 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.
You can now configure the TCP connection factories to support `PushbackInputStream` s, allowing deserializers to "unread" (push back) bytes after "reading ahead".
See <<tcp-advanced-techniques>> for more information.