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

INT-1147 Allow socket handling Task Executors to be injected, allowing use of specific executors like WorkManagerTaskExecutor.

Browse Source
This commit is contained in:
Gary Russell
2010-07-12 19:01:18 +00:00
parent 6c779a1028
commit 3e9dd90f10
17 changed files with 172 additions and 96 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -17,6 +17,9 @@
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;
@@ -46,6 +49,10 @@ public abstract class AbstractInternetProtocolReceivingChannelAdapter
protected volatile String localAddress;
protected volatile Executor taskExecutor;
protected volatile int poolSize = 5;
public AbstractInternetProtocolReceivingChannelAdapter(int port) {
this.port = port;
@@ -91,6 +98,25 @@ public abstract class AbstractInternetProtocolReceivingChannelAdapter
taskScheduler.schedule(this, new Date());
}
/**
* Creates a default task executor if none was supplied.
*
* @param threadName
*/
protected void checkTaskExecutor(final String threadName) {
if (this.active && this.taskExecutor == null) {
Executor executor = Executors.newFixedThreadPool(this.poolSize, new ThreadFactory() {
public Thread newThread(Runnable runner) {
Thread thread = new Thread(runner);
thread.setName(threadName);
thread.setDaemon(true);
return thread;
}
});
this.taskExecutor = executor;
}
}
/* (non-Javadoc)
* @see org.springframework.integration.endpoint.AbstractEndpoint#doStop()
*/
@@ -111,4 +137,12 @@ public abstract class AbstractInternetProtocolReceivingChannelAdapter
this.localAddress = localAddress;
}
public void setPoolSize(int poolSize) {
this.poolSize = poolSize;
}
public void setTaskExecutor(Executor taskExecutor) {
this.taskExecutor = taskExecutor;
}
}

3
spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolSendingMessageHandler.java
View File

