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

INT-4527: Delayer Retries and Error Handling

Browse Source 
JIRA: https://jira.spring.io/browse/INT-4527
Fixes https://github.com/spring-projects/spring-integration/issues/2543

- Add error handling within the scope of a transaction
- Add `retryDelay` and `maxAttempts`
- Include the delivery attempt header in the `ErrorMessage`
- Add `transactionalRelease()` methods to the DSL

Check context in `ContextRefreshedEvent`.

Complete test case; don't schedule for re-release after a successful send after a failure.

Use `identityHashCode` for deliveries map key - error flow might change the message `hashCode`.

Debug logging

Polishing; remove parameter from the `DelayerEndpointSpec.transactionalRelease()`

* Polishing some code style
This commit is contained in:
Gary Russell
2018-08-23 13:51:42 -04:00
committed by Artem Bilan
parent 68bccdc155
commit 075d237c04
5 changed files with 469 additions and 154 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
spring-integration-core/src/main/java/org/springframework/integration/IntegrationMessageHeaderAccessor.java
View File

@@ -67,7 +67,7 @@ public class IntegrationMessageHeaderAccessor extends MessageHeaderAccessor {
public static final String ACKNOWLEDGMENT_CALLBACK = "acknowledgmentCallback";
private Set<String> readOnlyHeaders = new HashSet<String>();
private Set<String> readOnlyHeaders = new HashSet<>();
public IntegrationMessageHeaderAccessor(Message<?> message) {
super(message);
@@ -84,33 +84,33 @@ public class IntegrationMessageHeaderAccessor extends MessageHeaderAccessor {
public void setReadOnlyHeaders(String... readOnlyHeaders) {
Assert.noNullElements(readOnlyHeaders, "'readOnlyHeaders' must not be contain null items.");
if (!ObjectUtils.isEmpty(readOnlyHeaders)) {
this.readOnlyHeaders = new HashSet<String>(Arrays.asList(readOnlyHeaders));
this.readOnlyHeaders = new HashSet<>(Arrays.asList(readOnlyHeaders));
}
}
@Nullable
public Long getExpirationDate() {
return this.getHeader(EXPIRATION_DATE, Long.class);
return getHeader(EXPIRATION_DATE, Long.class);
}
@Nullable
public Object getCorrelationId() {
return this.getHeader(CORRELATION_ID);
return getHeader(CORRELATION_ID);
}
public int getSequenceNumber() {
Number sequenceNumber = this.getHeader(SEQUENCE_NUMBER, Number.class);
Number sequenceNumber = getHeader(SEQUENCE_NUMBER, Number.class);
return (sequenceNumber != null ? sequenceNumber.intValue() : 0);
}
public int getSequenceSize() {
Number sequenceSize = this.getHeader(SEQUENCE_SIZE, Number.class);
Number sequenceSize = getHeader(SEQUENCE_SIZE, Number.class);
return (sequenceSize != null ? sequenceSize.intValue() : 0);
}
@Nullable
public Integer getPriority() {
Number priority = this.getHeader(PRIORITY, Number.class);
Number priority = getHeader(PRIORITY, Number.class);
return (priority != null ? priority.intValue() : null);
}
@@ -133,19 +133,20 @@ public class IntegrationMessageHeaderAccessor extends MessageHeaderAccessor {
* @return the callback.
* @since 5.0.1
*/
@Nullable
public AcknowledgmentCallback getAcknowledgmentCallback() {
return getHeader(ACKNOWLEDGMENT_CALLBACK, AcknowledgmentCallback.class);
}
/**
* When a message-driven enpoint supports retry implicitly, this
* When a message-driven endpoint supports retry implicitly, this
* header is incremented for each delivery attempt.
* @return the delivery attempt.
* @since 5.0.1
*/
@Nullable
public AtomicInteger getDeliveryAttempt() {
return this.getHeader(DELIVERY_ATTEMPT, AtomicInteger.class);
return getHeader(DELIVERY_ATTEMPT, AtomicInteger.class);
}
@SuppressWarnings("unchecked")

111
spring-integration-core/src/main/java/org/springframework/integration/dsl/DelayerEndpointSpec.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016 the original author or authors.
* Copyright 2016-2018 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,19 +27,27 @@ import org.springframework.expression.Expression;
import org.springframework.integration.expression.FunctionExpression;
import org.springframework.integration.handler.DelayHandler;
import org.springframework.integration.store.MessageGroupStore;
import org.springframework.integration.transaction.TransactionInterceptorBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.interceptor.DefaultTransactionAttribute;
import org.springframework.transaction.interceptor.TransactionInterceptor;
import org.springframework.util.Assert;
/**
* A {@link ConsumerEndpointSpec} for a {@link DelayHandler}.
*
* @author Artem Bilan
* @author Gary Russell
*
* @since 5.0
*/
public final class DelayerEndpointSpec extends ConsumerEndpointSpec<DelayerEndpointSpec, DelayHandler> {
private final List<Advice> delayedAdvice = new LinkedList<Advice>();
private final List<Advice> delayedAdvice = new LinkedList<>();
DelayerEndpointSpec(DelayHandler delayHandler) {
super(delayHandler);
@@ -97,6 +105,105 @@ public final class DelayerEndpointSpec extends ConsumerEndpointSpec<DelayerEndpo
return this;
}
/**
* Set a message channel to which an {@link ErrorMessage} will be sent if sending the
* released message fails. If the error flow returns normally, the release is complete.
* If the error flow throws an exception, the release will be re-attempted.
* If there is a transaction advice on the release task, the error flow is called
* within the transaction.
* @param channel the channel.
* @return the endpoint spec.
* @see #maxAttempts(int)
* @see #retryDelay(long)
* @since 5.0.8
*/
public DelayerEndpointSpec delayedMessageErrorChannel(MessageChannel channel) {
this.handler.setDelayedMessageErrorChannel(channel);
return this;
}
/**
* Set a message channel name to which an {@link ErrorMessage} will be sent if sending
* the released message fails. If the error flow returns normally, the release is
* complete. If the error flow throws an exception, the release will be re-attempted.
* If there is a transaction advice on the release task, the error flow is called
* within the transaction.
* @param channel the channel name.
* @return the endpoint spec.
* @see #maxAttempts(int)
* @see #retryDelay(long)
* @since 5.0.8
*/
public DelayerEndpointSpec delayedMessageErrorChannel(String channel) {
this.handler.setDelayedMessageErrorChannelName(channel);
return this;
}
/**
* Set the maximum number of release attempts for when message release fails.
* Default {@value DelayHandler#DEFAULT_MAX_ATTEMPTS}.
* @param maxAttempts the max attempts.
* @return the endpoint spec.
* @see #retryDelay(long)
* @since 5.0.8
*/
public DelayerEndpointSpec maxAttempts(int maxAttempts) {
this.handler.setMaxAttempts(maxAttempts);
return this;
}
/**
* Set an additional delay to apply when retrying after a release failure.
* Default {@value DelayHandler#DEFAULT_RETRY_DELAY}.
* @param retryDelay the retry delay.
* @return the endpoint spec.
* @see #maxAttempts(int)
* @since 5.0.8
*/
public DelayerEndpointSpec retryDelay(long retryDelay) {
this.handler.setRetryDelay(retryDelay);
return this;
}
/**
* Specify a {@link TransactionInterceptor} {@link Advice} with default
* {@link PlatformTransactionManager} and {@link DefaultTransactionAttribute} for the
* {@link MessageHandler}.
* @return the spec.
* @since 5.0.8
*/
public DelayerEndpointSpec transactionalRelease() {
TransactionInterceptor transactionInterceptor = new TransactionInterceptorBuilder().build();
this.componentsToRegister.put(transactionInterceptor, null);
return delayedAdvice(transactionInterceptor);
}
/**
* Specify a {@link TransactionInterceptor} {@link Advice} for the {@link MessageHandler}.
* @param transactionInterceptor the {@link TransactionInterceptor} to use.
* @return the spec.
* @see TransactionInterceptorBuilder
* @since 5.0.8
*/
public DelayerEndpointSpec transactionalRelease(TransactionInterceptor transactionInterceptor) {
return delayedAdvice(transactionInterceptor);
}
/**
* Specify a {@link TransactionInterceptor} {@link Advice} with the provided
* {@code PlatformTransactionManager} and default {@link DefaultTransactionAttribute}
* for the {@link MessageHandler}.
* @param transactionManager the {@link PlatformTransactionManager} to use.
* @return the spec.
* @since 5.0.8
*/
public DelayerEndpointSpec transactionalRelease(PlatformTransactionManager transactionManager) {
return transactionalRelease(
new TransactionInterceptorBuilder()
.transactionManager(transactionManager)
.build());
}
/**
* Specify the function to determine delay value against {@link Message}.
* Typically used with a Java 8 Lambda expression:

340
spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2017 the original author or authors.
* Copyright 2002-2018 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.
@@ -18,10 +18,14 @@ package org.springframework.integration.handler;
import java.io.Serializable;
import java.util.Collection;
import java.util.Collections;
import java.util.Date;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import org.aopalliance.aop.Advice;
@@ -31,6 +35,7 @@ import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.expression.EvaluationContext;
import org.springframework.expression.EvaluationException;
import org.springframework.expression.Expression;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.expression.ExpressionUtils;
import org.springframework.integration.store.MessageGroup;
@@ -40,37 +45,38 @@ import org.springframework.integration.store.SimpleMessageStore;
import org.springframework.integration.support.management.IntegrationManagedResource;
import org.springframework.jmx.export.annotation.ManagedResource;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;
/**
* A {@link MessageHandler} that is capable of delaying the continuation of a
* Message flow based on the result of evaluation {@code delayExpression} on an inbound {@link Message}
* or a default delay value configured on this handler. Note that the
* continuation of the flow is delegated to a {@link TaskScheduler}, and
* therefore, the calling thread does not block. The advantage of this approach
* is that many delays can be managed concurrently, even very long delays,
* without producing a buildup of blocked Threads.
* A {@link MessageHandler} that is capable of delaying the continuation of a Message flow
* based on the result of evaluation {@code delayExpression} on an inbound {@link Message}
* or a default delay value configured on this handler. Note that the continuation of the
* flow is delegated to a {@link TaskScheduler}, and therefore, the calling thread does
* not block. The advantage of this approach is that many delays can be managed
* concurrently, even very long delays, without producing a buildup of blocked Threads.
* <p>
* One thing to keep in mind, however, is that any active transactional context
* will not propagate from the original sender to the eventual recipient. This
* is a side-effect of passing the Message to the output channel after the
* delay with a different Thread in control.
* One thing to keep in mind, however, is that any active transactional context will not
* propagate from the original sender to the eventual recipient. This is a side-effect of
* passing the Message to the output channel after the delay with a different Thread in
* control.
* <p>
* When this handler's {@code delayExpression} property is configured, that evaluation result value
* will take precedence over the handler's {@code defaultDelay} value.
* The actual evaluation result value may be a long, a String that can be parsed
* as a long, or a Date. If it is a long, it will be interpreted as the length
* of time to delay in milliseconds counting from the current time (e.g. a
* value of 5000 indicates that the Message can be released as soon as five
* seconds from the current time). If the value is a Date, it will be
* delayed at least until that Date occurs (i.e. the delay in that case is
* equivalent to {@code headerDate.getTime() - new Date().getTime()}).
* When this handler's {@code delayExpression} property is configured, that evaluation
* result value will take precedence over the handler's {@code defaultDelay} value. The
* actual evaluation result value may be a long, a String that can be parsed as a long, or
* a Date. If it is a long, it will be interpreted as the length of time to delay in
* milliseconds counting from the current time (e.g. a value of 5000 indicates that the
* Message can be released as soon as five seconds from the current time). If the value is
* a Date, it will be delayed at least until that Date occurs (i.e. the delay in that case
* is equivalent to {@code headerDate.getTime() - new Date().getTime()}).
*
* @author Mark Fisher
* @author Artem Bilan
@@ -84,31 +90,44 @@ import org.springframework.util.CollectionUtils;
public class DelayHandler extends AbstractReplyProducingMessageHandler implements DelayHandlerManagement,
ApplicationListener<ContextRefreshedEvent> {
public static final int DEFAULT_MAX_ATTEMPTS = 5;
public static final long DEFAULT_RETRY_DELAY = 1_000;
private final String messageGroupId;
private volatile long defaultDelay;
private final ConcurrentMap<String, AtomicInteger> deliveries = new ConcurrentHashMap<>();
private long defaultDelay;
private Expression delayExpression;
private volatile boolean ignoreExpressionFailures = true;
private boolean ignoreExpressionFailures = true;
private volatile MessageGroupStore messageStore;
private MessageGroupStore messageStore;
private volatile List<Advice> delayedAdviceChain;
private List<Advice> delayedAdviceChain;
private final AtomicBoolean initialized = new AtomicBoolean();
private volatile MessageHandler releaseHandler = new ReleaseMessageHandler();
private MessageHandler releaseHandler = new ReleaseMessageHandler();
private EvaluationContext evaluationContext;
private MessageChannel delayedMessageErrorChannel;
private String delayedMessageErrorChannelName;
private int maxAttempts = DEFAULT_MAX_ATTEMPTS;
private long retryDelay = DEFAULT_RETRY_DELAY;
/**
* Create a DelayHandler with the given 'messageGroupId' that is used as 'key' for {@link MessageGroup}
* to store delayed Messages in the {@link MessageGroupStore}. The sending of Messages after
* the delay will be handled by registered in the ApplicationContext default {@link ThreadPoolTaskScheduler}.
*
* Create a DelayHandler with the given 'messageGroupId' that is used as 'key' for
* {@link MessageGroup} to store delayed Messages in the {@link MessageGroupStore}.
* The sending of Messages after the delay will be handled by registered in the
* ApplicationContext default {@link ThreadPoolTaskScheduler}.
* @param messageGroupId The message group identifier.
*
* @see IntegrationObjectSupport#getTaskScheduler()
*/
public DelayHandler(String messageGroupId) {
@@ -117,9 +136,8 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
}
/**
* Create a DelayHandler with the given default delay. The sending of Messages
* after the delay will be handled by the provided {@link TaskScheduler}.
*
* Create a DelayHandler with the given default delay. The sending of Messages after
* the delay will be handled by the provided {@link TaskScheduler}.
* @param messageGroupId The message group identifier.
* @param taskScheduler A task scheduler.
*/
@@ -129,11 +147,10 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
}
/**
* Set the default delay in milliseconds. If no {@code delayExpression} property
* has been provided, the default delay will be applied to all Messages. If
* a delay should <em>only</em> be applied to Messages with evaluation result from
* Set the default delay in milliseconds. If no {@code delayExpression} property has
* been provided, the default delay will be applied to all Messages. If a delay should
* <em>only</em> be applied to Messages with evaluation result from
* {@code delayExpression}, then set this value to 0.
*
* @param defaultDelay The default delay in milliseconds.
*/
public void setDefaultDelay(long defaultDelay) {
@@ -145,7 +162,6 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
* milliseconds) or a Date to delay until. If this property is set, the result of the
* expression evaluation (if not null) will take precedence over this handler's
* default delay.
*
* @param delayExpression The delay expression.
*/
public void setDelayExpression(Expression delayExpression) {
@@ -157,7 +173,6 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
* milliseconds) or a Date to delay until. If this property is set, the result of the
* expression evaluation (if not null) will take precedence over this handler's
* default delay.
*
* @param delayExpression The delay expression.
* @since 5.0
*/
@@ -166,15 +181,14 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
}
/**
* Specify whether {@code Exceptions} thrown by {@link #delayExpression} evaluation should be
* ignored (only logged). In this case case the delayer will fall back to the
* to the {@link #defaultDelay}.
* If this property is specified as {@code false}, any {@link #delayExpression} evaluation
* {@code Exception} will be thrown to the caller without falling back to the to the {@link #defaultDelay}.
* Default is {@code true}.
*
* @param ignoreExpressionFailures true if expression evaluation failures should be ignored.
*
* Specify whether {@code Exceptions} thrown by {@link #delayExpression} evaluation
* should be ignored (only logged). In this case case the delayer will fall back to
* the to the {@link #defaultDelay}. If this property is specified as {@code false},
* any {@link #delayExpression} evaluation {@code Exception} will be thrown to the
* caller without falling back to the to the {@link #defaultDelay}. Default is
* {@code true}.
* @param ignoreExpressionFailures true if expression evaluation failures should be
* ignored.
* @see #determineDelayForMessage
*/
public void setIgnoreExpressionFailures(boolean ignoreExpressionFailures) {
@@ -182,9 +196,8 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
}
/**
* Specify the {@link MessageGroupStore} that should be used to store Messages
* while awaiting the delay.
*
* Specify the {@link MessageGroupStore} that should be used to store Messages while
* awaiting the delay.
* @param messageStore The message store.
*/
public void setMessageStore(MessageGroupStore messageStore) {
@@ -193,11 +206,10 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
}
/**
* Specify the {@code List<Advice>} to advise {@link DelayHandler.ReleaseMessageHandler} proxy.
* Usually used to add transactions to delayed messages retrieved from a transactional message store.
*
* Specify the {@code List<Advice>} to advise
* {@link DelayHandler.ReleaseMessageHandler} proxy. Usually used to add transactions
* to delayed messages retrieved from a transactional message store.
* @param delayedAdviceChain The advice chain.
*
* @see #createReleaseMessageTask
*/
public void setDelayedAdviceChain(List<Advice> delayedAdviceChain) {
@@ -205,6 +217,71 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
this.delayedAdviceChain = delayedAdviceChain;
}
/**
* Set a message channel to which an {@link ErrorMessage} will be sent if sending the
* released message fails. If the error flow returns normally, the release is
* complete. If the error flow throws an exception, the release will be re-attempted.
* If there is a transaction advice on the release task, the error flow is called
* within the transaction.
* @param delayedMessageErrorChannel the channel.
* @see #setMaxAttempts(int)
* @see #setRetryDelay(long)
* @since 5.0.8
*/
public void setDelayedMessageErrorChannel(MessageChannel delayedMessageErrorChannel) {
this.delayedMessageErrorChannel = delayedMessageErrorChannel;
}
/**
* Set a message channel name to which an {@link ErrorMessage} will be sent if sending
* the released message fails. If the error flow returns normally, the release is
* complete. If the error flow throws an exception, the release will be re-attempted.
* If there is a transaction advice on the release task, the error flow is called
* within the transaction.
* @param delayedMessageErrorChannelName the channel name.
* @see #setMaxAttempts(int)
* @see #setRetryDelay(long)
* @since 5.0.8
*/
public void setDelayedMessageErrorChannelName(String delayedMessageErrorChannelName) {
this.delayedMessageErrorChannelName = delayedMessageErrorChannelName;
}
/**
* Set the maximum number of release attempts for when message release fails. Default
* {@value #DEFAULT_MAX_ATTEMPTS}.
* @param maxAttempts the max attempts.
* @see #setRetryDelay(long)
* @since 5.0.8
*/
public void setMaxAttempts(int maxAttempts) {
this.maxAttempts = maxAttempts;
}
/**
* Set an additional delay to apply when retrying after a release failure. Default
* {@value #DEFAULT_RETRY_DELAY}.
* @param retryDelay the retry delay.
* @see #setMaxAttempts(int)
* @since 5.0.8
*/
public void setRetryDelay(long retryDelay) {
this.retryDelay = retryDelay;
}
private MessageChannel getErrorChannel() {
if (this.delayedMessageErrorChannel != null) {
return this.delayedMessageErrorChannel;
}
if (this.delayedMessageErrorChannelName != null) {
if (getChannelResolver() != null) {
this.delayedMessageErrorChannel = getChannelResolver()
.resolveDestination(this.delayedMessageErrorChannelName);
}
}
return this.delayedMessageErrorChannel;
}
@Override
public String getComponentType() {
return "delayer";
@@ -241,14 +318,13 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
}
/**
* Checks if 'requestMessage' wasn't delayed before
* ({@link #releaseMessageAfterDelay} and {@link DelayHandler.DelayedMessageWrapper}).
* Than determine 'delay' for 'requestMessage' ({@link #determineDelayForMessage})
* and if {@code delay > 0} schedules 'releaseMessage' task after 'delay'.
*
* Checks if 'requestMessage' wasn't delayed before ({@link #releaseMessageAfterDelay}
* and {@link DelayHandler.DelayedMessageWrapper}). Than determine 'delay' for
* 'requestMessage' ({@link #determineDelayForMessage}) and if {@code delay > 0}
* schedules 'releaseMessage' task after 'delay'.
* @param requestMessage - the Message which may be delayed.
* @return - {@code null} if 'requestMessage' is delayed,
* otherwise - 'payload' from 'requestMessage'.
* @return - {@code null} if 'requestMessage' is delayed, otherwise - 'payload' from
* 'requestMessage'.
* @see #releaseMessage
*/
@Override
@@ -256,9 +332,9 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
boolean delayed = requestMessage.getPayload() instanceof DelayedMessageWrapper;
if (!delayed) {
long delay = this.determineDelayForMessage(requestMessage);
long delay = determineDelayForMessage(requestMessage);
if (delay > 0) {
this.releaseMessageAfterDelay(requestMessage, delay);
releaseMessageAfterDelay(requestMessage, delay);
return null;
}
}
@@ -310,7 +386,6 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
throw new MessageHandlingException(message, "Error occurred during 'delay' value determination",
delayValueException);
}
}
}
return delay;
@@ -332,7 +407,6 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
this.messageStore.addMessageToGroup(this.messageGroupId, delayedMessage);
}
Runnable releaseTask;
if (this.messageStore instanceof SimpleMessageStore) {
@@ -370,15 +444,78 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
}
private void releaseMessage(Message<?> message) {
this.releaseHandler.handleMessage(message);
String identity = ObjectUtils.getIdentityHexString(message);
this.deliveries.putIfAbsent(identity, new AtomicInteger());
try {
this.releaseHandler.handleMessage(message);
this.deliveries.remove(identity);
}
catch (Exception e) {
if (getErrorChannel() != null) {
ErrorMessage errorMessage = new ErrorMessage(e,
Collections.singletonMap(IntegrationMessageHeaderAccessor.DELIVERY_ATTEMPT,
new AtomicInteger(this.deliveries.get(identity).get() + 1)),
message);
try {
if (!(getErrorChannel().send(errorMessage))) {
if (this.logger.isDebugEnabled()) {
this.logger.debug("Failed to send error message: " + errorMessage);
}
rescheduleForRetry(message, identity);
}
else {
this.deliveries.remove(identity);
}
}
catch (Exception e1) {
if (this.logger.isDebugEnabled()) {
logger.debug("Error flow threw an exception for message: " + message, e1);
}
rescheduleForRetry(message, identity);
}
}
else {
if (this.logger.isDebugEnabled()) {
logger.debug("Release flow threw an exception for message: " + message, e);
}
if (!rescheduleForRetry(message, identity)) {
throw e; // there might be an error handler on the scheduler
}
}
}
}
private boolean rescheduleForRetry(Message<?> message, String identity) {
if (this.deliveries.get(identity).incrementAndGet() >= this.maxAttempts) {
this.logger.error("Discarding; maximum release attempts reached for: " + message);
this.deliveries.remove(identity);
return false;
}
if (this.retryDelay <= 0) {
rescheduleNow(message);
}
else {
rescheduleAt(message, new Date(System.currentTimeMillis() + this.retryDelay));
}
return true;
}
private void rescheduleNow(final Message<?> message) {
rescheduleAt(message, new Date());
}
protected void rescheduleAt(final Message<?> message, Date startTime) {
getTaskScheduler()
.schedule(() -> releaseMessage(message), startTime);
}
private void doReleaseMessage(Message<?> message) {
if (removeDelayedMessageFromMessageStore(message)) {
if (removeDelayedMessageFromMessageStore(message)
|| this.deliveries.get(ObjectUtils.getIdentityHexString(message)).get() > 0) {
if (!(this.messageStore instanceof SimpleMessageStore)) {
this.messageStore.removeMessagesFromGroup(this.messageGroupId, message);
}
this.handleMessageInternal(message);
handleMessageInternal(message);
}
else {
if (logger.isDebugEnabled()) {
@@ -412,56 +549,54 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
}
/**
* Used for reading persisted Messages in the 'messageStore'
* to reschedule them e.g. upon application restart.
* The logic is based on iteration over {@code messageGroup.getMessages()}
* and schedules task for 'delay' logic.
* This behavior is dictated by the avoidance of invocation thread overload.
* Used for reading persisted Messages in the 'messageStore' to reschedule them e.g.
* upon application restart. The logic is based on iteration over
* {@code messageGroup.getMessages()} and schedules task for 'delay' logic. This
* behavior is dictated by the avoidance of invocation thread overload.
*/
@Override
public synchronized void reschedulePersistedMessages() {
MessageGroup messageGroup = this.messageStore.getMessageGroup(this.messageGroupId);
for (final Message<?> message : messageGroup.getMessages()) {
getTaskScheduler().schedule((Runnable) () -> {
// This is fine to keep the reference to the message,
// because the scheduled task is performed immediately.
long delay = determineDelayForMessage(message);
if (delay > 0) {
releaseMessageAfterDelay(message, delay);
}
else {
releaseMessage(message);
}
}, new Date());
getTaskScheduler()
.schedule(() -> {
// This is fine to keep the reference to the message,
// because the scheduled task is performed immediately.
long delay = determineDelayForMessage(message);
if (delay > 0) {
releaseMessageAfterDelay(message, delay);
}
else {
releaseMessage(message);
}
}, new Date());
}
}
/**
* Handles {@link ContextRefreshedEvent} to invoke {@link #reschedulePersistedMessages}
* as late as possible after application context startup.
* Also it checks {@link #initialized} to ignore
* other {@link ContextRefreshedEvent}s which may be published
* in the 'parent-child' contexts, e.g. in the Spring-MVC applications.
*
* @param event - {@link ContextRefreshedEvent} which occurs
* after Application context is completely initialized.
*
* Handles {@link ContextRefreshedEvent} to invoke
* {@link #reschedulePersistedMessages} as late as possible after application context
* startup. Also it checks {@link #initialized} to ignore other
* {@link ContextRefreshedEvent}s which may be published in the 'parent-child'
* contexts, e.g. in the Spring-MVC applications.
* @param event - {@link ContextRefreshedEvent} which occurs after Application context
* is completely initialized.
* @see #reschedulePersistedMessages
*/
@Override
public void onApplicationEvent(ContextRefreshedEvent event) {
if (!this.initialized.getAndSet(true)) {
this.reschedulePersistedMessages();
if (event.getApplicationContext().equals(getApplicationContext())
&& !this.initialized.getAndSet(true)) {
reschedulePersistedMessages();
}
}
/**
* Delegate {@link MessageHandler} implementation for 'release Message task'.
* Used as 'pointcut' to wrap 'release Message task' with <code>adviceChain</code>.
* Delegate {@link MessageHandler} implementation for 'release Message task'. Used as
* 'pointcut' to wrap 'release Message task' with <code>adviceChain</code>.
*
* @see @createReleaseMessageTask
* @see @releaseMessage
* @see #createReleaseMessageTask
* @see #releaseMessage
*/
private class ReleaseMessageHandler implements MessageHandler {
@@ -476,7 +611,6 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
}
public static final class DelayedMessageWrapper implements Serializable {
private static final long serialVersionUID = -4739802369074947045L;

132
spring-integration-core/src/test/java/org/springframework/integration/handler/DelayHandlerTests.java
View File

@@ -16,19 +16,23 @@
package org.springframework.integration.handler;
import static org.hamcrest.Matchers.equalTo;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNotSame;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.mock;
import java.util.Calendar;
import java.util.Date;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.junit.After;
import org.junit.Before;
@@ -41,6 +45,7 @@ import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.context.support.StaticApplicationContext;
import org.springframework.expression.Expression;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.integration.StaticMessageHeaderAccessor;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.MessagePublishingErrorHandler;
import org.springframework.integration.channel.QueueChannel;
@@ -50,10 +55,12 @@ import org.springframework.integration.store.MessageGroupStore;
import org.springframework.integration.store.SimpleMessageStore;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.integration.test.util.TestUtils.TestApplicationContext;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageDeliveryException;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
@@ -106,12 +113,14 @@ public class DelayHandlerTests {
private void startDelayerHandler() {
delayHandler.afterPropertiesSet();
delayHandler.onApplicationEvent(new ContextRefreshedEvent(TestUtils.createTestApplicationContext()));
TestApplicationContext ac = TestUtils.createTestApplicationContext();
delayHandler.setApplicationContext(ac);
delayHandler.onApplicationEvent(new ContextRefreshedEvent(ac));
}
@Test
public void noDelayHeaderAndDefaultDelayIsZero() throws Exception {
this.startDelayerHandler();
public void noDelayHeaderAndDefaultDelayIsZero() {
startDelayerHandler();
Message<?> message = MessageBuilder.withPayload("test").build();
input.send(message);
assertSame(message.getPayload(), resultHandler.lastMessage.getPayload());
@@ -119,9 +128,9 @@ public class DelayHandlerTests {
}
@Test
public void noDelayHeaderAndDefaultDelayIsPositive() throws Exception {
public void noDelayHeaderAndDefaultDelayIsPositive() {
delayHandler.setDefaultDelay(10);
this.startDelayerHandler();
startDelayerHandler();
Message<?> message = MessageBuilder.withPayload("test").build();
input.send(message);
waitForLatch(10000);
@@ -130,10 +139,49 @@ public class DelayHandlerTests {
}
@Test
public void delayHeaderAndDefaultDelayWouldTimeout() throws Exception {
public void errorFlowAndRetries() throws Exception {
delayHandler.setDefaultDelay(10);
delayHandler.setRetryDelay(15);
startDelayerHandler();
Message<?> message = MessageBuilder.withPayload("test")
.setHeader("foo", new AtomicInteger())
.build();
final CountDownLatch latch = new CountDownLatch(1);
final AtomicInteger count = new AtomicInteger();
delayHandler.setDelayedMessageErrorChannel((m, t) -> {
count.incrementAndGet();
int deliveries = StaticMessageHeaderAccessor.getDeliveryAttempt(m).get();
((MessagingException) m.getPayload())
.getFailedMessage()
.getHeaders()
.get("foo", AtomicInteger.class)
.incrementAndGet();
if (deliveries < 3) {
throw new RuntimeException("fail");
}
else if (deliveries == 3) {
return false;
}
else {
latch.countDown();
return true;
}
});
delayHandler.setOutputChannel((m, t) -> {
throw new MessagingException(m);
});
input.send(message);
assertTrue(latch.await(10, TimeUnit.SECONDS));
Thread.sleep(50);
assertThat(count.get(), equalTo(4));
assertThat(TestUtils.getPropertyValue(this.delayHandler, "deliveries", Map.class).size(), equalTo(0));
}
@Test
public void delayHeaderAndDefaultDelayWouldTimeout() {
delayHandler.setDefaultDelay(5000);
this.setDelayExpression();
this.startDelayerHandler();
startDelayerHandler();
Message<?> message = MessageBuilder.withPayload("test")
.setHeader("delay", 100).build();
input.send(message);
@@ -143,10 +191,10 @@ public class DelayHandlerTests {
}
@Test
public void delayHeaderIsNegativeAndDefaultDelayWouldTimeout() throws Exception {
public void delayHeaderIsNegativeAndDefaultDelayWouldTimeout() {
delayHandler.setDefaultDelay(5000);
this.setDelayExpression();
this.startDelayerHandler();
startDelayerHandler();
Message<?> message = MessageBuilder.withPayload("test")
.setHeader("delay", -7000).build();
input.send(message);
@@ -156,10 +204,10 @@ public class DelayHandlerTests {
}
@Test
public void delayHeaderIsInvalidFallsBackToDefaultDelay() throws Exception {
public void delayHeaderIsInvalidFallsBackToDefaultDelay() {
delayHandler.setDefaultDelay(5);
this.setDelayExpression();
this.startDelayerHandler();
startDelayerHandler();
Message<?> message = MessageBuilder.withPayload("test")
.setHeader("delay", "not a number").build();
input.send(message);
@@ -169,10 +217,10 @@ public class DelayHandlerTests {
}
@Test
public void delayHeaderIsDateInTheFutureAndDefaultDelayWouldTimeout() throws Exception {
public void delayHeaderIsDateInTheFutureAndDefaultDelayWouldTimeout() {
delayHandler.setDefaultDelay(5000);
this.setDelayExpression();
this.startDelayerHandler();
startDelayerHandler();
Message<?> message = MessageBuilder.withPayload("test")
.setHeader("delay", new Date(new Date().getTime() + 150)).build();
input.send(message);
@@ -182,10 +230,10 @@ public class DelayHandlerTests {
}
@Test
public void delayHeaderIsDateInThePastAndDefaultDelayWouldTimeout() throws Exception {
public void delayHeaderIsDateInThePastAndDefaultDelayWouldTimeout() {
delayHandler.setDefaultDelay(5000);
this.setDelayExpression();
this.startDelayerHandler();
startDelayerHandler();
Message<?> message = MessageBuilder.withPayload("test")
.setHeader("delay", new Date(new Date().getTime() - 60 * 1000)).build();
input.send(message);
@@ -195,9 +243,9 @@ public class DelayHandlerTests {
}
@Test
public void delayHeaderIsNullDateAndDefaultDelayIsZero() throws Exception {
public void delayHeaderIsNullDateAndDefaultDelayIsZero() {
this.setDelayExpression();
this.startDelayerHandler();
startDelayerHandler();
Date nullDate = null;
Message<?> message = MessageBuilder.withPayload("test")
.setHeader("delay", nullDate).build();
@@ -208,9 +256,9 @@ public class DelayHandlerTests {
}
@Test(expected = TestTimedOutException.class)
public void delayHeaderIsFutureDateAndTimesOut() throws Exception {
public void delayHeaderIsFutureDateAndTimesOut() {
this.setDelayExpression();
this.startDelayerHandler();
startDelayerHandler();
Date future = new Date(new Date().getTime() + 60 * 1000);
Message<?> message = MessageBuilder.withPayload("test")
.setHeader("delay", future).build();
@@ -219,10 +267,10 @@ public class DelayHandlerTests {
}
@Test
public void delayHeaderIsValidStringAndDefaultDelayWouldTimeout() throws Exception {
public void delayHeaderIsValidStringAndDefaultDelayWouldTimeout() {
delayHandler.setDefaultDelay(5000);
this.setDelayExpression();
this.startDelayerHandler();
startDelayerHandler();
Message<?> message = MessageBuilder.withPayload("test")
.setHeader("delay", "20").build();
input.send(message);
@@ -234,8 +282,8 @@ public class DelayHandlerTests {
@Test
public void verifyShutdownWithoutWaitingByDefault() throws Exception {
delayHandler.setDefaultDelay(5000);
this.startDelayerHandler();
delayHandler.handleMessage(new GenericMessage<String>("foo"));
startDelayerHandler();
delayHandler.handleMessage(new GenericMessage<>("foo"));
taskScheduler.destroy();
final CountDownLatch latch = new CountDownLatch(1);
@@ -265,8 +313,8 @@ public class DelayHandlerTests {
}
@Test(expected = MessageDeliveryException.class)
public void handlerThrowsExceptionWithNoDelay() throws Exception {
this.startDelayerHandler();
public void handlerThrowsExceptionWithNoDelay() {
startDelayerHandler();
output.unsubscribe(resultHandler);
output.subscribe(message -> {
throw new UnsupportedOperationException("intentional test failure");
@@ -276,13 +324,14 @@ public class DelayHandlerTests {
}
@Test
public void errorChannelHeaderAndHandlerThrowsExceptionWithDelay() throws Exception {
public void errorChannelHeaderAndHandlerThrowsExceptionWithDelay() {
this.delayHandler.setRetryDelay(1);
DirectChannel errorChannel = new DirectChannel();
MessagePublishingErrorHandler errorHandler = new MessagePublishingErrorHandler();
errorHandler.setDefaultErrorChannel(errorChannel);
taskScheduler.setErrorHandler(errorHandler);
this.setDelayExpression();
this.startDelayerHandler();
startDelayerHandler();
output.unsubscribe(resultHandler);
errorChannel.subscribe(resultHandler);
output.subscribe(message -> {
@@ -302,7 +351,8 @@ public class DelayHandlerTests {
}
@Test
public void errorChannelNameHeaderAndHandlerThrowsExceptionWithDelay() throws Exception {
public void errorChannelNameHeaderAndHandlerThrowsExceptionWithDelay() {
this.delayHandler.setRetryDelay(1);
String errorChannelName = "customErrorChannel";
StaticApplicationContext context = new StaticApplicationContext();
context.registerSingleton(errorChannelName, DirectChannel.class);
@@ -313,7 +363,7 @@ public class DelayHandlerTests {
errorHandler.setBeanFactory(context);
taskScheduler.setErrorHandler(errorHandler);
this.setDelayExpression();
this.startDelayerHandler();
startDelayerHandler();
output.unsubscribe(resultHandler);
customErrorChannel.subscribe(resultHandler);
output.subscribe(message -> {
@@ -333,16 +383,18 @@ public class DelayHandlerTests {
}
@Test
public void defaultErrorChannelAndHandlerThrowsExceptionWithDelay() throws Exception {
public void defaultErrorChannelAndHandlerThrowsExceptionWithDelay() {
this.delayHandler.setRetryDelay(1);
StaticApplicationContext context = new StaticApplicationContext();
context.registerSingleton(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME, DirectChannel.class);
context.refresh();
DirectChannel defaultErrorChannel = (DirectChannel) context.getBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME);
DirectChannel defaultErrorChannel = (DirectChannel) context
.getBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME);
MessagePublishingErrorHandler errorHandler = new MessagePublishingErrorHandler();
errorHandler.setBeanFactory(context);
taskScheduler.setErrorHandler(errorHandler);
this.setDelayExpression();
this.startDelayerHandler();
startDelayerHandler();
output.unsubscribe(resultHandler);
defaultErrorChannel.subscribe(resultHandler);
output.subscribe(message -> {
@@ -365,7 +417,7 @@ public class DelayHandlerTests {
MessageGroupStore messageGroupStore = new SimpleMessageStore();
this.delayHandler.setDefaultDelay(2000);
this.delayHandler.setMessageStore(messageGroupStore);
this.startDelayerHandler();
startDelayerHandler();
Message<?> message = MessageBuilder.withPayload("test").build();
this.input.send(message);
@@ -390,7 +442,7 @@ public class DelayHandlerTests {
this.delayHandler.setDefaultDelay(200);
this.delayHandler.setMessageStore(messageGroupStore);
this.delayHandler.setBeanFactory(mock(BeanFactory.class));
this.startDelayerHandler();
startDelayerHandler();
waitForLatch(10000);
@@ -402,21 +454,23 @@ public class DelayHandlerTests {
@Test //INT-1132
// Can happen in the parent-child context e.g. Spring-MVC applications
public void testDoubleOnApplicationEvent() throws Exception {
public void testDoubleOnApplicationEvent() {
this.delayHandler = Mockito.spy(this.delayHandler);
Mockito.doAnswer(invocation -> null).when(this.delayHandler).reschedulePersistedMessages();
ContextRefreshedEvent contextRefreshedEvent = new ContextRefreshedEvent(TestUtils.createTestApplicationContext());
TestApplicationContext ac = TestUtils.createTestApplicationContext();
this.delayHandler.setApplicationContext(ac);
ContextRefreshedEvent contextRefreshedEvent = new ContextRefreshedEvent(ac);
this.delayHandler.onApplicationEvent(contextRefreshedEvent);
this.delayHandler.onApplicationEvent(contextRefreshedEvent);
Mockito.verify(this.delayHandler, Mockito.times(1)).reschedulePersistedMessages();
}
@Test(expected = MessageHandlingException.class)
public void testInt2243IgnoreExpressionFailuresAsFalse() throws Exception {
public void testInt2243IgnoreExpressionFailuresAsFalse() {
this.setDelayExpression();
this.delayHandler.setIgnoreExpressionFailures(false);
this.startDelayerHandler();
startDelayerHandler();
this.delayHandler.handleMessage(new GenericMessage<String>("test"));
}
@@ -502,6 +556,7 @@ public class DelayHandlerTests {
this.lastThread = Thread.currentThread();
latch.countDown();
}
}
@@ -511,6 +566,7 @@ public class DelayHandlerTests {
TestTimedOutException() {
super("timed out while waiting for latch");
}
}
}