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

INT-3654: Rework TCP Connection Close

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

Previously, single-use `TcpConnections` self-closed when their use was complete.
This is unnatural and caused issues such as INT-3722.

Remove the self-closing behavior; connection close is now (properly) the responsibility
of the client using the connection:

- Client Side:
-- `TcpOutboundGateway` after the reply is received
-- `TcpSendingMessageHandler` after the send, when there is no collaborating inbound adapter
-- `TcpReceivingChannelAdapter` when it is collaborating (after receiving the reply)

- Server Side:
-- `TcpInboundGateway` after the reply is sent
-- `TcpReceivingChannelAdapter` after the receive, when there is no collaborating outbound adapter
-- `TcpSendingMessageHandler` when it is colllaborating (after sending the reply)

As before, the `CachingClientConnectionFactory` always sets single use on the target factory
to force it to create new connections on demand. It is always a single-use factory itself
so the clients return the connections to the pool (via `close()`).

__Needs a migration guide entry__

INT-3654: Fix Late Listener Registration

Timing failures in `TcpOutboundGatewayTests.testFailoverCached()` (null listener).

The `FailoverClientConnectionFactory` registers its listener with the actual
connections it retrieves from a delegate.

When nesting failover and cached connection factories, we need to propagate the
`enableManualListenerRegistration` to the delegate factories so the connection
will wait for its listener to be registered.
This commit is contained in:
Gary Russell
2015-06-16 15:51:39 -04:00
committed by Artem Bilan
parent 1dd17ad319
commit ec5230abc7
24 changed files with 281 additions and 378 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2011 the original author or authors.
* Copyright 2002-2015 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.
@@ -61,6 +61,8 @@ public class TcpInboundGateway extends MessagingGatewaySupport implements
private volatile boolean isClientMode;
private volatile boolean isSingleUse;
private volatile long retryInterval = 60000;
private volatile ScheduledFuture<?> scheduledFuture;
@@ -75,28 +77,42 @@ public class TcpInboundGateway extends MessagingGatewaySupport implements
@Override
public boolean onMessage(Message<?> message) {
if (this.shuttingDown) {
if (logger.isInfoEnabled()) {
logger.info("Inbound message ignored; shutting down; " + message.toString());
boolean isErrorMessage = message instanceof ErrorMessage;
try {
if (this.shuttingDown) {
if (logger.isInfoEnabled()) {
logger.info("Inbound message ignored; shutting down; " + message.toString());
}
}
else {
if (isErrorMessage) {
/*
* Socket errors are sent here so they can be conveyed to any waiting thread.
* There's not one here; simply ignore.
*/
return false;
}
this.activeCount.incrementAndGet();
try {
return doOnMessage(message);
}
finally {
this.activeCount.decrementAndGet();
}
}
return false;
}
finally {
String connectionId = (String) message.getHeaders().get(IpHeaders.CONNECTION_ID);
if (connectionId != null && !isErrorMessage && this.isSingleUse) {
if (this.serverConnectionFactory != null) {
this.serverConnectionFactory.closeConnection(connectionId);
}
else {
this.clientConnectionFactory.closeConnection(connectionId);
}
}
}
else {
if (message instanceof ErrorMessage) {
/*
* Socket errors are sent here so they can be conveyed to any waiting thread.
* There's not one here; simply ignore.
*/
return false;
}
this.activeCount.incrementAndGet();
try {
return doOnMessage(message);
}
finally {
this.activeCount.decrementAndGet();
}
}
return false;
}
private boolean doOnMessage(Message<?> message) {
@@ -153,14 +169,17 @@ public class TcpInboundGateway extends MessagingGatewaySupport implements
Assert.notNull(connectionFactory, "Connection factory must not be null");
if (connectionFactory instanceof AbstractServerConnectionFactory) {
this.serverConnectionFactory = (AbstractServerConnectionFactory) connectionFactory;
} else if (connectionFactory instanceof AbstractClientConnectionFactory) {
}
else if (connectionFactory instanceof AbstractClientConnectionFactory) {
this.clientConnectionFactory = (AbstractClientConnectionFactory) connectionFactory;
} else {
}
else {
throw new IllegalArgumentException("Connection factory must be either an " +
"AbstractServerConnectionFactory or an AbstractClientConnectionFactory");
}
connectionFactory.registerListener(this);
connectionFactory.registerSender(this);
this.isSingleUse = connectionFactory.isSingleUse();
}
@Override

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

@@ -35,7 +35,6 @@ import org.springframework.integration.handler.AbstractReplyProducingMessageHand
import org.springframework.integration.ip.IpHeaders;
import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory;
import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory;
import org.springframework.integration.ip.tcp.connection.CloseDeferrable;
import org.springframework.integration.ip.tcp.connection.TcpConnection;
import org.springframework.integration.ip.tcp.connection.TcpConnectionFailedCorrelationEvent;
import org.springframework.integration.ip.tcp.connection.TcpListener;
@@ -64,6 +63,8 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler
private volatile AbstractClientConnectionFactory connectionFactory;
private volatile boolean isSingleUse;
private final Map<String, AsyncReply> pendingReplies = new ConcurrentHashMap<String, AsyncReply>();
private final Semaphore semaphore = new Semaphore(1, true);
@@ -105,10 +106,10 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler
Assert.notNull(connectionFactory, this.getClass().getName() +
" requires a client connection factory");
boolean haveSemaphore = false;
TcpConnection connection = null;
String connectionId = null;
try {
boolean singleUseConnection = this.connectionFactory.isSingleUse();
if (!singleUseConnection) {
if (!this.isSingleUse) {
logger.debug("trying semaphore");
if (!this.semaphore.tryAcquire(this.requestTimeout, TimeUnit.MILLISECONDS)) {
throw new MessageTimeoutException(requestMessage, "Timed out waiting for connection");
@@ -118,19 +119,19 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler
logger.debug("got semaphore");
}
}
TcpConnection connection = this.connectionFactory.getConnection();
connection = this.connectionFactory.getConnection();
AsyncReply reply = new AsyncReply(this.remoteTimeoutExpression.getValue(this.evaluationContext,
requestMessage, Long.class));
connectionId = connection.getConnectionId();
pendingReplies.put(connectionId, reply);
if (logger.isDebugEnabled()) {
logger.debug("Added " + connection.getConnectionId());
logger.debug("Added pending reply " + connectionId);
}
connection.send(requestMessage);
Message<?> replyMessage = reply.getReply();
if (replyMessage == null) {
if (logger.isDebugEnabled()) {
logger.debug("Remote Timeout on " + connection.getConnectionId());
logger.debug("Remote Timeout on " + connectionId);
}
// The connection is dirty - force it closed.
this.connectionFactory.forceClose(connection);
@@ -151,6 +152,12 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler
finally {
if (connectionId != null) {
pendingReplies.remove(connectionId);
if (logger.isDebugEnabled()) {
logger.debug("Removed pending reply " + connectionId);
}
if (this.isSingleUse) {
connection.close();
}
}
if (haveSemaphore) {
this.semaphore.release();
@@ -158,9 +165,6 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler
logger.debug("released semaphore");
}
}
if (this.connectionFactory instanceof CloseDeferrable) {
((CloseDeferrable) this.connectionFactory).closeDeferred(connectionId);
}
}
}
@@ -203,13 +207,11 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler
}
}
public void setConnectionFactory(AbstractConnectionFactory connectionFactory) {
// TODO: In 3.0 Change parameter type to AbstractClientConnectionFactory
Assert.isTrue(connectionFactory instanceof AbstractClientConnectionFactory,
this.getClass().getName() + " requires a client connection factory");
this.connectionFactory = (AbstractClientConnectionFactory) connectionFactory;
public void setConnectionFactory(AbstractClientConnectionFactory connectionFactory) {
this.connectionFactory = connectionFactory;
connectionFactory.registerListener(this);
connectionFactory.registerSender(this);
this.isSingleUse = connectionFactory.isSingleUse();
}
@Override
@@ -238,9 +240,6 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler
@Override
public void start() {
if (this.connectionFactory instanceof CloseDeferrable) {
((CloseDeferrable) this.connectionFactory).enableCloseDeferral(true);
}
this.connectionFactory.start();
}

77
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2011 the original author or authors.
* Copyright 2002-2015 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.
@@ -18,10 +18,9 @@ package org.springframework.integration.ip.tcp;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.atomic.AtomicInteger;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.integration.context.OrderlyShutdownCapable;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.ip.IpHeaders;
import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory;
import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory;
import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory;
@@ -29,6 +28,8 @@ import org.springframework.integration.ip.tcp.connection.ClientModeCapable;
import org.springframework.integration.ip.tcp.connection.ClientModeConnectionManager;
import org.springframework.integration.ip.tcp.connection.ConnectionFactory;
import org.springframework.integration.ip.tcp.connection.TcpListener;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.util.Assert;
/**
@@ -48,6 +49,8 @@ public class TcpReceivingChannelAdapter
private AbstractConnectionFactory serverConnectionFactory;
private volatile boolean isSingleUse;
private volatile boolean isClientMode;
private volatile long retryInterval = 60000;
@@ -62,29 +65,48 @@ public class TcpReceivingChannelAdapter
private final AtomicInteger activeCount = new AtomicInteger();
@Override
public boolean onMessage(Message<?> message) {
if (this.shuttingDown) {
if (logger.isInfoEnabled()) {
logger.info("Inbound message ignored; shutting down; " + message.toString());
boolean isErrorMessage = message instanceof ErrorMessage;
try {
if (this.shuttingDown) {
if (logger.isInfoEnabled()) {
logger.info("Inbound message ignored; shutting down; " + message.toString());
}
}
else {
if (isErrorMessage) {
/*
* Socket errors are sent here so they can be conveyed to any waiting thread.
* There's not one here; simply ignore.
*/
return false;
}
this.activeCount.incrementAndGet();
try {
sendMessage(message);
}
finally {
this.activeCount.decrementAndGet();
}
}
return false;
}
finally {
String connectionId = (String) message.getHeaders().get(IpHeaders.CONNECTION_ID);
if (connectionId != null && !isErrorMessage && this.isSingleUse) {
if (this.serverConnectionFactory != null) {
// if there's no collaborating outbound adapter, close immediately, otherwise
// it will close after sending the reply.
if (this.serverConnectionFactory.getSender() == null) {
this.serverConnectionFactory.closeConnection(connectionId);
}
}
else {
this.clientConnectionFactory.closeConnection(connectionId);
}
}
}
else {
if (message instanceof ErrorMessage) {
/*
* Socket errors are sent here so they can be conveyed to any waiting thread.
* There's not one here; simply ignore.
*/
return false;
}
this.activeCount.incrementAndGet();
try {
sendMessage(message);
}
finally {
this.activeCount.decrementAndGet();
}
}
return false;
}
@Override
@@ -148,10 +170,12 @@ public class TcpReceivingChannelAdapter
public void setConnectionFactory(AbstractConnectionFactory connectionFactory) {
if (connectionFactory instanceof AbstractClientConnectionFactory) {
this.clientConnectionFactory = connectionFactory;
} else {
}
else {
this.serverConnectionFactory = connectionFactory;
}
connectionFactory.registerListener(this);
this.isSingleUse = connectionFactory.isSingleUse();
}
public boolean isListening() {
@@ -186,6 +210,7 @@ public class TcpReceivingChannelAdapter
/**
* @return the isClientMode
*/
@Override
public boolean isClientMode() {
return this.isClientMode;
}
@@ -213,6 +238,7 @@ public class TcpReceivingChannelAdapter
this.retryInterval = retryInterval;
}
@Override
public boolean isClientModeConnected() {
if (this.isClientMode && this.clientModeConnectionManager != null) {
return this.clientModeConnectionManager.isConnected();
@@ -221,17 +247,20 @@ public class TcpReceivingChannelAdapter
}
}
@Override
public void retryConnection() {
if (this.active && this.isClientMode && this.clientModeConnectionManager != null) {
this.clientModeConnectionManager.run();
}
}
@Override
public int beforeShutdown() {
this.shuttingDown = true;
return this.activeCount.get();
}
@Override
public int afterShutdown() {
this.stop();
return this.activeCount.get();

30
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2011 the original author or authors.
* Copyright 2002-2015 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.
@@ -57,6 +57,8 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements
private volatile boolean isClientMode;
private volatile boolean isSingleUse;
private volatile long retryInterval = 60000;
private volatile ScheduledFuture<?> scheduledFuture;
@@ -109,6 +111,11 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements
throw new MessageHandlingException(message, "Error sending message", e);
}
}
finally {
if (this.isSingleUse) { // close after replying
connection.close();
}
}
}
else {
logger.error("Unable to find outbound socket for " + message);
@@ -121,8 +128,9 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements
}
else {
// we own the connection
TcpConnection connection = null;
try {
doWrite(message);
connection = doWrite(message);
}
catch (MessageHandlingException e) {
// retry - socket may have closed
@@ -130,20 +138,29 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements
if (logger.isDebugEnabled()) {
logger.debug("Fail on first write attempt", e);
}
doWrite(message);
connection = doWrite(message);
}
else {
throw e;
}
}
finally {
if (connection != null && this.isSingleUse
&& this.clientConnectionFactory.getListener() == null) {
// if there's no collaborating inbound adapter, close immediately, otherwise
// it will close after receiving the reply.
connection.close();
}
}
}
}
/**
* Method that actually does the write.
* @param message The message to write.
* @return the connection.
*/
protected void doWrite(Message<?> message) {
protected TcpConnection doWrite(Message<?> message) {
TcpConnection connection = null;
try {
connection = obtainConnection(message);
@@ -162,6 +179,7 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements
}
throw new MessageHandlingException(message, "Failed to handle message using " + connectionId, e);
}
return connection;
}
private void publishNoConnectionEvent(MessageHandlingException messageHandlingException, String connectionId) {
@@ -184,10 +202,12 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements
public void setConnectionFactory(AbstractConnectionFactory connectionFactory) {
if (connectionFactory instanceof AbstractClientConnectionFactory) {
this.clientConnectionFactory = connectionFactory;
} else {
}
else {
this.serverConnectionFactory = connectionFactory;
connectionFactory.registerSender(this);
}
this.isSingleUse = connectionFactory.isSingleUse();
}
@Override

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

@@ -155,7 +155,6 @@ public abstract class AbstractClientConnectionFactory extends AbstractConnection
connection.setMapper(this.getMapper());
connection.setDeserializer(this.getDeserializer());
connection.setSerializer(this.getSerializer());
connection.setSingleUse(this.isSingleUse());
}
/**

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 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.
@@ -29,12 +29,12 @@ import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -112,7 +112,7 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
private volatile boolean lookupHost = true;
private final List<TcpConnectionSupport> connections = new LinkedList<TcpConnectionSupport>();
private final Map<String, TcpConnectionSupport> connections = new ConcurrentHashMap<String, TcpConnectionSupport>();
private volatile TcpSocketSupport tcpSocketSupport = new DefaultTcpSocketSupport();
@@ -485,9 +485,9 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
public void stop() {
this.active = false;
synchronized (this.connections) {
Iterator<TcpConnectionSupport> iterator = this.connections.iterator();
Iterator<Entry<String, TcpConnectionSupport>> iterator = this.connections.entrySet().iterator();
while (iterator.hasNext()) {
TcpConnection connection = iterator.next();
TcpConnection connection = iterator.next().getValue();
connection.close();
iterator.remove();
}
@@ -541,7 +541,8 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
connection = wrapper;
}
return connection;
} finally {
}
finally {
this.addConnection(connection);
}
}
@@ -775,7 +776,10 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
connection.close();
return;
}
this.connections.add(connection);
this.connections.put(connection.getConnectionId(), connection);
if (logger.isDebugEnabled()) {
logger.debug(getComponentName() + ": Added new connection: " + connection.getConnectionId());
}
}
}
@@ -786,14 +790,21 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
private List<String> removeClosedConnectionsAndReturnOpenConnectionIds() {
synchronized (this.connections) {
List<String> openConnectionIds = new ArrayList<String>();
Iterator<TcpConnectionSupport> iterator = this.connections.iterator();
Iterator<Entry<String, TcpConnectionSupport>> iterator = this.connections.entrySet().iterator();
while (iterator.hasNext()) {
TcpConnection connection = iterator.next();
Entry<String, TcpConnectionSupport> entry = iterator.next();
TcpConnectionSupport connection = entry.getValue();
if (!connection.isOpen()) {
iterator.remove();
if (logger.isDebugEnabled()) {
logger.debug(getComponentName() + ": Removed closed connection: " + connection.getConnectionId());
}
}
else {
openConnectionIds.add(connection.getConnectionId());
openConnectionIds.add(entry.getKey());
if (logger.isTraceEnabled()) {
logger.trace(getComponentName() + ": Connection is open: " + connection.getConnectionId());
}
}
}
return openConnectionIds;
@@ -857,21 +868,20 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
*/
public boolean closeConnection(String connectionId) {
Assert.notNull(connectionId, "'connectionId' to close must not be null");
// closed connections are removed from #connections in #harvestClosedConnections()
synchronized(this.connections) {
boolean closed = false;
for (TcpConnectionSupport connection : connections) {
if (connectionId.equals(connection.getConnectionId())) {
try {
connection.close();
closed = true;
break;
}
catch (Exception e) {
if (logger.isDebugEnabled()) {
logger.debug("Failed to close connection " + connectionId, e);
}
connection.publishConnectionExceptionEvent(e);
TcpConnectionSupport connection = this.connections.get(connectionId);
if (connection != null) {
try {
connection.close();
closed = true;
}
catch (Exception e) {
if (logger.isDebugEnabled()) {
logger.debug("Failed to close connection " + connectionId, e);
}
connection.publishConnectionExceptionEvent(e);
}
}
return closed;

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

@@ -102,8 +102,9 @@ public abstract class AbstractServerConnectionFactory
}
/**
* Transfers attributes such as (de)serializer, singleUse etc to a new connection.
* For single use sockets, enforces a socket timeout (default 10 seconds).
* Transfers attributes such as (de)serializer, mapper etc to a new connection.
* For single use sockets, enforces a socket timeout (default 10 seconds) to prevent
* DoS attacks.
* @param connection The new connection.
* @param socket The new socket.
*/
@@ -116,7 +117,6 @@ public abstract class AbstractServerConnectionFactory
connection.setMapper(getMapper());
connection.setDeserializer(getDeserializer());
connection.setSerializer(getSerializer());
connection.setSingleUse(isSingleUse());
/*
* If we are configured
* for single use; need to enforce a timeout on the socket so we will close
@@ -126,7 +126,8 @@ public abstract class AbstractServerConnectionFactory
if (isSingleUse() && getSoTimeout() < 0) {
try {
socket.setSoTimeout(DEFAULT_REPLY_TIMEOUT);
} catch (SocketException e) {
}
catch (SocketException e) {
logger.error("Error setting default reply timeout", e);
}
}

54
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java
View File

@@ -17,10 +17,7 @@
package org.springframework.integration.ip.tcp.connection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executor;
import org.springframework.core.serializer.Deserializer;
@@ -42,19 +39,12 @@ import org.springframework.messaging.support.ErrorMessage;
* @since 2.2
*
*/
public class CachingClientConnectionFactory extends AbstractClientConnectionFactory implements CloseDeferrable {
public class CachingClientConnectionFactory extends AbstractClientConnectionFactory {
private final AbstractClientConnectionFactory targetConnectionFactory;
private final SimplePool<TcpConnectionSupport> pool;
private final Map<String, CachedConnection> deferredClosures =
new ConcurrentHashMap<String, CachedConnection>();
private final Set<String> okToRelease = new HashSet<String>();
private volatile boolean deferClose;
/**
* Construct a caching connection factory that delegates to the provided factory, with
* the provided pool size.
@@ -63,7 +53,7 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact
*/
public CachingClientConnectionFactory(AbstractClientConnectionFactory target, int poolSize) {
super("", 0);
// override single-use to true to force "close" after use
// override single-use to true so the target creates multiple connections
target.setSingleUse(true);
this.targetConnectionFactory = target;
this.pool = new SimplePool<TcpConnectionSupport>(poolSize,
@@ -145,24 +135,6 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact
return new CachedConnection(this.pool.getItem(), getListener());
}
@Override
public void enableCloseDeferral(boolean defer) {
this.deferClose = defer;
}
@Override
public void closeDeferred(String connectionId) {
synchronized(this.okToRelease) {
CachedConnection deferred = this.deferredClosures.remove(connectionId);
if (deferred != null) {
deferred.doClose();
}
else {
this.okToRelease.add(connectionId);
}
}
}
private class CachedConnection extends TcpConnectionInterceptorSupport {
private volatile boolean released;
@@ -174,17 +146,6 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact
@Override
public void close() {
synchronized(okToRelease) {
if (deferClose && !this.released && !okToRelease.remove(getConnectionId())) {
deferredClosures.put(getConnectionId(), this);
}
else {
doClose();
}
}
}
private synchronized void doClose() {
if (this.released) {
if (logger.isDebugEnabled()) {
logger.debug("Connection " + getConnectionId() + " has already been released");
@@ -254,8 +215,7 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact
logger.debug("Message discarded; no listener: " + message);
}
}
close(); // return to pool after response is received
return true; // true so the single-use connection doesn't close itself
return true;
}
private void physicallyClose() {
@@ -455,7 +415,7 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact
@Override
public boolean isSingleUse() {
return this.targetConnectionFactory.isSingleUse();
return true;
}
/**
@@ -497,6 +457,12 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact
super.forceClose(connection);
}
@Override
public void enableManualListenerRegistration() {
super.enableManualListenerRegistration();
this.targetConnectionFactory.enableManualListenerRegistration();
}
@Override
public void start() {
setActive(true);

42
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CloseDeferrable.java
View File

@@ -1,42 +0,0 @@
/*
* Copyright 2015 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;
/**
* Temporary interface on the {@code CachingClientConnectionFactory} enabling the gateway
* to defer the implicit close after onMessage so the connection is not reused until after the
* gateway has completely finished with it. Will be removed when INT-3654 is resolved, whereby
* the gateway will be completely responsible for the close.
*
* @author Gary Russell
* @since 4.1.5
*
*/
public interface CloseDeferrable {
/**
* Enable deferred closure.
* @param defer true to defer.
*/
void enableCloseDeferral(boolean defer);
/**
* Close (release) the connection if deferred.
* @param connectionId the connection id.
*/
void closeDeferred(String connectionId);
}

35
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactory.java
View File

@@ -28,7 +28,6 @@ import org.springframework.core.serializer.Serializer;
import org.springframework.integration.ip.IpHeaders;
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.util.Assert;
/**
@@ -55,6 +54,7 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac
for (AbstractClientConnectionFactory factory : factories) {
Assert.state(!(this.isSingleUse() ^ factory.isSingleUse()),
"Inconsistent singleUse - delegate factories must match this one");
factory.enableManualListenerRegistration();
}
}
@@ -82,24 +82,6 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac
@Override
public void registerListener(TcpListener listener) {
super.registerListener(listener);
for (AbstractClientConnectionFactory factory : this.factories) {
factory.registerListener(new TcpListener() {
@Override
public boolean onMessage(Message<?> message) {
if (!(message instanceof ErrorMessage)) {
throw new UnsupportedOperationException("This should never be called");
}
return false;
}
});
}
}
@Override
public void enableManualListenerRegistration() {
for (AbstractClientConnectionFactory factory : this.factories) {
factory.enableManualListenerRegistration();
}
}
@Override
@@ -117,7 +99,9 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac
return connection;
}
FailoverTcpConnection failoverTcpConnection = new FailoverTcpConnection(this.factories);
failoverTcpConnection.registerListener(this.getListener());
if (getListener() != null) {
failoverTcpConnection.registerListener(getListener());
}
failoverTcpConnection.incrementEpoch();
return failoverTcpConnection;
}
@@ -126,6 +110,7 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac
@Override
public void start() {
for (AbstractClientConnectionFactory factory : this.factories) {
factory.enableManualListenerRegistration();
factory.start();
}
this.setActive(true);
@@ -314,16 +299,6 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac
return this.connectionId + ":" + epoch;
}
@Override
public void setSingleUse(boolean singleUse) {
this.delegate.setSingleUse(singleUse);
}
@Override
public boolean isSingleUse() {
return this.delegate.isSingleUse();
}
@Override
public boolean isServer() {
return this.delegate.isServer();

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

@@ -82,12 +82,6 @@ public interface TcpConnection extends Runnable {
*/
String getConnectionId();
/**
*
* @return True if connection is used once.
*/
boolean isSingleUse();
/**
*
* @return True if connection is used once.

10
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorSupport.java
View File

@@ -101,21 +101,11 @@ public abstract class TcpConnectionInterceptorSupport extends TcpConnectionSuppo
return this.theConnection.getConnectionId();
}
@Override
public boolean isSingleUse() {
return this.theConnection.isSingleUse();
}
@Override
public void run() {
this.theConnection.run();
}
@Override
public void setSingleUse(boolean singleUse) {
this.theConnection.setSingleUse(singleUse);
}
@Override
public void setMapper(TcpMessageMapper mapper) {
this.theConnection.setMapper(mapper);

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

@@ -18,7 +18,6 @@ package org.springframework.integration.ip.tcp.connection;
import java.net.InetAddress;
import java.net.Socket;
import java.net.SocketException;
import java.util.Collections;
import java.util.Map;
import java.util.UUID;
@@ -65,20 +64,14 @@ public abstract class TcpConnectionSupport implements TcpConnection {
private volatile TcpListener listener;
private volatile TcpListener actualListener;
private volatile TcpSender sender;
private volatile boolean singleUse;
private final boolean server;
private volatile String connectionId;
private final AtomicLong sequence = new AtomicLong();
private volatile int soLinger = -1;
private volatile String hostName = "unknown";
private volatile String hostAddress = "unknown";
@@ -133,10 +126,6 @@ public abstract class TcpConnectionSupport implements TcpConnection {
int port = socket.getPort();
int localPort = socket.getLocalPort();
this.connectionId = this.hostName + ":" + port + ":" + localPort + ":" + UUID.randomUUID().toString();
try {
this.soLinger = socket.getSoLinger();
}
catch (SocketException e) { }
this.applicationEventPublisher = applicationEventPublisher;
if (connectionFactoryName != null) {
this.connectionFactoryName = connectionFactoryName;
@@ -146,21 +135,6 @@ public abstract class TcpConnectionSupport implements TcpConnection {
}
}
public void afterSend(Message<?> message) throws Exception {
if (logger.isDebugEnabled()) {
logger.debug("Message sent " + message);
}
if (this.singleUse) {
// if (we're a server socket, or a send-only socket), and soLinger <> 0, close
if ((this.isServer() || this.actualListener == null) && this.soLinger != 0) {
if (logger.isDebugEnabled()) {
logger.debug("Closing single-use connection" + this.getConnectionId());
}
this.closeConnection(false);
}
}
}
/**
* Closes this connection.
*/
@@ -259,17 +233,6 @@ public abstract class TcpConnectionSupport implements TcpConnection {
*/
public void registerListener(TcpListener listener) {
this.listener = listener;
// Determine the actual listener for this connection
if (!(this.listener instanceof TcpConnectionInterceptor)) {
this.actualListener = this.listener;
}
else {
TcpConnectionInterceptor outerInterceptor = (TcpConnectionInterceptor) this.listener;
while (outerInterceptor.getListener() instanceof TcpConnectionInterceptor) {
outerInterceptor = (TcpConnectionInterceptor) outerInterceptor.getListener();
}
this.actualListener = outerInterceptor.getListener();
}
this.listenerRegisteredLatch.countDown();
}
@@ -310,6 +273,9 @@ public abstract class TcpConnectionSupport implements TcpConnection {
@Override
public TcpListener getListener() {
if (this.manualListenerRegistration) {
if (logger.isDebugEnabled()) {
logger.debug(getConnectionId() + " Waiting for listener registration");
}
waitForListenerRegistration();
}
return this.listener;
@@ -333,23 +299,6 @@ public abstract class TcpConnectionSupport implements TcpConnection {
return sender;
}
/**
* @param singleUse true if this socket is to used once and
* discarded.
*/
public void setSingleUse(boolean singleUse) {
this.singleUse = singleUse;
}
/**
*
* @return True if connection is used once.
*/
@Override
public boolean isSingleUse() {
return this.singleUse;
}
@Override
public boolean isServer() {
return server;

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

@@ -110,7 +110,9 @@ public class TcpNetConnection extends TcpConnectionSupport implements Scheduling
this.closeConnection(true);
throw e;
}
this.afterSend(message);
if (logger.isDebugEnabled()) {
logger.debug("Message sent " + message);
}
}
@Override
@@ -144,28 +146,19 @@ public class TcpNetConnection extends TcpConnectionSupport implements Scheduling
}
/**
* If there is no listener, and this connection is not for single use,
* If there is no listener,
* this method exits. When there is a listener, the method runs in a
* loop reading input from the connection's stream, data is converted
* to an object using the {@link Deserializer} and the listener's
* {@link TcpListener#onMessage(Message)} method is called. For single use
* connections with no listener, the socket is closed after its timeout
* expires. If data is received on a single use socket with no listener,
* a warning is logged.
* {@link TcpListener#onMessage(Message)} method is called.
*/
@Override
public void run() {
boolean singleUse = this.isSingleUse();
TcpListener listener = this.getListener();
if (listener == null && !singleUse) {
logger.debug("TcpListener exiting - no listener and not single use");
return;
}
TcpListener listener = getListener();
boolean okToRun = true;
if (logger.isDebugEnabled()) {
logger.debug(this.getConnectionId() + " Reading...");
}
boolean intercepted = false;
while (okToRun) {
Message<?> message = null;
try {
@@ -184,33 +177,21 @@ public class TcpNetConnection extends TcpConnectionSupport implements Scheduling
}
try {
if (listener == null) {
logger.warn("Unexpected message - no inbound adapter registered with connection " + message);
continue;
throw new NoListenerException("No listener");
}
intercepted = this.getListener().onMessage(message);
listener.onMessage(message);
}
catch (NoListenerException nle) {
if (singleUse) {
logger.debug("Closing single use socket after inbound message " + this.getConnectionId());
this.closeConnection(true);
okToRun = false;
} else {
logger.warn("Unexpected message - no inbound adapter registered with connection " + message);
catch (NoListenerException nle) { // could also be thrown by an interceptor
if (logger.isWarnEnabled()) {
logger.warn("Unexpected message - no endpoint registered with connection interceptor: "
+ getConnectionId()
+ " - "
+ message);
}
}
catch (Exception e2) {
logger.error("Exception sending message: " + message, e2);
}
/*
* For single use sockets, we close after receipt if we are on the client
* side, and the data was not intercepted,
* or the server side has no outbound adapter registered
*/
if (singleUse && ((!this.isServer() && !intercepted) || (this.isServer() && this.getSender() == null))) {
logger.debug("Closing single use socket after inbound message " + this.getConnectionId());
this.closeConnection(false);
okToRun = false;
}
}
}
}
@@ -238,11 +219,11 @@ public class TcpNetConnection extends TcpConnectionSupport implements Scheduling
}
if (doClose) {
boolean noReadErrorOnClose = this.isNoReadErrorOnClose();
this.closeConnection(true);
closeConnection(true);
if (!(e instanceof SoftEndOfStreamException)) {
if (e instanceof SocketTimeoutException && this.isSingleUse()) {
if (e instanceof SocketTimeoutException) {
if (logger.isDebugEnabled()) {
logger.debug("Closed single use socket after timeout:" + this.getConnectionId());
logger.debug("Closed socket after timeout:" + this.getConnectionId());
}
}
else {

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

@@ -155,7 +155,9 @@ public class TcpNioConnection extends TcpConnectionSupport {
this.closeConnection(true);
throw e;
}
this.afterSend(message);
if (logger.isDebugEnabled()) {
logger.debug("Message sent " + message);
}
}
}
@@ -197,7 +199,7 @@ public class TcpNioConnection extends TcpConnectionSupport {
}
/**
* If there is no listener, and this connection is not for single use,
* If there is no listener,
* this method exits. When there is a listener, this method assembles
* data into messages by invoking convertAndSend whenever there is
* data in the input Stream. Method exits when a message is complete
@@ -212,10 +214,6 @@ public class TcpNioConnection extends TcpConnectionSupport {
boolean moreDataAvailable = true;
while(moreDataAvailable) {
try {
if (this.getListener() == null && !this.isSingleUse()) {
logger.debug("TcpListener exiting - no listener and not single use");
return;
}
try {
if (dataAvailable()) {
Message<?> message = convert();
@@ -345,11 +343,12 @@ public class TcpNioConnection extends TcpConnectionSupport {
}
catch (Exception e) {
this.closeConnection(true);
if (e instanceof SocketTimeoutException && this.isSingleUse()) {
if (e instanceof SocketTimeoutException) {
if (logger.isDebugEnabled()) {
logger.debug("Closing single use socket after timeout " + this.getConnectionId());
logger.debug("Closing socket after timeout " + this.getConnectionId());
}
} else {
}
else {
if (!(e instanceof SoftEndOfStreamException)) {
throw e;
}
@@ -360,36 +359,28 @@ public class TcpNioConnection extends TcpConnectionSupport {
}
private void sendToChannel(Message<?> message) {
boolean intercepted = false;
try {
if (message != null) {
intercepted = getListener().onMessage(message);
TcpListener listener = getListener();
if (listener == null) {
throw new NoListenerException("No listener");
}
listener.onMessage(message);
}
}
catch (Exception e) {
if (e instanceof NoListenerException) {
if (this.isSingleUse()) {
if (logger.isDebugEnabled()) {
logger.debug("Closing single use channel after inbound message " + this.getConnectionId());
}
this.closeConnection(true);
if (e instanceof NoListenerException) { // could also be thrown by an interceptor
if (logger.isWarnEnabled()) {
logger.warn("Unexpected message - no endpoint registered with connection: "
+ getConnectionId()
+ " - "
+ message);
}
}
else {
logger.error("Exception sending message: " + message, e);
}
}
/*
* For single use sockets, we close after receipt if we are on the client
* side, and the data was not intercepted,
* or the server side has no outbound adapter registered
*/
if (this.isSingleUse() && ((!this.isServer() && !intercepted) || (this.isServer() && this.getSender() == null))) {
if (logger.isDebugEnabled()) {
logger.debug("Closing single use channel after inbound message " + this.getConnectionId());
}
this.closeConnection(false);
}
}
private void doRead() throws Exception {

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

@@ -71,7 +71,6 @@ import org.springframework.integration.ip.udp.MulticastReceivingChannelAdapter;
import org.springframework.integration.ip.udp.MulticastSendingMessageHandler;
import org.springframework.integration.ip.udp.UnicastReceivingChannelAdapter;
import org.springframework.integration.ip.udp.UnicastSendingMessageHandler;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;

5
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2015 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.
@@ -32,7 +32,6 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.support.AbstractApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.messaging.Message;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.history.MessageHistory;
import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory;
@@ -46,6 +45,7 @@ import org.springframework.integration.ip.tcp.serializer.ByteArrayRawSerializer;
import org.springframework.integration.ip.util.TestingUtilities;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.Message;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -169,6 +169,7 @@ public class ConnectionToConnectionTests {
clientNet.start();
TcpConnection connection = clientNet.getConnection();
connection.send(MessageBuilder.withPayload("Test").build());
connection.close();
Message<?> message = serverSideChannel.receive(10000);
assertNotNull(message);
MessageHistory history = MessageHistory.read(message);

4
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SyslogdTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2015 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.
@@ -29,7 +29,7 @@ public class SyslogdTests {
AbstractApplicationContext ctx = new ClassPathXmlApplicationContext("SyslogdTests-context.xml", SyslogdTests.class);
System.out.println("Hit enter to terminate");
System.in.read();
ctx.destroy();
ctx.close();
}
}

19
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java
View File

@@ -19,6 +19,7 @@ package org.springframework.integration.ip.tcp;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.mockito.Mockito.doThrow;
@@ -51,6 +52,7 @@ import javax.net.ServerSocketFactory;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.hamcrest.Matchers;
import org.junit.Test;
import org.mockito.Mockito;
@@ -61,7 +63,6 @@ import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.integration.MessageTimeoutException;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory;
import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory;
import org.springframework.integration.ip.tcp.connection.CachingClientConnectionFactory;
import org.springframework.integration.ip.tcp.connection.FailoverClientConnectionFactory;
import org.springframework.integration.ip.tcp.connection.TcpConnectionSupport;
@@ -85,7 +86,7 @@ public class TcpOutboundGatewayTests {
@Test
public void testGoodNetSingle() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
AbstractClientConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
@@ -175,7 +176,7 @@ public class TcpOutboundGatewayTests {
}
}
});
AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
AbstractClientConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
ccf.setSerializer(new DefaultSerializer());
ccf.setDeserializer(new DefaultDeserializer());
ccf.setSoTimeout(10000);
@@ -230,7 +231,7 @@ public class TcpOutboundGatewayTests {
}
}
});
AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
AbstractClientConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
ccf.setSerializer(new DefaultSerializer());
ccf.setDeserializer(new DefaultDeserializer());
ccf.setSoTimeout(10000);
@@ -316,7 +317,8 @@ public class TcpOutboundGatewayTests {
* own response, not that for the first.
* @throws Exception
*/
private void testGoodNetGWTimeoutGuts(final int port, AbstractConnectionFactory ccf) throws InterruptedException {
private void testGoodNetGWTimeoutGuts(final int port, AbstractClientConnectionFactory ccf)
throws InterruptedException {
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
/*
@@ -548,6 +550,7 @@ public class TcpOutboundGatewayTests {
AbstractClientConnectionFactory factory1 = mock(AbstractClientConnectionFactory.class);
TcpConnectionSupport mockConn1 = makeMockConnection();
when(factory1.getConnection()).thenReturn(mockConn1);
when(factory1.isSingleUse()).thenReturn(true);
doThrow(new IOException("fail")).when(mockConn1).send(Mockito.any(Message.class));
CachingClientConnectionFactory cachingFactory1 = new CachingClientConnectionFactory(factory1, 1);
@@ -555,7 +558,7 @@ public class TcpOutboundGatewayTests {
factory2.setSerializer(new DefaultSerializer());
factory2.setDeserializer(new DefaultDeserializer());
factory2.setSoTimeout(10000);
factory2.setSingleUse(false);
factory2.setSingleUse(true);
CachingClientConnectionFactory cachingFactory2 = new CachingClientConnectionFactory(factory2, 1);
// Failover
@@ -563,6 +566,8 @@ public class TcpOutboundGatewayTests {
factories.add(cachingFactory1);
factories.add(cachingFactory2);
FailoverClientConnectionFactory failoverFactory = new FailoverClientConnectionFactory(factories);
failoverFactory.setSingleUse(true);
failoverFactory.afterPropertiesSet();
failoverFactory.start();
TcpOutboundGateway gateway = new TcpOutboundGateway();
@@ -693,7 +698,7 @@ public class TcpOutboundGatewayTests {
fail("expected failure");
}
catch (Exception e) {
assertTrue(e.getCause() instanceof EOFException);
assertThat(e.getCause(), Matchers.instanceOf(EOFException.class));
}
assertEquals(0, TestUtils.getPropertyValue(gateway, "pendingReplies", Map.class).size());
Message<?> reply = replyChannel.receive(0);

4
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 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.
@@ -1162,7 +1162,7 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest
Message<?> m = inbound.receive(1000);
assertNotNull(m);
assertEquals(testPayload, new String((byte[]) m.getPayload()));
ctx.destroy();
ctx.close();
}
@Test

8
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingNoSocketTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2015 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.
@@ -15,11 +15,13 @@
*/
package org.springframework.integration.ip.tcp;
import static org.junit.Assert.assertEquals;
import static org.hamcrest.Matchers.startsWith;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.fail;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandlingException;
@@ -49,7 +51,7 @@ public class TcpSendingNoSocketTests {
fail("Exception expected");
}
catch (MessageHandlingException e) {
assertEquals("Unable to find outbound socket", e.getMessage());
assertThat(e.getMessage(), startsWith("Unable to find outbound socket"));
}
}

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

@@ -594,16 +594,20 @@ public class CachingClientConnectionFactoryTests {
TcpNetClientConnectionFactory out = new TcpNetClientConnectionFactory("localhost", port);
CachingClientConnectionFactory cache = new CachingClientConnectionFactory(out, 1);
cache.setSingleUse(false);
cache.setConnectionWaitTimeout(100);
cache.start();
TcpConnectionSupport connection1 = cache.getConnection();
connection1.send(new GenericMessage<String>("foo"));
connection1.close();
TcpConnectionSupport connection2 = cache.getConnection();
connection2.send(new GenericMessage<String>("foo"));
connection2.close();
assertTrue(latch1.await(10, TimeUnit.SECONDS));
assertSame(connectionIds.get(0), connectionIds.get(1));
for (int i = 0; i < 100; i++) {
TcpConnectionSupport connection = cache.getConnection();
connection.send(new GenericMessage<String>("foo"));
connection.close();
}
assertTrue(latch2.await(10, TimeUnit.SECONDS));
assertSame(connectionIds.get(0), connectionIds.get(101));
@@ -665,6 +669,7 @@ public class CachingClientConnectionFactoryTests {
@Override
public Void answer(InvocationOnMock invocation) throws Throwable {
invocation.callRealMethod();
String log = (String) invocation.getArguments()[0];
if (log.startsWith("Response")) {
Executors.newSingleThreadScheduledExecutor().execute(new Runnable() {

25
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactoryTests.java
View File

@@ -298,13 +298,17 @@ public class FailoverClientConnectionFactoryTests {
private void testRealGuts(AbstractClientConnectionFactory client1, AbstractClientConnectionFactory client2,
AbstractServerConnectionFactory server1, AbstractServerConnectionFactory server2) throws Exception {
int port1;
int port2;
int port1 = 0;
int port2 = 0;
Executor exec = Executors.newCachedThreadPool();
client1.setTaskExecutor(exec);
client2.setTaskExecutor(exec);
server1.setTaskExecutor(exec);
server2.setTaskExecutor(exec);
client1.setBeanName("client1");
client2.setBeanName("client2");
server1.setBeanName("server1");
server2.setBeanName("server2");
ApplicationEventPublisher pub = new ApplicationEventPublisher() {
@Override
@@ -313,9 +317,9 @@ public class FailoverClientConnectionFactoryTests {
@Override
public void publishEvent(Object event) {
}
};
client1.setApplicationEventPublisher(pub);
client2.setApplicationEventPublisher(pub);
@@ -360,16 +364,21 @@ public class FailoverClientConnectionFactoryTests {
Message<String> message = new GenericMessage<String>("foo");
outGateway.setRemoteTimeout(120000);
outGateway.handleMessage(message);
Socket socket = getSocket(client1);
port1 = socket.getLocalPort();
Socket socket = null;
if (!singleUse) {
socket = getSocket(client1);
port1 = socket.getLocalPort();
}
assertTrue(singleUse | connectionId.get().contains(Integer.toString(port1)));
Message<?> replyMessage = replyChannel.receive(10000);
assertNotNull(replyMessage);
server1.stop();
TestingUtilities.waitUntilFactoryHasThisNumberOfConnections(client1, 0);
outGateway.handleMessage(message);
socket = getSocket(client2);
port2 = socket.getLocalPort();
if (!singleUse) {
socket = getSocket(client2);
port2 = socket.getLocalPort();
}
assertTrue(singleUse | connectionId.get().contains(Integer.toString(port2)));
replyMessage = replyChannel.receive(10000);
assertNotNull(replyMessage);

15
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java
View File

@@ -98,16 +98,16 @@ import org.springframework.util.ReflectionUtils.FieldFilter;
public class TcpNioConnectionTests {
private final ApplicationEventPublisher nullPublisher = new ApplicationEventPublisher() {
@Override
public void publishEvent(ApplicationEvent event) {
}
@Override
public void publishEvent(Object event) {
}
};
@Test
@@ -306,7 +306,7 @@ public class TcpNioConnectionTests {
factory.processNioSelections(1, selector, null, connections);
assertEquals(0, connections.size()); // third is closed
assertEquals(0, TestUtils.getPropertyValue(factory, "connections", List.class).size());
assertEquals(0, TestUtils.getPropertyValue(factory, "connections", Map.class).size());
}
@Test
@@ -723,9 +723,9 @@ public class TcpNioConnectionTests {
@Override
public void publishEvent(Object event) {
}
});
final CountDownLatch assemblerLatch = new CountDownLatch(1);
final AtomicReference<Thread> assembler = new AtomicReference<Thread>();
@@ -752,7 +752,8 @@ public class TcpNioConnectionTests {
Socket socket = SocketFactory.getDefault().createSocket("localhost", port);
assertTrue(connectionLatch.await(10, TimeUnit.SECONDS));
TcpNioConnection connection = (TcpNioConnection) TestUtils.getPropertyValue(factory, "connections", List.class).get(0);
TcpNioConnection connection = (TcpNioConnection) TestUtils.getPropertyValue(factory, "connections", Map.class)
.values().iterator().next();
Log logger = spy(TestUtils.getPropertyValue(connection, "logger", Log.class));
DirectFieldAccessor dfa = new DirectFieldAccessor(connection);
dfa.setPropertyValue("logger", logger);