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

INT-2599 TCP Client Mode Corrections

Browse Source 
1. Schema docs indicated client-mode is true by default, when
it is false.

2. Use IntegrationObjectSupport taskScheduler instead of
a field in the subclass.

INT-2599 polishing
This commit is contained in:
Gary Russell
2012-06-01 09:34:01 -04:00
committed by Oleg Zhurakousky
parent 6be155fbc5
commit 4be2e53d8f
11 changed files with 69 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpInboundChannelAdapterParser.java
View File

@@ -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();
}

16
spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpOutboundChannelAdapterParser.java
View File

@@ -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();
}

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

@@ -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);
}
/**

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

@@ -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);
}
/**

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

@@ -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);
}
/**

6
spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.2.xsd
View File

@@ -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.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
@@ -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.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
@@ -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.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>

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

@@ -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"));
}

10
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ClientModeControlBusTests.java
View File

@@ -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()");
}

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

@@ -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);

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

@@ -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);

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

@@ -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());