diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpInboundGateway.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpInboundGateway.java index 23bb903b55..6dd9379741 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpInboundGateway.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpInboundGateway.java @@ -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 diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java index a360a121e1..390a3456f8 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java @@ -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 pendingReplies = new ConcurrentHashMap(); 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(); } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapter.java index c55bb0a765..844b02ad62 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapter.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapter.java @@ -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(); diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java index 4f31826c58..e51fa12482 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java @@ -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 diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java index 28f1d68550..4d035ad3fa 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java @@ -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()); } /** diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java index fe89eaf3c8..f5dc8f0528 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java @@ -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 connections = new LinkedList(); + private final Map connections = new ConcurrentHashMap(); 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 iterator = this.connections.iterator(); + Iterator> 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 removeClosedConnectionsAndReturnOpenConnectionIds() { synchronized (this.connections) { List openConnectionIds = new ArrayList(); - Iterator iterator = this.connections.iterator(); + Iterator> iterator = this.connections.entrySet().iterator(); while (iterator.hasNext()) { - TcpConnection connection = iterator.next(); + Entry 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; diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java index 3687eaae0e..69bf863cb8 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java @@ -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); } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java index e1329a398e..759524e3a7 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java @@ -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 pool; - private final Map deferredClosures = - new ConcurrentHashMap(); - - private final Set okToRelease = new HashSet(); - - 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(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); diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CloseDeferrable.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CloseDeferrable.java deleted file mode 100644 index 75f48d2e45..0000000000 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CloseDeferrable.java +++ /dev/null @@ -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); - -} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactory.java index 1d4770897b..9d34f21731 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactory.java @@ -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(); diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnection.java index 8c97018d07..a51de10aea 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnection.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnection.java @@ -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. diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorSupport.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorSupport.java index 46df3f1024..81ec832c4e 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorSupport.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorSupport.java @@ -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); diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java index 39b82dd5c4..27616a65c4 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java @@ -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; diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java index 9bcc4eaa0b..68f56a3005 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java @@ -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 { diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java index ae01358b12..71b2172a47 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java @@ -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 { diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java index cbba44c2c8..452e8eeac3 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java @@ -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; diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests.java index 7581bac283..493b10f58c 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests.java @@ -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); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SyslogdTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SyslogdTests.java index 1c9716fcd6..95e8105e7e 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SyslogdTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SyslogdTests.java @@ -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(); } } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java index a4a7d38a1c..d522c8af86 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java @@ -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 = new AtomicReference(); @@ -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); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java index 4af2fa3163..6354ed9334 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java @@ -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 diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingNoSocketTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingNoSocketTests.java index 0e2c95a923..8ef74b1479 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingNoSocketTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingNoSocketTests.java @@ -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")); } } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java index ffbb2b2a89..d4be32edb9 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java @@ -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("foo")); + connection1.close(); TcpConnectionSupport connection2 = cache.getConnection(); connection2.send(new GenericMessage("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("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() { diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactoryTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactoryTests.java index 530a3d7a9d..5688d12902 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactoryTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactoryTests.java @@ -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 message = new GenericMessage("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); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java index b746b0d0a3..9c28c42335 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java @@ -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 assembler = new AtomicReference(); @@ -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);