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

INT-3442 RedisQMDE Delay stop Until Last Read

Browse Source 
JIRA: https://jira.spring.io/browse/INT-3442

Previously `RedisQueueMessageDrivenEndpoint.stop()` returned immediately.
It caused an issue when one more `message` might be read and processed to stopped app.

* Introduce `AbstractEndpoint#lifecycleCondition` and wait on it from `RedisQueueMessageDrivenEndpoint.doStop()`
and `signal()` it from `ListenerTask`.
* Since everything is done around `lifecycleLock` the `RedisQueueMessageDrivenEndpoint.stop()` waits for the proper
'last' message process.
* Add tiny `Thread.sleep(1)` to the `popMessageAndSend` cycle to free `lifecycleLock` for other Threads, e.g. `stop()`

INT-3442: Get rid of `lock` from listener cycle

* Introduce `stopTimeout` to minimize the `stop` thread blocking
* If the message is returned after that timeout it moved back to Redis List using `RPUSH`

INT-3442: Addressing PR comments
This commit is contained in:
Artem Bilan
2014-06-15 14:39:38 +03:00
committed by Gary Russell
parent f4bd03440f
commit 4b1eed27e3
3 changed files with 97 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

58
spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpoint.java
View File

@@ -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}.
* <p>
* If the queue does contain elements, the data is retrieved immediately. However,
* <p> 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.
* <p>
* A timeout of zero can be used to block indefinitely. If not set explicitly
* <p> A timeout of zero can be used to block indefinitely. If not set explicitly
* the timeout value will default to {@code 1000}
* <p>
* See also: http://redis.io/commands/brpop
*
* <p> 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();
}
}
}
}