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 77baf97439..5b5566f2b6 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 @@ -39,6 +39,7 @@ import org.apache.commons.logging.LogFactory; import org.springframework.context.SmartLifecycle; import org.springframework.core.serializer.Deserializer; import org.springframework.core.serializer.Serializer; +import org.springframework.integration.MessagingException; import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer; import org.springframework.util.Assert; @@ -100,6 +101,8 @@ public abstract class AbstractConnectionFactory private List connections = new LinkedList(); + protected final Object lifecycleMonitor = new Object(); + /** * Sets socket attributes on the socket. * @param socket The socket. @@ -340,19 +343,28 @@ public abstract class AbstractConnectionFactory * Starts the listening process. */ public void start() { - this.active = true; - this.getTaskExecutor().execute(this); + synchronized (this.lifecycleMonitor) { + if (!this.active) { + this.active = true; + this.getTaskExecutor().execute(this); + } + } } /** * Creates a taskExecutor (if one was not provided). */ protected Executor getTaskExecutor() { - if (this.taskExecutor == null) { - this.privateExecutor = true; - this.taskExecutor = Executors.newFixedThreadPool(this.poolSize); + synchronized (this.lifecycleMonitor) { + if (!this.active) { + throw new MessagingException("Connection Factory not started"); + } + if (this.taskExecutor == null) { + this.privateExecutor = true; + this.taskExecutor = Executors.newFixedThreadPool(this.poolSize); + } + return this.taskExecutor; } - return this.taskExecutor; } /** @@ -369,20 +381,25 @@ public abstract class AbstractConnectionFactory iterator.remove(); } } - if (this.privateExecutor) { - ExecutorService executorService = (ExecutorService) this.taskExecutor; - executorService.shutdown(); - try { - if (!executorService.awaitTermination(10, TimeUnit.SECONDS)) { - logger.debug("Forcing executor shutdown"); - executorService.shutdownNow(); + synchronized (this.lifecycleMonitor) { + if (this.privateExecutor) { + ExecutorService executorService = (ExecutorService) this.taskExecutor; + executorService.shutdown(); + try { if (!executorService.awaitTermination(10, TimeUnit.SECONDS)) { - logger.debug("Executor failed to shutdown"); + logger.debug("Forcing executor shutdown"); + executorService.shutdownNow(); + if (!executorService.awaitTermination(10, TimeUnit.SECONDS)) { + logger.debug("Executor failed to shutdown"); + } } + } catch (InterruptedException e) { + executorService.shutdownNow(); + Thread.currentThread().interrupt(); + } finally { + this.taskExecutor = null; + this.privateExecutor = false; } - } catch (InterruptedException e) { - executorService.shutdownNow(); - Thread.currentThread().interrupt(); } } } @@ -534,6 +551,10 @@ public abstract class AbstractConnectionFactory protected void addConnection(TcpConnection connection) { synchronized (this.connections) { + if (!this.active) { + connection.close(); + return; + } this.connections.add(connection); } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetClientConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetClientConnectionFactory.java index 5374492332..9ce5f63b7f 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetClientConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetClientConnectionFactory.java @@ -20,6 +20,8 @@ import java.net.Socket; import javax.net.SocketFactory; +import org.springframework.integration.MessagingException; + /** * A client connection factory that creates {@link TcpNetConnection}s. * @author Gary Russell diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/FactoryStopStartTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/FactoryStopStartTests.java new file mode 100644 index 0000000000..db2932cecc --- /dev/null +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/FactoryStopStartTests.java @@ -0,0 +1,45 @@ +/* + * Copyright 2002-2011 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; + +import org.junit.Test; +import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory; +import org.springframework.integration.ip.tcp.connection.TcpNetServerConnectionFactory; +import org.springframework.integration.ip.util.SocketTestUtils; + + +/** + * @author Gary Russell + * @since 2.0.4 + * + */ +public class FactoryStopStartTests { + + @Test + public void testRestart() { + int port = SocketTestUtils.findAvailableServerSocket(); + AbstractServerConnectionFactory factory = new TcpNetServerConnectionFactory(port); + factory.setSoTimeout(10000); + factory.start(); + factory.stop(); + factory.start(); + factory.stop(); + } + + public static void main(String[] args) { + new FactoryStopStartTests().testRestart(); + } +} 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 28b1f95741..b4b21dbdee 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 @@ -106,6 +106,7 @@ public class TcpSendingMessageHandlerTests { ccf.setSerializer(serializer); ccf.setDeserializer(serializer); ccf.setSoTimeout(10000); + ccf.start(); TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); handler.setConnectionFactory(ccf); TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();