From 87d8e86297f2ef7ac2945e6e579f109dd34e36bb Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Fri, 6 Dec 2013 17:22:40 -0500 Subject: [PATCH 1/4] INT-3227 Fix Race Condition for WARN Log JIRA: https://jira.springsource.org/browse/INT-3227 In MessagingTemplate.TemporaryReplyChannel, it was possible the `clientHasReceived` boolean could be set by the other thread after normal message receipt, but before the sender tests it for the purpose of emitting a duplicate reply log message. Capture the boolean value of `clientHasReceived` before counting down the latch. --- .../integration/core/MessagingTemplate.java | 54 ++++++++++++++----- 1 file changed, 42 insertions(+), 12 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/core/MessagingTemplate.java b/spring-integration-core/src/main/java/org/springframework/integration/core/MessagingTemplate.java index 41df197255..63d9a48d0d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/core/MessagingTemplate.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/core/MessagingTemplate.java @@ -130,6 +130,7 @@ public class MessagingTemplate implements MessagingOperations, BeanFactoryAware, this.receiveTimeout = receiveTimeout; } + @Override public void setBeanFactory(BeanFactory beanFactory) { if (this.channelResolver == null && beanFactory != null) { this.channelResolver = new BeanFactoryChannelResolver(beanFactory); @@ -148,6 +149,7 @@ public class MessagingTemplate implements MessagingOperations, BeanFactoryAware, this.throwExceptionOnLateReply = throwExceptionOnLateReply; } + @Override public void afterPropertiesSet() { synchronized (this.initializationMonitor) { if (this.initialized) { @@ -157,18 +159,22 @@ public class MessagingTemplate implements MessagingOperations, BeanFactoryAware, } } + @Override public

void send(final Message

message) { this.send(this.getRequiredDefaultChannel(), message); } + @Override public

void send(final MessageChannel channel, final Message

message) { this.doSend(channel, message); } + @Override public

void send(final String channelName, final Message

message) { this.send(this.resolveChannelName(channelName), message); } + @Override public void convertAndSend(T object) { Message message = this.messageConverter.toMessage(object); if (message != null) { @@ -176,6 +182,7 @@ public class MessagingTemplate implements MessagingOperations, BeanFactoryAware, } } + @Override public void convertAndSend(MessageChannel channel, T object) { Message message = this.messageConverter.toMessage(object); if (message != null) { @@ -183,6 +190,7 @@ public class MessagingTemplate implements MessagingOperations, BeanFactoryAware, } } + @Override public void convertAndSend(String channelName, T object) { Message message = this.messageConverter.toMessage(object); if (message != null) { @@ -190,6 +198,7 @@ public class MessagingTemplate implements MessagingOperations, BeanFactoryAware, } } + @Override public void convertAndSend(T object, MessagePostProcessor postProcessor) { Message message = this.messageConverter.toMessage(object); message = postProcessor.postProcessMessage(message); @@ -198,6 +207,7 @@ public class MessagingTemplate implements MessagingOperations, BeanFactoryAware, } } + @Override public void convertAndSend(MessageChannel channel, T object, MessagePostProcessor postProcessor) { Message message = this.messageConverter.toMessage(object); message = postProcessor.postProcessMessage(message); @@ -206,6 +216,7 @@ public class MessagingTemplate implements MessagingOperations, BeanFactoryAware, } } + @Override public void convertAndSend(String channelName, T object, MessagePostProcessor postProcessor) { Message message = this.messageConverter.toMessage(object); message = postProcessor.postProcessMessage(message); @@ -214,6 +225,7 @@ public class MessagingTemplate implements MessagingOperations, BeanFactoryAware, } } + @Override public

Message

