+ *
* 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 'delayHeaderName' property is configured, that value, if * present on a Message, will take precedence over the handler's 'defaultDelay' * value. The actual header value may be a long, a String that can be parsed @@ -67,54 +57,49 @@ import org.springframework.util.Assert; * 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 toheaderDate.getTime() - new Date().getTime()).
- *
+ *
* @author Mark Fisher
+ * @author Artem Bilan
* @since 1.0.3
*/
-public class DelayHandler extends IntegrationObjectSupport implements MessageHandler, MessageProducer, Ordered, DisposableBean {
+public class DelayHandler extends AbstractReplyProducingMessageHandler implements ApplicationListenerDefault is "false". Switch this to "true" if you prefer - * fully completed tasks at the expense of a longer shutdown phase. - *
- * This property will only have an effect for TaskScheduler implementations
- * that extend from {@link ExecutorConfigurationSupport}.
- * @see ExecutorConfigurationSupport#setWaitForTasksToCompleteOnShutdown(boolean)
- */
- public void setWaitForTasksToCompleteOnShutdown(boolean waitForJobsToCompleteOnShutdown) {
- this.waitForTasksToCompleteOnShutdown = waitForJobsToCompleteOnShutdown;
- }
-
- public void setOrder(int order) {
- this.order = order;
- }
-
- public int getOrder() {
- return this.order;
- }
-
@Override
public String getComponentType() {
return "delayer";
}
- protected void onInit() throws Exception{
- if (this.getTaskScheduler() instanceof ExecutorConfigurationSupport) {
- ((ExecutorConfigurationSupport) this.getTaskScheduler()).setWaitForTasksToCompleteOnShutdown(this.waitForTasksToCompleteOnShutdown);
- }
- else if (logger.isWarnEnabled()) {
- logger.warn("The 'waitForJobsToCompleteOnShutdown' property is not supported for TaskScheduler of type [" +
- this.getTaskScheduler().getClass() + "]");
- }
+ @Override
+ protected void onInit() {
+ super.onInit();
if (this.messageStore == null) {
this.messageStore = new SimpleMessageStore();
}
- if (this.getTaskScheduler() instanceof InitializingBean) {
- ((InitializingBean) this.getTaskScheduler()).afterPropertiesSet();
- }
- if (this.getBeanFactory() != null){
- this.channelResolver = new BeanFactoryChannelResolver(this.getBeanFactory());
+ else {
+ Assert.isInstanceOf(MessageStore.class, this.messageStore);
}
}
- public final void handleMessage(final Message message) {
- long delay = this.determineDelayForMessage(message);
- if (delay > 0) {
- this.releaseMessageAfterDelay(message, delay);
- }
- else {
- // no delay, send directly
- this.sendMessageToReplyChannel(message);
+ /**
+ * Checks if 'requestMessage' wasn't delayed before
+ * ({@link #releaseMessageAfterDelay} and {@link DelayedMessageWrapper}).
+ * Than determine 'delay' for 'requestMessage' ({@link #determineDelayForMessage})
+ * and if delay > 0 schedules 'releaseMessage' task after 'delay'.
+ *
+ * @param requestMessage - the Message which may be delayed.
+ * @return - null if 'requestMessage' is delayed,
+ * otherwise - 'payload' from 'requestMessage'.
+ * @see #releaseMessage
+ */
+
+ @Override
+ protected Object handleRequestMessage(Message requestMessage) {
+ boolean delayed = requestMessage.getPayload() instanceof DelayedMessageWrapper;
+
+ if (!delayed) {
+ long delay = this.determineDelayForMessage(requestMessage);
+ if (delay > 0) {
+ this.releaseMessageAfterDelay(requestMessage, delay);
+ return null;
+ }
}
+
+ // no delay
+ Object payload = requestMessage.getPayload();
+ return delayed ? ((DelayedMessageWrapper) payload).getOriginal().getPayload() : payload;
}
private long determineDelayForMessage(Message message) {
@@ -231,98 +191,116 @@ public class DelayHandler extends IntegrationObjectSupport implements MessageHan
}
private void releaseMessageAfterDelay(final Message message, long delay) {
- Assert.state(this.messageStore != null, "MessageStore must not be null");
- final Message storedMessage = this.messageStore.addMessage(message);
+ Message delayedMessage = message;
+
+ DelayedMessageWrapper messageWrapper = null;
+ if (message.getPayload() instanceof DelayedMessageWrapper) {
+ messageWrapper = (DelayedMessageWrapper) message.getPayload();
+ }
+ else {
+ messageWrapper = new DelayedMessageWrapper(message);
+ delayedMessage = MessageBuilder.withPayload(messageWrapper).copyHeaders(message.getHeaders()).build();
+ this.messageStore.addMessageToGroup(this.messageGroupId, delayedMessage);
+ }
+
+ final Message messageToSchedule = delayedMessage;
+
this.getTaskScheduler().schedule(new Runnable() {
public void run() {
- try {
- releaseMessage(storedMessage.getHeaders().getId());
- }
- catch (Exception e) {
- Exception exception = new MessageHandlingException(message, "Failed to deliver Message after delay.", e);
- MessageChannel errorChannel = resolveErrorChannelIfPossible(message);
- if (errorChannel != null) {
- ErrorMessage errorMessage = new ErrorMessage(exception);
- try {
- messagingTemplate.send(errorChannel, errorMessage);
- }
- catch (Exception e2) {
- if (logger.isWarnEnabled()) {
- logger.warn("Failed to send MessagingException to error channel.", exception);
- }
- }
+ releaseMessage(messageToSchedule);
+ }
+ }, new Date(messageWrapper.getRequestDate() + delay));
+ }
+
+ private void releaseMessage(Message message) {
+ if (this.messageStore instanceof SimpleMessageStore
+ || ((MessageStore) this.messageStore).removeMessage(message.getHeaders().getId()) != null) {
+ this.messageStore.removeMessageFromGroup(this.messageGroupId, message);
+ this.handleMessageInternal(message);
+ }
+ else {
+ if (logger.isDebugEnabled()) {
+ logger.debug("No message in the Message Store to release: " + message +
+ ". Likely another instance has already released it.");
+ }
+ }
+ }
+
+ /**
+ * Used for reading persisted Messages in the 'messageStore'
+ * to reschedule them e.g. upon application restart.
+ * The logic is based on iteration over 'messageGroup.getMessages()'
+ * and schedules task about 'delay' logic.
+ * This behavior is dictated by the avoidance of invocation thread overload.
+ */
+ public void reschedulePersistedMessages() {
+ MessageGroup messageGroup = this.messageStore.getMessageGroup(this.messageGroupId);
+ for (final Message message : messageGroup.getMessages()) {
+ this.getTaskScheduler().schedule(new Runnable() {
+ public void run() {
+ long delay = determineDelayForMessage(message);
+ if (delay > 0) {
+ releaseMessageAfterDelay(message, delay);
}
- else if (logger.isWarnEnabled()) {
- logger.warn("No error channel available. MessagingException will be ignored.", exception);
+ else {
+ releaseMessage(message);
}
}
- }
- }, new Date(System.currentTimeMillis() + delay));
+ }, new Date());
+ }
}
- private void releaseMessage(UUID id) {
- Assert.state(this.messageStore != null, "MessageStore must not be null");
- Message message = this.messageStore.removeMessage(id);
- Assert.notNull(message, "Message with id: " + id + " no longer exists in MessageStore.");
- this.sendMessageToReplyChannel(message);
+ /**
+ * 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
+ */
+ public void onApplicationEvent(ContextRefreshedEvent event) {
+ if (!this.initialized.getAndSet(true)) {
+ this.reschedulePersistedMessages();
+ }
}
- private void sendMessageToReplyChannel(Message message) {
- MessageChannel replyChannel = this.resolveReplyChannel(message);
- this.messagingTemplate.send(replyChannel, message);
+ private static final class DelayedMessageWrapper implements Serializable {
+
+ private static final long serialVersionUID = -4739802369074947045L;
+
+ private final long requestDate = System.currentTimeMillis();
+
+ private final Message original;
+
+ public DelayedMessageWrapper(Message original) {
+ this.original = original;
+ }
+
+ public long getRequestDate() {
+ return this.requestDate;
+ }
+
+ public Message getOriginal() {
+ return this.original;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+
+ DelayedMessageWrapper that = (DelayedMessageWrapper) o;
+
+ return this.original.equals(that.original);
+ }
+
+ @Override
+ public int hashCode() {
+ return this.original.hashCode();
+ }
}
- private MessageChannel resolveReplyChannel(Message message) {
- MessageChannel replyChannel = this.outputChannel;
- if (replyChannel == null) {
- replyChannel = this.resolveChannelFromHeader(message, MessageHeaders.REPLY_CHANNEL);
- }
- if (replyChannel == null) {
- throw new ChannelResolutionException(
- "unable to resolve reply channel for message: " + message);
- }
- return replyChannel;
- }
-
- private MessageChannel resolveErrorChannelIfPossible(Message message) {
- MessageChannel errorChannel = null;
- try {
- errorChannel = this.resolveChannelFromHeader(message, MessageHeaders.ERROR_CHANNEL);
- }
- catch (Exception e) {
- if (logger.isWarnEnabled()) {
- logger.warn("Failed to resolve error channel from header.", e);
- }
- }
- if (errorChannel == null && this.channelResolver != null) {
- errorChannel = this.channelResolver.resolveChannelName(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME);
- }
- return errorChannel;
- }
-
- private MessageChannel resolveChannelFromHeader(Message message, String headerName) {
- MessageChannel channel = null;
- Object channelHeader = message.getHeaders().get(headerName);
- if (channelHeader != null) {
- if (channelHeader instanceof MessageChannel) {
- channel = (MessageChannel) channelHeader;
- }
- else if (channelHeader instanceof String) {
- Assert.state(this.channelResolver != null,
- "ChannelResolver is required for resolving '" + headerName + "' by name.");
- channel = this.channelResolver.resolveChannelName((String) channelHeader);
- }
- else {
- throw new ChannelResolutionException("expected a MessageChannel or String for '" +
- headerName + "', but type is [" + channelHeader.getClass() + "]");
- }
- }
- return channel;
- }
-
- public void destroy() throws Exception {
- if (this.getTaskScheduler() instanceof DisposableBean) {
- ((DisposableBean) this.getTaskScheduler()).destroy();
- }
- }
}
diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.2.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.2.xsd
index 5c3ce54be6..cc92ba353a 100644
--- a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.2.xsd
+++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.2.xsd
@@ -716,7 +716,7 @@
-
+