From 80a45715ac24d49b0ff09a2fcbb6ef5ab6b74193 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Fri, 27 Feb 2015 09:39:10 -0500 Subject: [PATCH] INT-3628: Use SchedulingAwareRunner for Long Tasks JIRA: https://jira.spring.io/browse/INT-3628 Certain tasks in Spring Integration are long running. Indicate so for the benefit of platforms such as WebLogic that can log warnings for long running tasks otherwise. Also fix conflicting versions of commons-io. --- build.gradle | 1 + .../tail/OSDelegatingFileTailingMessageProducer.java | 10 ++++++++-- ...bstractInternetProtocolReceivingChannelAdapter.java | 5 +++-- .../connection/AbstractServerConnectionFactory.java | 8 +++++++- .../ip/tcp/connection/TcpNetConnection.java | 10 ++++++++-- .../tcp/connection/TcpNioClientConnectionFactory.java | 10 ++++++++-- .../ip/udp/UnicastReceivingChannelAdapter.java | 7 ++++++- .../redis/inbound/RedisQueueInboundGateway.java | 8 +++++++- .../redis/inbound/RedisQueueMessageDrivenEndpoint.java | 8 +++++++- 9 files changed, 55 insertions(+), 12 deletions(-) diff --git a/build.gradle b/build.gradle index 07a4a0e295..70c76b1c9d 100644 --- a/build.gradle +++ b/build.gradle @@ -324,6 +324,7 @@ project('spring-integration-gemfire') { exclude group: 'org.springframework', module: 'spring-core' exclude group: 'org.springframework', module: 'spring-tx' } + compile "commons-io:commons-io:$commonsIoVersion" testCompile project(":spring-integration-stream") } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/tail/OSDelegatingFileTailingMessageProducer.java b/spring-integration-file/src/main/java/org/springframework/integration/file/tail/OSDelegatingFileTailingMessageProducer.java index aa0f73f1f9..9fe0e28be1 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/tail/OSDelegatingFileTailingMessageProducer.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/tail/OSDelegatingFileTailingMessageProducer.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors. + * Copyright 2002-2015 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. @@ -21,6 +21,7 @@ import java.io.InputStreamReader; import java.util.Date; import org.springframework.messaging.MessagingException; +import org.springframework.scheduling.SchedulingAwareRunnable; import org.springframework.scheduling.TaskScheduler; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.util.Assert; @@ -35,7 +36,7 @@ import org.springframework.util.Assert; * */ public class OSDelegatingFileTailingMessageProducer extends FileTailingMessageProducerSupport - implements Runnable { + implements SchedulingAwareRunnable { private volatile Process process; @@ -61,6 +62,11 @@ public class OSDelegatingFileTailingMessageProducer extends FileTailingMessagePr return super.getComponentType() + " (native)"; } + @Override + public boolean isLongLived() { + return true; + } + @Override protected void onInit() { Assert.notNull(getFile(), "File cannot be null"); 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 40b4c02520..938a655e71 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 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. @@ -22,6 +22,7 @@ import java.util.concurrent.Executors; import java.util.concurrent.ThreadFactory; import org.springframework.integration.endpoint.MessageProducerSupport; +import org.springframework.scheduling.SchedulingAwareRunnable; import org.springframework.util.Assert; /** @@ -32,7 +33,7 @@ import org.springframework.util.Assert; * @since 2.0 */ public abstract class AbstractInternetProtocolReceivingChannelAdapter - extends MessageProducerSupport implements Runnable, CommonSocketOptions { + extends MessageProducerSupport implements SchedulingAwareRunnable, CommonSocketOptions { private final int port; diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java index 408a679c70..3687eaae0e 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java @@ -21,6 +21,7 @@ import java.net.Socket; import java.net.SocketException; import org.springframework.integration.context.OrderlyShutdownCapable; +import org.springframework.scheduling.SchedulingAwareRunnable; import org.springframework.util.Assert; /** @@ -32,7 +33,7 @@ import org.springframework.util.Assert; * @since 2.0 */ public abstract class AbstractServerConnectionFactory - extends AbstractConnectionFactory implements Runnable, OrderlyShutdownCapable { + extends AbstractConnectionFactory implements SchedulingAwareRunnable, OrderlyShutdownCapable { private static final int DEFAULT_BACKLOG = 5; @@ -54,6 +55,11 @@ public abstract class AbstractServerConnectionFactory super(port); } + @Override + public boolean isLongLived() { + return true; + } + @Override public void start() { synchronized (this.lifecycleMonitor) { diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java index 512d64a5ad..265ed73f13 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java @@ -1,5 +1,5 @@ /* - * Copyright 2001-2014 the original author or authors. + * Copyright 2001-2015 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. @@ -27,6 +27,7 @@ import org.springframework.core.serializer.Deserializer; import org.springframework.core.serializer.Serializer; import org.springframework.integration.ip.tcp.serializer.SoftEndOfStreamException; import org.springframework.messaging.Message; +import org.springframework.scheduling.SchedulingAwareRunnable; /** * A TcpConnection that uses and underlying {@link Socket}. @@ -35,7 +36,7 @@ import org.springframework.messaging.Message; * @since 2.0 * */ -public class TcpNetConnection extends TcpConnectionSupport { +public class TcpNetConnection extends TcpConnectionSupport implements SchedulingAwareRunnable { private final Socket socket; @@ -63,6 +64,11 @@ public class TcpNetConnection extends TcpConnectionSupport { this.socket = socket; } + @Override + public boolean isLongLived() { + return true; + } + /** * Closes this connection. */ diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java index 43b76e4a39..ac8e026d41 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 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. @@ -30,6 +30,7 @@ import java.util.concurrent.BlockingQueue; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.LinkedBlockingQueue; +import org.springframework.scheduling.SchedulingAwareRunnable; import org.springframework.util.Assert; @@ -41,7 +42,7 @@ import org.springframework.util.Assert; * */ public class TcpNioClientConnectionFactory extends - AbstractClientConnectionFactory implements Runnable { + AbstractClientConnectionFactory implements SchedulingAwareRunnable { private volatile boolean usingDirectBuffers; @@ -113,6 +114,11 @@ public class TcpNioClientConnectionFactory extends this.tcpNioConnectionSupport = tcpNioSupport; } + @Override + public boolean isLongLived() { + return true; + } + @Override public void stop() { if (this.selector != null) { 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 1e75ea4c56..2c43671202 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 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. @@ -72,6 +72,11 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece this.mapper.setLengthCheck(lengthCheck); } + @Override + public boolean isLongLived() { + return true; + } + @Override protected void onInit() { super.onInit(); diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueInboundGateway.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueInboundGateway.java index 93cc10be0e..a811614002 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueInboundGateway.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueInboundGateway.java @@ -38,6 +38,7 @@ import org.springframework.jmx.export.annotation.ManagedOperation; import org.springframework.jmx.export.annotation.ManagedResource; import org.springframework.messaging.Message; import org.springframework.messaging.MessagingException; +import org.springframework.scheduling.SchedulingAwareRunnable; import org.springframework.util.Assert; /** @@ -323,7 +324,12 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport implements } - private class ListenerTask implements Runnable { + private class ListenerTask implements SchedulingAwareRunnable { + + @Override + public boolean isLongLived() { + return true; + } @Override public void run() { diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpoint.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpoint.java index 994f801edc..f1018d8363 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpoint.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpoint.java @@ -39,6 +39,7 @@ import org.springframework.jmx.export.annotation.ManagedResource; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessagingException; +import org.springframework.scheduling.SchedulingAwareRunnable; import org.springframework.util.Assert; /** @@ -307,7 +308,12 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport impl } - private class ListenerTask implements Runnable { + private class ListenerTask implements SchedulingAwareRunnable { + + @Override + public boolean isLongLived() { + return true; + } @Override public void run() {