diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java index 9ecfa93398..6744ea9649 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2014 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,6 +16,7 @@ package org.springframework.integration.endpoint; +import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.ReentrantLock; import org.springframework.context.SmartLifecycle; @@ -24,14 +25,14 @@ import org.springframework.scheduling.TaskScheduler; /** * The base class for Message Endpoint implementations. - * + * *

This class implements Lifecycle and provides an {@link #autoStartup} * property. If true, the endpoint will start automatically upon * initialization. Otherwise, it will require an explicit invocation of its * {@link #start()} method. The default value is true. * To require explicit startup, provide a value of false * to the {@link #setAutoStartup(boolean)} method. - * + * * @author Mark Fisher */ public abstract class AbstractEndpoint extends IntegrationObjectSupport implements SmartLifecycle { @@ -42,7 +43,9 @@ public abstract class AbstractEndpoint extends IntegrationObjectSupport implemen private volatile boolean running; - private final ReentrantLock lifecycleLock = new ReentrantLock(); + protected final ReentrantLock lifecycleLock = new ReentrantLock(); + + protected final Condition lifecycleCondition = this.lifecycleLock.newCondition(); public void setAutoStartup(boolean autoStartup) { 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 66b3b1b063..45ddb754f0 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 @@ -69,6 +69,8 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport impl private volatile long recoveryInterval = DEFAULT_RECOVERY_INTERVAL; + private volatile long stopTimeout = DEFAULT_RECEIVE_TIMEOUT; + private volatile boolean active; private volatile boolean listening; @@ -104,7 +106,6 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport impl * the retrieved data will be used as the payload for a new Spring Integration * Message. Otherwise, the data is deserialized as Spring Integration * Message. - * * @param expectMessage Defaults to false */ public void setExpectMessage(boolean expectMessage) { @@ -114,16 +115,12 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport impl /** * This timeout (milliseconds) is used when retrieving elements from the queue * specified by {@link #boundListOperations}. - *

- * If the queue does contain elements, the data is retrieved immediately. However, + *

If the queue does contain elements, the data is retrieved immediately. However, * if the queue is empty, the Redis connection is blocked until either an element * can be retrieved from the queue or until the specified timeout passes. - *

- * A timeout of zero can be used to block indefinitely. If not set explicitly + *

A timeout of zero can be used to block indefinitely. If not set explicitly * the timeout value will default to {@code 1000} - *

- * See also: http://redis.io/commands/brpop - * + *

See also: http://redis.io/commands/brpop * @param receiveTimeout Must be non-negative. Specified in milliseconds. */ public void setReceiveTimeout(long receiveTimeout) { @@ -131,6 +128,15 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport impl this.receiveTimeout = receiveTimeout; } + /** + * @param stopTimeout the timeout to block {@link #doStop()} until the last message will be processed + * or this timeout is reached. Should be less then or equal to {@link #receiveTimeout} + * @since 4.0.3 + */ + public void setStopTimeout(long stopTimeout) { + this.stopTimeout = stopTimeout; + } + public void setTaskExecutor(Executor taskExecutor) { this.taskExecutor = taskExecutor; } @@ -153,7 +159,8 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport impl } if (this.taskExecutor == null) { String beanName = this.getComponentName(); - this.taskExecutor = new SimpleAsyncTaskExecutor((beanName == null ? "" : beanName + "-") + this.getComponentType()); + this.taskExecutor = new SimpleAsyncTaskExecutor((beanName == null ? "" : beanName + "-") + + this.getComponentType()); } if (!(this.taskExecutor instanceof ErrorHandlingTaskExecutor) && this.getBeanFactory() != null) { MessagePublishingErrorHandler errorHandler = @@ -179,7 +186,8 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport impl catch (Exception e) { this.listening = false; if (this.active) { - logger.error("Failed to execute listening task. Will attempt to resubmit in " + this.recoveryInterval + " milliseconds.", e); + logger.error("Failed to execute listening task. Will attempt to resubmit in " + this.recoveryInterval + + " milliseconds.", e); this.publishException(e); this.sleepBeforeRecoveryAttempt(); } @@ -208,7 +216,12 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport impl } if (message != null) { - this.sendMessage(message); + if (this.listening) { + this.sendMessage(message); + } + else { + this.boundListOperations.rightPush(value); + } } } @@ -231,6 +244,7 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport impl } catch (InterruptedException e) { logger.debug("Thread interrupted while sleeping the recovery interval"); + Thread.currentThread().interrupt(); } } } @@ -252,7 +266,17 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport impl @Override protected void doStop() { - this.active = false; + try { + this.active = false; + this.lifecycleCondition.await(Math.min(this.stopTimeout, this.receiveTimeout), TimeUnit.MICROSECONDS); + } + catch (InterruptedException e) { + logger.debug("Thread interrupted while stopping the endpoint"); + Thread.currentThread().interrupt(); + } + finally { + this.listening = false; + } } public boolean isListening() { @@ -284,9 +308,9 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport impl @Override public void run() { - RedisQueueMessageDrivenEndpoint.this.listening = true; try { while (RedisQueueMessageDrivenEndpoint.this.active) { + RedisQueueMessageDrivenEndpoint.this.listening = true; RedisQueueMessageDrivenEndpoint.this.popMessageAndSend(); } } @@ -295,7 +319,13 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport impl RedisQueueMessageDrivenEndpoint.this.restart(); } else { - RedisQueueMessageDrivenEndpoint.this.listening = false; + RedisQueueMessageDrivenEndpoint.this.lifecycleLock.lock(); + try { + RedisQueueMessageDrivenEndpoint.this.lifecycleCondition.signalAll(); + } + finally { + RedisQueueMessageDrivenEndpoint.this.lifecycleLock.unlock(); + } } } } diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests.java index be47340180..384d8cd228 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests.java @@ -28,6 +28,8 @@ import java.util.Date; import java.util.List; import java.util.UUID; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import org.hamcrest.Matchers; @@ -36,6 +38,7 @@ import org.junit.Test; import org.junit.runner.RunWith; import org.mockito.Mockito; +import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.InitializingBean; @@ -45,16 +48,19 @@ import org.springframework.context.ApplicationEventPublisher; import org.springframework.data.redis.RedisConnectionFailureException; import org.springframework.data.redis.RedisSystemException; import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.core.BoundListOperations; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer; import org.springframework.data.redis.serializer.StringRedisSerializer; +import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.event.IntegrationEvent; import org.springframework.integration.redis.event.RedisExceptionEvent; import org.springframework.integration.redis.rules.RedisAvailable; import org.springframework.integration.redis.rules.RedisAvailableTests; import org.springframework.integration.support.MessageBuilder; +import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessagingException; @@ -208,6 +214,46 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { assertEquals(payload, receive.getPayload()); } + @Test + @RedisAvailable + @SuppressWarnings("unchecked") + public void testInt3442ProperlyStop() throws Exception { + final String queueName = "si.test.testInt3442ProperlyStopTest"; + + final RedisTemplate redisTemplate = new RedisTemplate(); + redisTemplate.setConnectionFactory(this.connectionFactory); + redisTemplate.setEnableDefaultSerializer(false); + redisTemplate.setKeySerializer(new StringRedisSerializer()); + redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer()); + redisTemplate.afterPropertiesSet(); + + RedisQueueMessageDrivenEndpoint endpoint = new RedisQueueMessageDrivenEndpoint(queueName, + this.connectionFactory); + BoundListOperations boundListOperations = + TestUtils.getPropertyValue(endpoint, "boundListOperations", BoundListOperations.class); + boundListOperations = Mockito.spy(boundListOperations); + new DirectFieldAccessor(endpoint).setPropertyValue("boundListOperations", boundListOperations); + endpoint.setBeanFactory(Mockito.mock(BeanFactory.class)); + endpoint.setOutputChannel(new DirectChannel()); + endpoint.setReceiveTimeout(1000); + endpoint.setStopTimeout(100); + + ExecutorService executorService = Executors.newCachedThreadPool(); + endpoint.setTaskExecutor(executorService); + + endpoint.afterPropertiesSet(); + endpoint.start(); + + redisTemplate.boundListOps(queueName).leftPush("foo"); + endpoint.stop(); + + executorService.shutdown(); + assertTrue(executorService.awaitTermination(1, TimeUnit.SECONDS)); + + Mockito.verify(boundListOperations).rightPush(Mockito.any(byte[].class)); + } + + @Test @RedisAvailable @SuppressWarnings("unchecked")