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

Merge pull request #533 from garyrussell/INT-2648

Browse Source 
* INT-2648:
  INT-2648 UDP - Don't Hold a Scheduler Thread
This commit is contained in:
Oleg Zhurakousky
2012-07-10 16:33:15 -04:00
parent 62f96fa194 eb9c25615b
commit ef498d2d94
4 changed files with 14 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolReceivingChannelAdapter.java
View File

@@ -16,18 +16,15 @@
package org.springframework.integration.ip;
import java.util.Date;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.util.Assert;
/**
* Base class for inbound TCP/UDP Channel Adapters.
*
*
* @author Mark Fisher
* @author Gary Russell
* @since 2.0
@@ -59,7 +56,7 @@ public abstract class AbstractInternetProtocolReceivingChannelAdapter
}
/**
*
*
* @return The port on which this receiver is listening.
*/
public int getPort() {
@@ -99,17 +96,22 @@ public abstract class AbstractInternetProtocolReceivingChannelAdapter
return receiveBufferSize;
}
/**
* Protected by lifecycleLock
*/
@Override
protected void doStart() {
TaskScheduler taskScheduler = this.getTaskScheduler();
Assert.state(taskScheduler != null, "taskScheduler is required");
this.active = true;
taskScheduler.schedule(this, new Date());
if (!this.active) {
this.active = true;
String beanName = this.getComponentName();
checkTaskExecutor((beanName == null ? "" : beanName + "-") + this.getComponentType());
this.taskExecutor.execute(this);
}
}
/**
* Creates a default task executor if none was supplied.
*
*
* @param threadName
*/
protected void checkTaskExecutor(final String threadName) {

2
spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java
View File

@@ -77,7 +77,6 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece
if (logger.isDebugEnabled()) {
logger.debug("UDP Receiver running on port:" + this.getPort());
}
checkTaskExecutor("UDP-Incoming-Msg-Handler");
this.setListening(true);
@@ -231,6 +230,7 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece
this.mapper.setLookupHost(lookupHost);
}
@Override
public String getComponentType(){
return "ip:udp-inbound-channel-adapter";
}

10
spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/MultiClientTests.java
View File

@@ -25,7 +25,6 @@ import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.ip.util.SocketTestUtils;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.test.util.SocketUtils;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
/**
@@ -54,9 +53,6 @@ public class MultiClientTests {
adapter.setPoolSize(drivers);
QueueChannel queue = new QueueChannel(drivers * 3);
adapter.setOutputChannel(queue);
ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
taskScheduler.initialize();
adapter.setTaskScheduler(taskScheduler);
adapter.start();
final QueueChannel queueIn = new QueueChannel(1000);
SocketTestUtils.waitListening(adapter);
@@ -95,9 +91,6 @@ public class MultiClientTests {
adapter.setPoolSize(drivers);
QueueChannel queue = new QueueChannel(drivers * 3);
adapter.setOutputChannel(queue);
ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
taskScheduler.initialize();
adapter.setTaskScheduler(taskScheduler);
adapter.start();
final QueueChannel queueIn = new QueueChannel(1000);
SocketTestUtils.waitListening(adapter);
@@ -140,9 +133,6 @@ public class MultiClientTests {
adapter.setPoolSize(drivers);
QueueChannel queue = new QueueChannel(drivers * 3);
adapter.setOutputChannel(queue);
ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
taskScheduler.initialize();
adapter.setTaskScheduler(taskScheduler);
adapter.start();
final QueueChannel queueIn = new QueueChannel(1000);
SocketTestUtils.waitListening(adapter);

17
spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java
View File

@@ -19,7 +19,6 @@ import org.springframework.integration.handler.ServiceActivatingHandler;
import org.springframework.integration.ip.util.SocketTestUtils;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.test.util.SocketUtils;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
public class UdpChannelAdapterTests {
@@ -31,9 +30,6 @@ public class UdpChannelAdapterTests {
int port = SocketUtils.findAvailableUdpSocket();
UnicastReceivingChannelAdapter adapter = new UnicastReceivingChannelAdapter(port);
adapter.setOutputChannel(channel);
ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
taskScheduler.initialize();
adapter.setTaskScheduler(taskScheduler);
// SocketUtils.setLocalNicIfPossible(adapter);
adapter.start();
SocketTestUtils.waitListening(adapter);
@@ -53,10 +49,8 @@ public class UdpChannelAdapterTests {
QueueChannel channel = new QueueChannel(2);
int port = SocketUtils.findAvailableUdpSocket();
UnicastReceivingChannelAdapter adapter = new UnicastReceivingChannelAdapter(port);
adapter.setBeanName("test");
adapter.setOutputChannel(channel);
ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
taskScheduler.initialize();
adapter.setTaskScheduler(taskScheduler);
// SocketUtils.setLocalNicIfPossible(adapter);
adapter.start();
SocketTestUtils.waitListening(adapter);
@@ -82,9 +76,6 @@ public class UdpChannelAdapterTests {
int port = SocketUtils.findAvailableUdpSocket();
MulticastReceivingChannelAdapter adapter = new MulticastReceivingChannelAdapter("225.6.7.8", port);
adapter.setOutputChannel(channel);
ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
taskScheduler.initialize();
adapter.setTaskScheduler(taskScheduler);
String nic = SocketTestUtils.chooseANic(true);
if (nic == null) { // no multicast support
LogFactory.getLog(this.getClass()).error("No Multicast support");
@@ -112,9 +103,6 @@ public class UdpChannelAdapterTests {
int port = SocketUtils.findAvailableUdpSocket();
UnicastReceivingChannelAdapter adapter = new MulticastReceivingChannelAdapter("225.6.7.9", port);
adapter.setOutputChannel(channel);
ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
taskScheduler.initialize();
adapter.setTaskScheduler(taskScheduler);
String nic = SocketTestUtils.chooseANic(true);
if (nic == null) { // no multicast support
LogFactory.getLog(this.getClass()).error("No Multicast support");
@@ -140,9 +128,6 @@ public class UdpChannelAdapterTests {
int port = SocketUtils.findAvailableUdpSocket();
UnicastReceivingChannelAdapter adapter = new UnicastReceivingChannelAdapter(port);
adapter.setOutputChannel(channel);
ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
taskScheduler.initialize();
adapter.setTaskScheduler(taskScheduler);
// SocketUtils.setLocalNicIfPossible(adapter);
adapter.setOutputChannel(channel);
ServiceActivatingHandler handler = new ServiceActivatingHandler(new FailingService());