diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpInboundChannelAdapterParser.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpInboundChannelAdapterParser.java index d58c7a253a..5f07b60579 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpInboundChannelAdapterParser.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpInboundChannelAdapterParser.java @@ -25,16 +25,17 @@ import org.springframework.integration.ip.tcp.TcpReceivingChannelAdapter; import org.w3c.dom.Element; /** - * Channel Adapter that receives UDP datagram packets and maps them to Messages. - * + * Channel Adapter that receives TCP stream frames and maps them to Messages. + * * @author Gary Russell * @since 2.0 */ public class TcpInboundChannelAdapterParser extends AbstractChannelAdapterParser { - + + @Override protected AbstractBeanDefinition doParse(Element element, ParserContext parserContext, String channelName) { BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(TcpReceivingChannelAdapter.class); - IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, IpAdapterParserUtils.TCP_CONNECTION_FACTORY); builder.addPropertyReference("outputChannel", channelName); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, @@ -48,7 +49,7 @@ public class TcpInboundChannelAdapterParser extends AbstractChannelAdapterParser IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IpAdapterParserUtils.RETRY_INTERVAL); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, - IpAdapterParserUtils.SCHEDULER); + IpAdapterParserUtils.SCHEDULER, "taskScheduler"); return builder.getBeanDefinition(); } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpOutboundChannelAdapterParser.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpOutboundChannelAdapterParser.java index 79cc16475d..a99ee72b86 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpOutboundChannelAdapterParser.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpOutboundChannelAdapterParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2012 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -16,25 +16,25 @@ package org.springframework.integration.ip.config; +import org.w3c.dom.Element; + import org.springframework.beans.factory.support.AbstractBeanDefinition; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.xml.ParserContext; import org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser; import org.springframework.integration.config.xml.IntegrationNamespaceUtils; -import org.w3c.dom.Element; +import org.springframework.integration.ip.tcp.TcpSendingMessageHandler; /** * @author Gary Russell * @since 2.0 */ public class TcpOutboundChannelAdapterParser extends AbstractOutboundChannelAdapterParser { - - private static final String BASE_PACKAGE = "org.springframework.integration.ip.tcp"; + @Override protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) { - BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(BASE_PACKAGE + - ".TcpSendingMessageHandler"); - IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, + BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(TcpSendingMessageHandler.class); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, IpAdapterParserUtils.TCP_CONNECTION_FACTORY); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IpAdapterParserUtils.AUTO_STARTUP); @@ -45,7 +45,7 @@ public class TcpOutboundChannelAdapterParser extends AbstractOutboundChannelAdap IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IpAdapterParserUtils.RETRY_INTERVAL); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, - IpAdapterParserUtils.SCHEDULER); + IpAdapterParserUtils.SCHEDULER, "taskScheduler"); return builder.getBeanDefinition(); } 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 c571b06f12..31e9f98e32 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 @@ -33,7 +33,6 @@ import org.springframework.integration.ip.tcp.connection.TcpConnection; import org.springframework.integration.ip.tcp.connection.TcpListener; import org.springframework.integration.ip.tcp.connection.TcpSender; import org.springframework.scheduling.TaskScheduler; -import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.util.Assert; /** @@ -59,8 +58,6 @@ public class TcpInboundGateway extends MessagingGatewaySupport implements private volatile boolean isClientMode; - private volatile TaskScheduler scheduler; - private volatile long retryInterval = 60000; private volatile ScheduledFuture scheduledFuture; @@ -182,7 +179,8 @@ public class TcpInboundGateway extends MessagingGatewaySupport implements ClientModeConnectionManager manager = new ClientModeConnectionManager( this.clientConnectionFactory); this.clientModeConnectionManager = manager; - this.scheduledFuture = this.getScheduler().scheduleAtFixedRate(manager, this.retryInterval); + Assert.state(this.getTaskScheduler() != null, "Client mode requires a task scheduler"); + this.scheduledFuture = this.getTaskScheduler().scheduleAtFixedRate(manager, this.retryInterval); } } } @@ -221,23 +219,12 @@ public class TcpInboundGateway extends MessagingGatewaySupport implements } /** - * @return the scheduler - */ - protected TaskScheduler getScheduler() { - if (this.scheduler == null) { - ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); - scheduler.initialize(); - this.scheduler = scheduler; - } - return scheduler; - } - - /** - * @param scheduler - * the scheduler to set + * @param scheduler the scheduler to set + * @deprecated Use {@link TcpInboundGateway#setTaskScheduler(TaskScheduler)} */ + @Deprecated public void setScheduler(TaskScheduler scheduler) { - this.scheduler = scheduler; + this.setTaskScheduler(scheduler); } /** 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 e5dcc2c502..376971cc41 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 @@ -29,7 +29,6 @@ import org.springframework.integration.ip.tcp.connection.ClientModeConnectionMan import org.springframework.integration.ip.tcp.connection.ConnectionFactory; import org.springframework.integration.ip.tcp.connection.TcpListener; import org.springframework.scheduling.TaskScheduler; -import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.util.Assert; /** @@ -51,8 +50,6 @@ public class TcpReceivingChannelAdapter private volatile boolean isClientMode; - private volatile TaskScheduler scheduler; - private volatile long retryInterval = 60000; private volatile ScheduledFuture scheduledFuture; @@ -110,7 +107,8 @@ public class TcpReceivingChannelAdapter ClientModeConnectionManager manager = new ClientModeConnectionManager( this.clientConnectionFactory); this.clientModeConnectionManager = manager; - this.scheduledFuture = this.getScheduler().scheduleAtFixedRate(manager, this.retryInterval); + Assert.state(this.getTaskScheduler() != null, "Client mode requires a task scheduler"); + this.scheduledFuture = this.getTaskScheduler().scheduleAtFixedRate(manager, this.retryInterval); } } } @@ -194,23 +192,12 @@ public class TcpReceivingChannelAdapter } /** - * @return the scheduler - */ - protected TaskScheduler getScheduler() { - if (this.scheduler == null) { - ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); - scheduler.initialize(); - this.scheduler = scheduler; - } - return this.scheduler; - } - - /** - * @param scheduler - * the scheduler to set + * @param scheduler the scheduler to set + * @deprecated Use {@link #setTaskScheduler(TaskScheduler)} */ + @Deprecated public void setScheduler(TaskScheduler scheduler) { - this.scheduler = scheduler; + this.setTaskScheduler(scheduler); } /** 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 25f80bfe9e..3420c6ee15 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 @@ -34,7 +34,6 @@ import org.springframework.integration.ip.tcp.connection.TcpConnection; import org.springframework.integration.ip.tcp.connection.TcpSender; import org.springframework.integration.mapping.MessageMappingException; import org.springframework.scheduling.TaskScheduler; -import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.util.Assert; /** @@ -61,8 +60,6 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements private volatile boolean isClientMode; - private volatile TaskScheduler scheduler; - private volatile long retryInterval = 60000; private volatile ScheduledFuture scheduledFuture; @@ -216,7 +213,8 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements ClientModeConnectionManager manager = new ClientModeConnectionManager( this.clientConnectionFactory); this.clientModeConnectionManager = manager; - this.scheduledFuture = this.getScheduler().scheduleAtFixedRate(manager, this.retryInterval); + Assert.state(this.getTaskScheduler() != null, "Client mode requires a task scheduler"); + this.scheduledFuture = this.getTaskScheduler().scheduleAtFixedRate(manager, this.retryInterval); } } } @@ -314,23 +312,17 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements } /** - * @return the scheduler + * @param scheduler the scheduler to set + * @deprecated Use {@link #setTaskScheduler(TaskScheduler)} */ - protected TaskScheduler getScheduler() { - if (this.scheduler == null) { - ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); - scheduler.initialize(); - this.scheduler = scheduler; - } - return this.scheduler; + @Deprecated + public void setScheduler(TaskScheduler scheduler) { + this.setTaskScheduler(scheduler); } - /** - * @param scheduler - * the scheduler to set - */ - public void setScheduler(TaskScheduler scheduler) { - this.scheduler = scheduler; + @Override // super class is protected + public void setTaskScheduler(TaskScheduler taskScheduler) { + super.setTaskScheduler(taskScheduler); } /** diff --git a/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.2.xsd b/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.2.xsd index 5745e91552..c90c3c7ebe 100644 --- a/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.2.xsd +++ b/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.2.xsd @@ -165,7 +165,7 @@ task executors such as a WorkManagerTaskExecutor. If set to true, causes the adapter to act as a client with respect to establishing the connection, rather than listening for incoming connections. Requires a type="client" connection factory, with single-use set to false. - Defaults to true. + Defaults to false. @@ -221,7 +221,7 @@ task executors such as a WorkManagerTaskExecutor. If set to true, causes the adapter to establish a connection when started, rather than when the first message is sent. Requires a type="client" connection factory, with single-use set to false. - Defaults to true. + Defaults to false. @@ -290,7 +290,7 @@ task executors such as a WorkManagerTaskExecutor. If set to true, causes the gateway to act as a client with respect to establishing the connection, rather than listening for incoming connections. Requires a type="client" connection factory, with single-use set to false. - Defaults to true. + Defaults to false. 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 8d49ad1e16..a8615299b7 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 @@ -430,7 +430,7 @@ public class ParserUnitTests { assertEquals("ip:tcp-inbound-gateway", tcpInboundGateway2.getComponentType()); assertNull(dfa.getPropertyValue("errorChannel")); assertEquals(Boolean.FALSE, dfa.getPropertyValue("isClientMode")); - assertNull(dfa.getPropertyValue("scheduler")); + assertNull(dfa.getPropertyValue("taskScheduler")); assertEquals(60000L, dfa.getPropertyValue("retryInterval")); } @@ -542,7 +542,7 @@ public class ParserUnitTests { assertSame(client1, dfa.getPropertyValue("clientConnectionFactory")); assertEquals(25, dfa.getPropertyValue("order")); assertEquals(Boolean.FALSE, dfa.getPropertyValue("isClientMode")); - assertNull(dfa.getPropertyValue("scheduler")); + assertNull(dfa.getPropertyValue("taskScheduler")); assertEquals(60000L, dfa.getPropertyValue("retryInterval")); } @@ -559,7 +559,7 @@ public class ParserUnitTests { assertSame(client1, dfa.getPropertyValue("clientConnectionFactory")); assertNull(dfa.getPropertyValue("errorChannel")); assertEquals(Boolean.FALSE, dfa.getPropertyValue("isClientMode")); - assertNull(dfa.getPropertyValue("scheduler")); + assertNull(dfa.getPropertyValue("taskScheduler")); assertEquals(60000L, dfa.getPropertyValue("retryInterval")); } @@ -591,7 +591,7 @@ public class ParserUnitTests { assertSame(cfC3, dfa.getPropertyValue("clientConnectionFactory")); assertNull(dfa.getPropertyValue("serverConnectionFactory")); assertEquals(Boolean.TRUE, dfa.getPropertyValue("isClientMode")); - assertSame(sched, dfa.getPropertyValue("scheduler")); + assertSame(sched, dfa.getPropertyValue("taskScheduler")); assertEquals(123000L, dfa.getPropertyValue("retryInterval")); } @@ -601,7 +601,7 @@ public class ParserUnitTests { assertSame(cfC4, dfa.getPropertyValue("clientConnectionFactory")); assertNull(dfa.getPropertyValue("serverConnectionFactory")); assertEquals(Boolean.TRUE, dfa.getPropertyValue("isClientMode")); - assertSame(sched, dfa.getPropertyValue("scheduler")); + assertSame(sched, dfa.getPropertyValue("taskScheduler")); assertEquals(124000L, dfa.getPropertyValue("retryInterval")); } @@ -611,7 +611,7 @@ public class ParserUnitTests { assertSame(cfC5, dfa.getPropertyValue("clientConnectionFactory")); assertNull(dfa.getPropertyValue("serverConnectionFactory")); assertEquals(Boolean.TRUE, dfa.getPropertyValue("isClientMode")); - assertSame(sched, dfa.getPropertyValue("scheduler")); + assertSame(sched, dfa.getPropertyValue("taskScheduler")); assertEquals(125000L, dfa.getPropertyValue("retryInterval")); } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ClientModeControlBusTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ClientModeControlBusTests.java index 25663cef2f..af0107a359 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ClientModeControlBusTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ClientModeControlBusTests.java @@ -15,11 +15,15 @@ */ package org.springframework.integration.ip.tcp; -import static org.junit.Assert.*; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; +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.integration.test.util.TestUtils; +import org.springframework.scheduling.TaskScheduler; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -38,6 +42,9 @@ public class ClientModeControlBusTests { @Autowired TcpReceivingChannelAdapter tcpIn; + @Autowired + TaskScheduler taskScheduler; // default + @Test public void test() throws Exception { assertTrue(controlBus.boolResult("@tcpIn.isClientMode()")); @@ -50,6 +57,7 @@ public class ClientModeControlBusTests { } } assertTrue(controlBus.boolResult("@tcpIn.isRunning()")); + assertSame(taskScheduler, TestUtils.getPropertyValue(tcpIn, "taskScheduler")); controlBus.voidResult("@tcpIn.retryConnection()"); } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpInboundGatewayTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpInboundGatewayTests.java index b393c41624..af0fe84d69 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpInboundGatewayTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpInboundGatewayTests.java @@ -52,6 +52,7 @@ import org.springframework.integration.ip.util.TestingUtilities; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.support.channel.ChannelResolver; import org.springframework.integration.test.util.SocketUtils; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; public class TcpInboundGatewayTests { @@ -152,6 +153,10 @@ public class TcpInboundGatewayTests { } }); assertTrue(latch1.await(10, TimeUnit.SECONDS)); + ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); + taskScheduler.setPoolSize(1); + taskScheduler.initialize(); + gateway.setTaskScheduler(taskScheduler); gateway.start(); Message message = channel.receive(10000); assertNotNull(message); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java index 785979af6d..d96144f6a3 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java @@ -58,6 +58,7 @@ import org.springframework.integration.ip.tcp.connection.TcpNioServerConnectionF import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer; import org.springframework.integration.ip.util.TestingUtilities; import org.springframework.integration.test.util.SocketUtils; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; /** * @author Gary Russell @@ -126,6 +127,10 @@ public class TcpReceivingChannelAdapterTests { adapter.afterPropertiesSet(); assertTrue(latch1.await(10, TimeUnit.SECONDS)); adapter.setRetryInterval(10000); + ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); + taskScheduler.setPoolSize(1); + taskScheduler.initialize(); + adapter.setTaskScheduler(taskScheduler); adapter.start(); Message message = channel.receive(10000); assertNotNull(message); 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 f3248ad80c..78a039220b 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 @@ -65,6 +65,7 @@ import org.springframework.integration.ip.util.TestingUtilities; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.test.util.SocketUtils; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; /** * @author Gary Russell @@ -173,6 +174,10 @@ public class TcpSendingMessageHandlerTests { handler.setClientMode(true); handler.setRetryInterval(10000); handler.afterPropertiesSet(); + ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); + taskScheduler.setPoolSize(1); + taskScheduler.initialize(); + handler.setTaskScheduler(taskScheduler); handler.start(); adapter.start(); handler.handleMessage(MessageBuilder.withPayload("Test").build());