diff --git a/spring-integration-core/src/main/java/org/springframework/integration/IntegrationMessageHeaderAccessor.java b/spring-integration-core/src/main/java/org/springframework/integration/IntegrationMessageHeaderAccessor.java index a585d1dcaa..03dfa00c9d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/IntegrationMessageHeaderAccessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/IntegrationMessageHeaderAccessor.java @@ -67,7 +67,7 @@ public class IntegrationMessageHeaderAccessor extends MessageHeaderAccessor { public static final String ACKNOWLEDGMENT_CALLBACK = "acknowledgmentCallback"; - private Set readOnlyHeaders = new HashSet(); + private Set 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(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") diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/DelayerEndpointSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/DelayerEndpointSpec.java index 9cc7fb4166..9f7b077ae2 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/DelayerEndpointSpec.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/DelayerEndpointSpec.java @@ -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 { - private final List delayedAdvice = new LinkedList(); + private final List delayedAdvice = new LinkedList<>(); DelayerEndpointSpec(DelayHandler delayHandler) { super(delayHandler); @@ -97,6 +105,105 @@ public final class DelayerEndpointSpec extends ConsumerEndpointSpec - * 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. *

- * 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 { + 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 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 delayedAdviceChain; + private List 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 only 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 + * only 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} to advise {@link DelayHandler.ReleaseMessageHandler} proxy. - * Usually used to add transactions to delayed messages retrieved from a transactional message store. - * + * Specify the {@code List} 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 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 adviceChain. + * Delegate {@link MessageHandler} implementation for 'release Message task'. Used as + * 'pointcut' to wrap 'release Message task' with adviceChain. * - * @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; diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/DelayHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/handler/DelayHandlerTests.java index 7199850190..1ae8d63cb2 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/handler/DelayHandlerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/handler/DelayHandlerTests.java @@ -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("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("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"); } + } } diff --git a/src/reference/asciidoc/delayer.adoc b/src/reference/asciidoc/delayer.adoc index 4ee6df7325..5426afa15e 100644 --- a/src/reference/asciidoc/delayer.adoc +++ b/src/reference/asciidoc/delayer.adoc @@ -134,6 +134,8 @@ TIP: `ThreadPoolTaskScheduler` has a property `errorHandler`, which can be injec This handler allows processing an `Exception` from the thread of the scheduled task sending the delayed message. By default, it uses an `org.springframework.scheduling.support.TaskUtils$LoggingErrorHandler`, and you can see a stack trace in the logs. You might want to consider using an `org.springframework.integration.channel.MessagePublishingErrorHandler`, which sends an `ErrorMessage` into an `error-channel`, either from the failed message's header or into the default `error-channel`. +This error handling is performed after a transaction rolls back (if present). +See <>. [[delayer-message-store]] ==== Delayer and a Message Store @@ -172,8 +174,6 @@ The following example shows an `advice-chain` within a ``: ---- ==== -// TODO: It would be good to have an example of the element here. - The `DelayHandler` can be exported as a JMX `MBean` with managed operations (`getDelayedMessageCount` and `reschedulePersistedMessages`), which allows the rescheduling of delayed persisted messages at runtime -- for example, if the `TaskScheduler` has previously been stopped. These operations can be invoked through a `Control Bus` command, as the following example shows: @@ -187,3 +187,20 @@ Message delayerReschedulingMessage = ==== NOTE: For more information regarding the message store, JMX, and the control bus, see <>. + +[[delayer-release-failures]] +==== Release Failures + +Starting with version 5.0.8, there are two new properties on the delayer: + +- `maxAttempts` (default 5) +- `retryDelay` (default 1 second) + +When a message is released, if the downstream flow fails, the release will be attempted after the `retryDelay`. +If the `maxAttempts` is reached, the message is discarded (unless the release is transactional, in which case the message will remain in the store, but will no longer be scheduled for release, until the application is restarted, or the `reschedulePersistedMessages()` method is invoked, as discussed above). + +In addition, you can configure a `delayedMessageErrorChannel`; when a release fails, an `ErrorMessage` is sent to that channel with the exception as the payload and has the `originalMessage` property. +The `ErrorMessage` contains a header `IntegrationMessageHeaderAccessor.DELIVERY_ATTEMPT` containing the current count. + +If the error flow consumes the error message and exits normally, no further action is taken; if the release is transactional, the transaction will commit and the message deleted from the store. +If the error flow throws an exception, the release will be retried up to `maxAttempts` as discussed above.