@@ -20,7 +20,6 @@ import java.net.DatagramSocket;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketAddress;
import java.util.concurrent.ExecutorService;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -47,8 +46,6 @@ public abstract class AbstractInternetProtocolSendingMessageHandler implements M
protected volatile int soTimeout = -1;
protected volatile ExecutorService executorService;
public AbstractInternetProtocolSendingMessageHandler(String host, int port) {
Assert.notNull(host, "host must not be null");
this.destinationAddress = new InetSocketAddress(host, port);

2
spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java
View File

@@ -91,6 +91,8 @@ public abstract class IpAdapterParserUtils {
static final String LOCAL_ADDRESS = "local-address";
static final String TASK_EXECUTOR = "task-executor";
/**
* Adds a constructor-arg to the provided bean definition builder

2
spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpInboundChannelAdapterParser.java
View File

@@ -53,6 +53,8 @@ public class IpInboundChannelAdapterParser extends AbstractChannelAdapterParser
IpAdapterParserUtils.POOL_SIZE);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder,
element, "channel", "outputChannel");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
IpAdapterParserUtils.TASK_EXECUTOR);
return builder.getBeanDefinition();
}

4
spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpInboundGatewayParser.java
View File

@@ -22,6 +22,7 @@ import org.w3c.dom.Element;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.core.Conventions;
import org.springframework.integration.config.xml.AbstractInboundGatewayParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.integration.ip.tcp.SimpleTcpNetInboundGateway;
/**
@@ -38,6 +39,7 @@ public class IpInboundGatewayParser extends AbstractInboundGatewayParser {
@Override
protected boolean isEligibleAttribute(String attributeName) {
return !attributeName.equals(IpAdapterParserUtils.MESSAGE_FORMAT)
&& !attributeName.equals(IpAdapterParserUtils.TASK_EXECUTOR)
&& super.isEligibleAttribute(attributeName);
}
@@ -46,6 +48,8 @@ public class IpInboundGatewayParser extends AbstractInboundGatewayParser {
builder.addPropertyValue(
Conventions.attributeNameToPropertyName(IpAdapterParserUtils.MESSAGE_FORMAT),
IpAdapterParserUtils.getMessageFormat(element));
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
IpAdapterParserUtils.TASK_EXECUTOR);
}
}

2
spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpOutboundChannelAdapterParser.java
View File

@@ -98,6 +98,8 @@ public class IpOutboundChannelAdapterParser extends AbstractOutboundChannelAdapt
}
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.RECEIVE_BUFFER_SIZE);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
IpAdapterParserUtils.TASK_EXECUTOR);
return builder;
}

33
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpReceivingChannelAdapter.java
View File

@@ -17,11 +17,8 @@ package org.springframework.integration.ip.tcp;
import java.net.Socket;
import java.net.SocketException;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;
import org.springframework.integration.ip.AbstractInternetProtocolReceivingChannelAdapter;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
/**
* Abstract class for tcp/ip incoming channel adapters. Implementations
@@ -35,10 +32,6 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
public abstract class AbstractTcpReceivingChannelAdapter extends
AbstractInternetProtocolReceivingChannelAdapter {
protected volatile ThreadPoolTaskScheduler threadPoolTaskScheduler;
protected volatile int poolSize = -1;
protected volatile SocketMessageMapper mapper = new SocketMessageMapper();
protected volatile boolean soKeepAlive;
@@ -56,29 +49,14 @@ public abstract class AbstractTcpReceivingChannelAdapter extends
}
/**
* Creates the ThreadPoolTaskScheduler, if necessary, and calls
* Checks that we have a task executor and calls
* {@link #server()}.
*/
public void run() {
if (logger.isDebugEnabled()) {
logger.debug(this.getClass().getSimpleName() + " running on port: " + port);
}
if (this.active && this.threadPoolTaskScheduler == null) {
this.threadPoolTaskScheduler = new ThreadPoolTaskScheduler();
this.threadPoolTaskScheduler.setThreadFactory(new ThreadFactory() {
private AtomicInteger n = new AtomicInteger();
public Thread newThread(Runnable runner) {
Thread thread = new Thread(runner);
thread.setName("TCP-Incoming-Msg-Handler-" + n.getAndIncrement());
thread.setDaemon(true);
return thread;
}
});
if (this.poolSize > 0) {
this.threadPoolTaskScheduler.setPoolSize(this.poolSize);
}
this.threadPoolTaskScheduler.initialize();
}
checkTaskExecutor("TCP-Incoming-Msg-Handler");
server();
}
@@ -119,13 +97,6 @@ public abstract class AbstractTcpReceivingChannelAdapter extends
mapper.setMessageFormat(messageFormat);
}
/**
* @param poolSize the poolSize to set
*/
public void setPoolSize(int poolSize) {
this.poolSize = poolSize;
}
/**
* @param close the close to set
*/

30
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGateway.java
View File

@@ -18,6 +18,7 @@ package org.springframework.integration.ip.tcp;
import java.io.IOException;
import java.net.Socket;
import java.net.SocketException;
import java.util.concurrent.Executor;
import org.springframework.integration.core.Message;
import org.springframework.integration.gateway.AbstractMessagingGateway;
@@ -67,6 +68,8 @@ public class SimpleTcpNetInboundGateway extends AbstractMessagingGateway {
protected String localAddress;
protected Executor taskExecutor;
@Override
protected void doStart() {
super.doStart();
@@ -81,18 +84,19 @@ public class SimpleTcpNetInboundGateway extends AbstractMessagingGateway {
@Override
protected void onInit() throws Exception {
this.delegate = new WriteCapableTcpNetReceivingChannelAdapter(port);
this.delegate.setMessageFormat(messageFormat);
this.delegate.setPoolSize(poolSize);
this.delegate.setReceiveBufferSize(receiveBufferSize);
this.delegate.setSoKeepAlive(soKeepAlive);
this.delegate.setSoReceiveBufferSize(soReceiveBufferSize);
this.delegate.setSoSendBufferSize(soSendBufferSize);
this.delegate.setSoTimeout(soTimeout);
this.delegate = new WriteCapableTcpNetReceivingChannelAdapter(this.port);
this.delegate.setMessageFormat(this.messageFormat);
this.delegate.setPoolSize(this.poolSize);
this.delegate.setReceiveBufferSize(this.receiveBufferSize);
this.delegate.setSoKeepAlive(this.soKeepAlive);
this.delegate.setSoReceiveBufferSize(this.soReceiveBufferSize);
this.delegate.setSoSendBufferSize(this.soSendBufferSize);
this.delegate.setSoTimeout(this.soTimeout);
this.delegate.setTaskScheduler(getTaskScheduler());
this.delegate.setCustomSocketReaderClassName(customSocketReaderClassName);
this.delegate.setClose(close);
this.delegate.setLocalAddress(localAddress);
this.delegate.setCustomSocketReaderClassName(this.customSocketReaderClassName);
this.delegate.setClose(this.close);
this.delegate.setLocalAddress(this.localAddress);
this.delegate.setTaskExecutor(this.taskExecutor);
super.onInit();
}
@@ -204,6 +208,10 @@ public class SimpleTcpNetInboundGateway extends AbstractMessagingGateway {
this.close = close;
}
public void setTaskExecutor(Executor taskExecutor) {
this.taskExecutor = taskExecutor;
}
public boolean isListening() {
return delegate.isListening();
}

2
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetReceivingChannelAdapter.java
View File

@@ -70,7 +70,7 @@ public class TcpNetReceivingChannelAdapter extends
while (true) {
final Socket socket = serverSocket.accept();
setSocketOptions(socket);
this.threadPoolTaskScheduler.execute(new Runnable() {
this.taskExecutor.execute(new Runnable() {
public void run() {
handleSocket(socket);
}});

2
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioReceivingChannelAdapter.java
View File

@@ -135,7 +135,7 @@ public class TcpNioReceivingChannelAdapter extends
}
key.attach(reader);
}
this.threadPoolTaskScheduler.execute(new Runnable() {
this.taskExecutor.execute(new Runnable() {
public void run() {
doRead(key);
if (key.channel().isOpen()) {

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

@@ -23,7 +23,6 @@ import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.util.concurrent.ThreadFactory;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
@@ -32,7 +31,6 @@ import org.springframework.integration.core.MessageHeaders;
import org.springframework.integration.core.MessagingException;
import org.springframework.integration.ip.AbstractInternetProtocolReceivingChannelAdapter;
import org.springframework.integration.ip.IpHeaders;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
/**
* A channel adapter to receive incoming UDP packets. Packets can optionally be preceded by a
@@ -48,10 +46,6 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece
protected final DatagramPacketMessageMapper mapper = new DatagramPacketMessageMapper();
protected volatile ThreadPoolTaskScheduler threadPoolTaskScheduler;
protected volatile int poolSize = -1;
protected volatile int soSendBufferSize = -1;
private static Pattern addressPattern = Pattern.compile("([^:]*):([0-9]*)");
@@ -79,29 +73,11 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece
}
public void setPoolSize(int poolSize) {
this.poolSize = poolSize;
}
public void run() {
if (logger.isDebugEnabled()) {
logger.debug("UDP Receiver running on port:" + port);
}
if (this.active && this.threadPoolTaskScheduler == null) {
this.threadPoolTaskScheduler = new ThreadPoolTaskScheduler();
this.threadPoolTaskScheduler.setThreadFactory(new ThreadFactory() {
public Thread newThread(Runnable runner) {
Thread thread = new Thread(runner);
thread.setName("UDP-Incoming-Msg-Handler");
thread.setDaemon(true);
return thread;
}
});
if (this.poolSize > 0) {
this.threadPoolTaskScheduler.setPoolSize(this.poolSize);
}
this.threadPoolTaskScheduler.initialize();
}
checkTaskExecutor("UDP-Incoming-Msg-Handler");
listening = true;
@@ -109,7 +85,7 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece
// Just schedule the packet for processing.
while (this.active) {
try {
scheduleSendMessage(receive());
asyncSendMessage(receive());
}
catch (SocketTimeoutException e) {
// continue
@@ -156,8 +132,8 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece
}
}
protected boolean scheduleSendMessage(final DatagramPacket packet) {
this.threadPoolTaskScheduler.execute(new Runnable(){
protected boolean asyncSendMessage(final DatagramPacket packet) {
this.taskExecutor.execute(new Runnable(){
public void run() {
Message<byte[]> message = null;
try {

46
spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java
View File

@@ -26,11 +26,13 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessagingException;
import org.springframework.integration.ip.AbstractInternetProtocolSendingMessageHandler;
@@ -52,7 +54,7 @@ import org.springframework.util.Assert;
* @since 2.0
*/
public class UnicastSendingMessageHandler extends
AbstractInternetProtocolSendingMessageHandler implements Runnable {
AbstractInternetProtocolSendingMessageHandler implements Runnable, InitializingBean {
protected final DatagramPacketMessageMapper mapper = new DatagramPacketMessageMapper();
@@ -84,6 +86,8 @@ public class UnicastSendingMessageHandler extends
private CountDownLatch ackLatch;
private boolean ackThreadRunning;
protected volatile Executor taskExecutor;
/**
* Basic constructor; no reliability; no acknowledgment.
@@ -160,19 +164,27 @@ public class UnicastSendingMessageHandler extends
if (ackTimeout > 0) {
this.ackTimeout = ackTimeout;
}
if (acknowledge) {
this.acknowledge = acknowledge;
if (this.acknowledge) {
Assert.hasLength(ackHost);
this.acknowledge = true;
this.executorService = Executors
.newSingleThreadExecutor(new ThreadFactory() {
private AtomicInteger n = new AtomicInteger();
public Thread newThread(Runnable runner) {
Thread thread = new Thread(runner);
thread.setName("UDP-Ack-Handler-" + n.getAndIncrement());
thread.setDaemon(true);
return thread;
}
});
}
}
public void afterPropertiesSet() {
if (this.acknowledge) {
if (this.taskExecutor == null) {
Executor executor = Executors
.newSingleThreadExecutor(new ThreadFactory() {
private AtomicInteger n = new AtomicInteger();
public Thread newThread(Runnable runner) {
Thread thread = new Thread(runner);
thread.setName("UDP-Ack-Handler-" + n.getAndIncrement());
thread.setDaemon(true);
return thread;
}
});
this.taskExecutor = executor;
}
}
}
@@ -184,7 +196,7 @@ public class UnicastSendingMessageHandler extends
synchronized(this) {
if (!this.ackThreadRunning) {
ackLatch = new CountDownLatch(1);
this.executorService.execute(this);
this.taskExecutor.execute(this);
try {
ackLatch.await(10000, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) { }
@@ -306,7 +318,7 @@ public class UnicastSendingMessageHandler extends
return;
}
this.fatalException = null;
this.executorService.execute(this);
this.taskExecutor.execute(this);
}
public void shutDown() {
@@ -328,5 +340,9 @@ public class UnicastSendingMessageHandler extends
this.localAddress = localAddress;
}
public void setTaskExecutor(Executor taskExecutor) {
this.taskExecutor = taskExecutor;
}
}

47
spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd
View File

@@ -28,7 +28,15 @@
<xsd:complexType>
<xsd:complexContent>
<xsd:extension base="ipAdapterType">
<xsd:attribute name="pool-size" type="xsd:string" />
<xsd:attribute name="pool-size" type="xsd:string" >
<xsd:annotation>
<xsd:documentation>
The number of threads that will be used for socket/channel handling. Only applies
if an external task-executor is NOT being used. When using an external task executor,
its configuration specifies the number of threads.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="receive-buffer-size" type="xsd:string" />
<xsd:attribute name="multicast-address" type="xsd:string" />
<xsd:attribute name="custom-socket-reader-class-name" type="xsd:string" >
@@ -48,6 +56,15 @@ next message.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="task-executor" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Specifies a specific Executor to be used for socket handling. If not supplied, an internal
pooled executor will be used (See pool-size). Needed on some platforms that require the use of specific
task executors such as a WorkManagerTaskExecutor.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
@@ -80,6 +97,15 @@ the custom message format. See java docs for TcpNetSendingChannelAdapter and Tcp
<xsd:attribute name="so-linger" type="xsd:string" />
<xsd:attribute name="so-tcp-no-delay" type="xsd:string" />
<xsd:attribute name="so-traffic-class" type="xsd:string" />
<xsd:attribute name="task-executor" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Specifies a specific Executor to be used for handling acknowledgments in the UDP adapter. If not supplied, an internal
pooled executor will be used. Needed on some platforms that require the use of specific
task executors such as a WorkManagerTaskExecutor.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
@@ -94,7 +120,15 @@ the custom message format. See java docs for TcpNetSendingChannelAdapter and Tcp
<xsd:complexType>
<xsd:complexContent>
<xsd:extension base="gatewayType">
<xsd:attribute name="pool-size" type="xsd:string" />
<xsd:attribute name="pool-size" type="xsd:string" >
<xsd:annotation>
<xsd:documentation>
The number of threads that will be used for socket/channel handling. Only applies
if an external task-executor is NOT being used. When using an external task executor,
its configuration specifies the number of threads.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="custom-socket-reader-class-name" type="xsd:string" >
<xsd:annotation>
<xsd:documentation>
@@ -111,6 +145,15 @@ the custom message format. See java docs for TcpNetSendingChannelAdapter and Tcp
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="task-executor" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Specifies a specific Executor to be used for socket handling. If not supplied, an internal
pooled executor will be used (See pool-size). Needed on some platforms that require the use of specific
task executors such as a WorkManagerTaskExecutor.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>

15
spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml
View File

@@ -3,9 +3,11 @@
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:ip="http://www.springframework.org/schema/integration/ip"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/integration/ip http://www.springframework.org/schema/integration/ip/spring-integration-ip.xsd">
xmlns:task="http://www.springframework.org/schema/task"
xsi:schemaLocation="http://www.springframework.org/schema/integration/ip http://www.springframework.org/schema/integration/ip/spring-integration-ip.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.0.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd">
<bean id="tcpIpUtils" class="org.springframework.integration.ip.util.SocketUtils" />
@@ -13,6 +15,8 @@
<int:channel id="tcpChannel" />
<int:channel id="replyChannel" />
<task:executor id="externalTE" pool-size="10"/>
<ip:inbound-channel-adapter id="testInUdp"
channel="udpChannel"
check-length="true"
@@ -27,6 +31,7 @@
so-send-buffer-size="31"
so-timeout="32"
local-address="127.0.0.1"
task-executor="externalTE"
/>
<ip:inbound-channel-adapter id="testInUdpMulticast"
@@ -122,7 +127,8 @@
so-receive-buffer-size="52"
so-send-buffer-size="53"
so-timeout="54"
local-address="127.0.0.1"
local-address="127.0.0.1"
task-executor="externalTE"
/>
<ip:outbound-channel-adapter id="testOutUdpiMulticast"
@@ -221,6 +227,7 @@
so-send-buffer-size="125"
so-timeout="126"
local-address="127.0.0.1"
task-executor="externalTE"
/>
<ip:outbound-gateway id="simpleOutGateway"

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

@@ -16,6 +16,8 @@
package org.springframework.integration.ip.config;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotSame;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import org.junit.Test;
@@ -24,6 +26,7 @@ import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.ApplicationContext;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.ip.tcp.CustomNetSocketReader;
import org.springframework.integration.ip.tcp.CustomNetSocketWriter;
import org.springframework.integration.ip.tcp.CustomNioSocketReader;
@@ -118,6 +121,10 @@ public class ParserUnitTests {
@Qualifier(value="org.springframework.integration.ip.tcp.SimpleTcpNetOutboundGateway#1")
SimpleTcpNetOutboundGateway simpleTcpNetOutboundGatewayClose;
@Autowired
@Qualifier(value="externalTE")
TaskExecutor taskExecutor;
@Test
public void testInUdp() {
DirectFieldAccessor dfa = new DirectFieldAccessor(udpIn);
@@ -128,6 +135,7 @@ public class ParserUnitTests {
assertEquals(31, dfa.getPropertyValue("soSendBufferSize"));
assertEquals(32, dfa.getPropertyValue("soTimeout"));
assertEquals("127.0.0.1", dfa.getPropertyValue("localAddress"));
assertSame(taskExecutor, dfa.getPropertyValue("taskExecutor"));
}
@Test
@@ -141,6 +149,7 @@ public class ParserUnitTests {
assertEquals(31, dfa.getPropertyValue("soSendBufferSize"));
assertEquals(32, dfa.getPropertyValue("soTimeout"));
assertEquals("127.0.0.1", dfa.getPropertyValue("localAddress"));
assertNotSame(taskExecutor, dfa.getPropertyValue("taskExecutor"));
}
@Test
@@ -220,6 +229,7 @@ public class ParserUnitTests {
assertEquals(53, dfa.getPropertyValue("soSendBufferSize"));
assertEquals(54, dfa.getPropertyValue("soTimeout"));
assertEquals("127.0.0.1", dfa.getPropertyValue("localAddress"));
assertSame(taskExecutor, dfa.getPropertyValue("taskExecutor"));
}
@Test
@@ -317,6 +327,8 @@ public class ParserUnitTests {
assertEquals(23, dfa.getPropertyValue("poolSize"));
assertEquals(false, dfa.getPropertyValue("close"));
assertEquals("127.0.0.1", dfa.getPropertyValue("localAddress"));
assertSame(taskExecutor, dfa.getPropertyValue("taskExecutor"));
assertSame(taskExecutor, delegateDfa.getPropertyValue("taskExecutor"));
}
@Test

1
spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketSendingHandlerTests.java
View File

@@ -95,6 +95,7 @@ public class DatagramPacketSendingHandlerTests {
UnicastSendingMessageHandler handler =
new UnicastSendingMessageHandler("localhost", testPort, true,
true, "localhost", ackPort, 5000);
handler.afterPropertiesSet();
Executors.newSingleThreadExecutor().execute(new Runnable() {
public void run() {
try {

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

@@ -64,6 +64,7 @@ public class UdpChannelAdapterTests {
// whichNic,
SocketUtils.findAvailableUdpSocket(), 5000);
// handler.setLocalAddress(whichNic);
handler.afterPropertiesSet();
Message<byte[]> message = MessageBuilder.withPayload("ABCD".getBytes()).build();
handler.handleMessage(message);
Message<byte[]> receivedMessage = (Message<byte[]>) channel.receive(2000);