From 3e9dd90f101a9748dd145a93eeeb021b7c10ad2e Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 12 Jul 2010 19:01:18 +0000 Subject: [PATCH] INT-1147 Allow socket handling Task Executors to be injected, allowing use of specific executors like WorkManagerTaskExecutor. --- ...ternetProtocolReceivingChannelAdapter.java | 34 ++++++++++++++ ...InternetProtocolSendingMessageHandler.java | 3 -- .../ip/config/IpAdapterParserUtils.java | 2 + .../config/IpInboundChannelAdapterParser.java | 2 + .../ip/config/IpInboundGatewayParser.java | 4 ++ .../IpOutboundChannelAdapterParser.java | 2 + .../AbstractTcpReceivingChannelAdapter.java | 33 +------------ .../ip/tcp/SimpleTcpNetInboundGateway.java | 30 +++++++----- .../ip/tcp/TcpNetReceivingChannelAdapter.java | 2 +- .../ip/tcp/TcpNioReceivingChannelAdapter.java | 2 +- .../udp/UnicastReceivingChannelAdapter.java | 32 ++----------- .../ip/udp/UnicastSendingMessageHandler.java | 46 ++++++++++++------ .../ip/config/spring-integration-ip-2.0.xsd | 47 ++++++++++++++++++- .../ip/config/ParserUnitTests-context.xml | 15 ++++-- .../ip/config/ParserUnitTests.java | 12 +++++ .../DatagramPacketSendingHandlerTests.java | 1 + .../ip/udp/UdpChannelAdapterTests.java | 1 + 17 files changed, 172 insertions(+), 96 deletions(-) diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolReceivingChannelAdapter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolReceivingChannelAdapter.java index 1cd1a4c0a5..108bdc9b04 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolReceivingChannelAdapter.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolReceivingChannelAdapter.java @@ -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; + } + } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolSendingMessageHandler.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolSendingMessageHandler.java index 92ce8ee4e9..61db004aa1 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolSendingMessageHandler.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolSendingMessageHandler.java @@ -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); diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java index 937ecb5e45..8d6275e177 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java @@ -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 diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpInboundChannelAdapterParser.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpInboundChannelAdapterParser.java index 49c7fa017b..261438abca 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpInboundChannelAdapterParser.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpInboundChannelAdapterParser.java @@ -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(); } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpInboundGatewayParser.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpInboundGatewayParser.java index 9543a5166d..b1b2e92a6f 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpInboundGatewayParser.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpInboundGatewayParser.java @@ -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); } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpOutboundChannelAdapterParser.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpOutboundChannelAdapterParser.java index 933184dc40..a47d17acf5 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpOutboundChannelAdapterParser.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpOutboundChannelAdapterParser.java @@ -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; } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpReceivingChannelAdapter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpReceivingChannelAdapter.java index cfba314d1f..3e8647e3ac 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpReceivingChannelAdapter.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpReceivingChannelAdapter.java @@ -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 */ diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGateway.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGateway.java index 7218c5ebdd..29445d7f5c 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGateway.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGateway.java @@ -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(); } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetReceivingChannelAdapter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetReceivingChannelAdapter.java index 80fdb656f9..2af42e1088 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetReceivingChannelAdapter.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetReceivingChannelAdapter.java @@ -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); }}); diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioReceivingChannelAdapter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioReceivingChannelAdapter.java index 0d211f64a3..9231098a61 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioReceivingChannelAdapter.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioReceivingChannelAdapter.java @@ -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()) { diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java index 7b2a4fecbb..2825f758c9 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java @@ -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 message = null; try { diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java index c4d690c0b0..0f771e0af7 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java @@ -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; + } + } diff --git a/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd b/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd index 6939002536..84cb0f2bd5 100644 --- a/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd +++ b/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd @@ -28,7 +28,15 @@ - + + + +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. + + + @@ -48,6 +56,15 @@ next message. + + + + 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. + + + @@ -80,6 +97,15 @@ the custom message format. See java docs for TcpNetSendingChannelAdapter and Tcp + + + +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. + + + @@ -94,7 +120,15 @@ the custom message format. See java docs for TcpNetSendingChannelAdapter and Tcp - + + + +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. + + + @@ -111,6 +145,15 @@ the custom message format. See java docs for TcpNetSendingChannelAdapter and Tcp + + + +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. + + + diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml index 1be330434c..80bc8d7916 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml @@ -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"> @@ -13,6 +15,8 @@ + + message = MessageBuilder.withPayload("ABCD".getBytes()).build(); handler.handleMessage(message); Message receivedMessage = (Message) channel.receive(2000);