receive() { MessageChannel channel = this.getRequiredDefaultChannel(); Assert.state(channel instanceof PollableChannel, @@ -221,10 +233,12 @@ public class MessagingTemplate implements MessagingOperations, BeanFactoryAware, return this.receive((PollableChannel) channel); } + @Override public

Message

receive(final PollableChannel channel) { return this.doReceive(channel); } + @Override public

Message

receive(String channelName) { MessageChannel channel = this.resolveChannelName(channelName); Assert.isInstanceOf(PollableChannel.class, channel, @@ -232,51 +246,61 @@ public class MessagingTemplate implements MessagingOperations, BeanFactoryAware, return this.receive((PollableChannel) channel); } + @Override public Object receiveAndConvert() throws MessagingException { Message message = this.receive(); return (message != null) ? this.messageConverter.fromMessage(message) : null; } + @Override public Object receiveAndConvert(PollableChannel channel) throws MessagingException { Message message = this.receive(channel); return (message != null) ? this.messageConverter.fromMessage(message) : null; } + @Override public Object receiveAndConvert(String channelName) throws MessagingException { Message message = this.receive(channelName); return (message != null) ? this.messageConverter.fromMessage(message) : null; } + @Override public Message sendAndReceive(final Message requestMessage) { return this.sendAndReceive(this.getRequiredDefaultChannel(), requestMessage); } + @Override public Message sendAndReceive(final MessageChannel channel, final Message requestMessage) { return this.doSendAndReceive(channel, requestMessage); } + @Override public Message sendAndReceive(final String channelName, final Message requestMessage) { return this.sendAndReceive(this.resolveChannelName(channelName), requestMessage); } + @Override public Object convertSendAndReceive(final Object request) { Message requestMessage = this.messageConverter.toMessage(request); Message replyMessage = this.sendAndReceive(requestMessage); return this.messageConverter.fromMessage(replyMessage); } + @Override public Object convertSendAndReceive(final MessageChannel channel, final Object request) { Message requestMessage = this.messageConverter.toMessage(request); Message replyMessage = this.sendAndReceive(channel, requestMessage); return this.messageConverter.fromMessage(replyMessage); } + @Override public Object convertSendAndReceive(final String channelName, final Object request) { Message requestMessage = this.messageConverter.toMessage(request); Message replyMessage = this.sendAndReceive(channelName, requestMessage); return this.messageConverter.fromMessage(replyMessage); } + @Override public Object convertSendAndReceive(final Object request, MessagePostProcessor requestPostProcessor) { Message requestMessage = this.messageConverter.toMessage(request); requestMessage = requestPostProcessor.postProcessMessage(requestMessage); @@ -284,6 +308,7 @@ public class MessagingTemplate implements MessagingOperations, BeanFactoryAware, return this.messageConverter.fromMessage(replyMessage); } + @Override public Object convertSendAndReceive(final MessageChannel channel, final Object request, MessagePostProcessor requestPostProcessor) { Message requestMessage = this.messageConverter.toMessage(request); requestMessage = requestPostProcessor.postProcessMessage(requestMessage); @@ -291,6 +316,7 @@ public class MessagingTemplate implements MessagingOperations, BeanFactoryAware, return this.messageConverter.fromMessage(replyMessage); } + @Override public Object convertSendAndReceive(final String channelName, final Object request, MessagePostProcessor requestPostProcessor) { Message requestMessage = this.messageConverter.toMessage(request); requestMessage = requestPostProcessor.postProcessMessage(requestMessage); @@ -405,10 +431,12 @@ public class MessagingTemplate implements MessagingOperations, BeanFactoryAware, } + @Override public Message receive() { return this.receive(-1); } + @Override public Message receive(long timeout) { try { if (this.receiveTimeout < 0) { @@ -430,25 +458,27 @@ public class MessagingTemplate implements MessagingOperations, BeanFactoryAware, return this.message; } + @Override public boolean send(Message message) { return this.send(message, -1); } + @Override public boolean send(Message message, long timeout) { this.message = message; + boolean clientHasReceived = this.clientHasReceived; this.latch.countDown(); - if (this.clientTimedOut || this.clientHasReceived || this.clientWontReceive) { - String exceptionMessage = ""; - if (this.clientTimedOut) { - exceptionMessage = "Reply message being sent, but the receiving thread has already timed out"; - } - else if (this.clientHasReceived) { - exceptionMessage = "Reply message being sent, but the receiving thread has already received a reply"; - } - else if (this.clientWontReceive) { - exceptionMessage = "Reply message being sent, but the receiving thread has already caught an exception and won't receive"; - } - + String exceptionMessage = null; + if (this.clientTimedOut) { + exceptionMessage = "Reply message being sent, but the receiving thread has already timed out"; + } + else if (clientHasReceived) { + exceptionMessage = "Reply message being sent, but the receiving thread has already received a reply"; + } + else if (this.clientWontReceive) { + exceptionMessage = "Reply message being sent, but the receiving thread has already caught an exception and won't receive"; + } + if (exceptionMessage != null) { if (logger.isWarnEnabled()) { logger.warn(exceptionMessage + ":" + message); } From 1a9f4bc9476b2fbf8a84bf2f382ba018ae1d0623 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 9 Dec 2013 13:53:56 +0200 Subject: [PATCH 2/4] INT-3219: Document Retry Exception Classification JIRA: https://jira.springsource.org/browse/INT-3219 * Upgrade to Spring Retry 1.0.3 * Add a section to Reference Manual about Retry Exception Classification --- build.gradle | 2 +- src/reference/docbook/handler-advice.xml | 28 +++++++++++++++++++++++- 2 files changed, 28 insertions(+), 2 deletions(-) diff --git a/build.gradle b/build.gradle index a2c9119800..6f9f62d2f4 100644 --- a/build.gradle +++ b/build.gradle @@ -70,7 +70,7 @@ subprojects { subproject -> springSecurityVersion = '3.1.3.RELEASE' springSocialTwitterVersion = '1.0.5.RELEASE' springWsVersion = '2.1.1.RELEASE' - springRetryVersion = '1.0.2.RELEASE' + springRetryVersion = '1.0.3.RELEASE' } eclipse { diff --git a/src/reference/docbook/handler-advice.xml b/src/reference/docbook/handler-advice.xml index 91fb6b1175..8a7a25e383 100644 --- a/src/reference/docbook/handler-advice.xml +++ b/src/reference/docbook/handler-advice.xml @@ -78,7 +78,7 @@ The retry advice (o.s.i.handler.advice.RequestHandlerRetryAdvice) leverages the rich retry mechanisms provided by the - spring-retry project. The core component + Spring Retry project. The core component of spring-retry is the RetryTemplate, which allows configuration of sophisticated retry scenarios, including RetryPolicy and BackoffPolicy strategies, with a number of implementations, @@ -288,6 +288,32 @@ Caused by: java.lang.RuntimeException: foo exception is thrown to the caller on each failure. + + Exception Classification for Retry + + + + Spring Retry has a great deal of flexibility for determining which + exceptions can invoke retry. The default configuration will retry + for all exceptions. Given that, user exceptions may be wrapped in + a MessagingException in the underlying handler, + we need to ensure that the classification examines the exception causes. + The default classifier just looks at the top level exception. + + + Since Spring Retry 1.0.3, the + BinaryExceptionClassifier has a property + traverseCauses (default false). When + true it will traverse exception causes until it + finds a match or there is no cause. + + + To use this classifier for retry, use a SimpleRetryPolicy + created with the constructor that takes the max attempts, the + Map of Exceptions + and the boolean (traverseCauses), and inject this policy into the + RetryTemplate. +

Circuit Breaker Advice From 1a21dd172fac997e6981070139bf5d901aa14883 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 4 Dec 2013 11:26:54 -0500 Subject: [PATCH 3/4] INT-3223 Tail - Exclusive Attributes JIRA: https://jira.springsource.org/browse/INT-3223 Disallow 'native-options' if any of the attributes required for the Apache implementation are specified. --- ...FileTailInboundChannelAdapterFactoryBean.java | 7 +++---- .../tail/FileTailingMessageProducerTests.java | 6 +++--- src/reference/docbook/file.xml | 16 ++++++++++++---- 3 files changed, 18 insertions(+), 11 deletions(-) diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterFactoryBean.java b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterFactoryBean.java index 4e50804d63..0cbff47592 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterFactoryBean.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterFactoryBean.java @@ -28,7 +28,7 @@ import org.springframework.integration.file.tail.ApacheCommonsFileTailingMessage import org.springframework.integration.file.tail.FileTailingMessageProducerSupport; import org.springframework.integration.file.tail.OSDelegatingFileTailingMessageProducer; import org.springframework.scheduling.TaskScheduler; -import org.springframework.util.StringUtils; +import org.springframework.util.Assert; /** * @author Gary Russell @@ -180,9 +180,8 @@ public class FileTailInboundChannelAdapterFactoryBean extends AbstractFactoryBea } } else { - if (this.nativeOptions != null && StringUtils.hasText(this.nativeOptions) && logger.isWarnEnabled()) { - logger.warn("'native-options' are ignored with an Apache commons-io 'Tailer' adapter"); - } + Assert.isTrue(this.nativeOptions == null, + "'native-options' is not allowed with 'delay', 'end', or 'reopen'"); adapter = new ApacheCommonsFileTailingMessageProducer(); if (this.delay != null) { ((ApacheCommonsFileTailingMessageProducer) adapter).setPollingDelay(this.delay); diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/tail/FileTailingMessageProducerTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/tail/FileTailingMessageProducerTests.java index 694fb680ab..ca41e653db 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/tail/FileTailingMessageProducerTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/tail/FileTailingMessageProducerTests.java @@ -45,10 +45,10 @@ import org.springframework.integration.file.tail.FileTailingMessageProducerSuppo */ public class FileTailingMessageProducerTests { - private static final String TAIL_OPTIONS_FOLLOW_NAME_MANY_LINES = "-F -n 99999999"; + private static final String TAIL_OPTIONS_FOLLOW_NAME_ALL_LINES = "-F -n +0"; @Rule - public TailRule tailRule = new TailRule(TAIL_OPTIONS_FOLLOW_NAME_MANY_LINES); + public TailRule tailRule = new TailRule(TAIL_OPTIONS_FOLLOW_NAME_ALL_LINES); private final Log logger = LogFactory.getLog(this.getClass()); @@ -76,7 +76,7 @@ public class FileTailingMessageProducerTests { @TailAvailable public void testOS() throws Exception { OSDelegatingFileTailingMessageProducer adapter = new OSDelegatingFileTailingMessageProducer(); - adapter.setOptions(TAIL_OPTIONS_FOLLOW_NAME_MANY_LINES); + adapter.setOptions(TAIL_OPTIONS_FOLLOW_NAME_ALL_LINES); testGuts(adapter, "reader"); } diff --git a/src/reference/docbook/file.xml b/src/reference/docbook/file.xml index 9cc835c8e2..515d3f0a00 100644 --- a/src/reference/docbook/file.xml +++ b/src/reference/docbook/file.xml @@ -168,7 +168,8 @@
'Tail'ing Files - Another popular use case is to get 'lines' from the end (or tail) of a file. Two implementations are provided; + Another popular use case is to get 'lines' from the end (or tail) of a file, capturing new lines when + they are added. Two implementations are provided; the first, OSDelegatingFileTailingMessageProducer, uses the native tail command (on operating systems that have one). This is likely the most efficient implementation on those platforms. For operating systems that do not have a tail command, the second implementation @@ -213,12 +214,12 @@ ]]> - This creates a native adapter with '-F -n 6' options (follow the file name, emit up to 6 lines before the current end). + This creates a native adapter with '-F -n +0' options (follow the file name, emitting all existing lines). If the tail command fails (on some platforms, a missing file causes the tail to fail, even with -F specified), the command will be retried every 10 seconds. @@ -231,11 +232,18 @@ reopen="true" file-delay="10000"/>]]> - This creates a commons-io Tailer adapter that examines the file for new lines every + This creates an Apache commons-io Tailer adapter that examines the file for new lines every 2 seconds, and checks for existence of a missing file every 10 seconds. The file will be tailed from the beginning (end="false") instead of the end (which is the default). The file will be reopened for each chunk (the default is to keep the file open). + + + Specifying the delay, end or reopen attributes, + forces the use of the Apache commons-io adapter and the native-options attribute is not + allowed. + +
From 9c9c93479a8938b0408d687f814fb76ac094899c Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 9 Dec 2013 14:47:00 -0500 Subject: [PATCH 4/4] INT-3223 Polishing JIRA: https://jira.springsource.org/browse/INT-3223 Don't use primitives in Tail factory bean setters so that the placeholders for attributes that force the Apache implementation can resolve to "". Test for an empty string in native-options and don't set. --- ...eTailInboundChannelAdapterFactoryBean.java | 13 ++++++++----- ...boundChannelAdapterParserTests-context.xml | 19 +++++++++++++++++-- 2 files changed, 25 insertions(+), 7 deletions(-) diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterFactoryBean.java b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterFactoryBean.java index 0cbff47592..92d9c43f89 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterFactoryBean.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterFactoryBean.java @@ -29,6 +29,7 @@ import org.springframework.integration.file.tail.FileTailingMessageProducerSuppo import org.springframework.integration.file.tail.OSDelegatingFileTailingMessageProducer; import org.springframework.scheduling.TaskScheduler; import org.springframework.util.Assert; +import org.springframework.util.StringUtils; /** * @author Gary Russell @@ -67,7 +68,9 @@ public class FileTailInboundChannelAdapterFactoryBean extends AbstractFactoryBea private volatile ApplicationEventPublisher applicationEventPublisher; public void setNativeOptions(String nativeOptions) { - this.nativeOptions = nativeOptions; + if (StringUtils.hasText(nativeOptions)) { + this.nativeOptions = nativeOptions; + } } public void setFile(File file) { @@ -82,11 +85,11 @@ public class FileTailInboundChannelAdapterFactoryBean extends AbstractFactoryBea this.taskScheduler = taskScheduler; } - public void setDelay(long delay) { + public void setDelay(Long delay) { this.delay = delay; } - public void setFileDelay(long fileDelay) { + public void setFileDelay(Long fileDelay) { this.fileDelay = fileDelay; } @@ -107,11 +110,11 @@ public class FileTailInboundChannelAdapterFactoryBean extends AbstractFactoryBea this.outputChannel = outputChannel; } - public void setAutoStartup(boolean autoStartup) { + public void setAutoStartup(Boolean autoStartup) { this.autoStartup = autoStartup; } - public void setPhase(int phase) { + public void setPhase(Integer phase) { this.phase = phase; } diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterParserTests-context.xml b/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterParserTests-context.xml index 60b7ebc22b..c6885c4861 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterParserTests-context.xml +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterParserTests-context.xml @@ -4,10 +4,14 @@ xmlns:int-file="http://www.springframework.org/schema/integration/file" xmlns:task="http://www.springframework.org/schema/task" xmlns:int="http://www.springframework.org/schema/integration" + xmlns:context="http://www.springframework.org/schema/context" + xmlns:util="http://www.springframework.org/schema/util" xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd http://www.springframework.org/schema/integration/file http://www.springframework.org/schema/integration/file/spring-integration-file.xsd http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd - http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd"> + http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd + http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util.xsd + http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd"> + + + + 2000 + + +