From 183941ec07ffe31c8fe2a6239e812ff293b5b113 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 21 Apr 2015 20:05:23 +0300 Subject: [PATCH] INT-3705: Improve Messaging Annotations handling JIRA: https://jira.spring.io/browse/INT-3705 * Add validation for annotation attributes which must be populated directly on the `MessageHandler` `@Bean` using setters * Add `sendTimeout()` annotation attribute for all Messaging Annotations * Remove `DEFAULT_SEND_TIMEOUT = 1000L` from `AbstractCorrelatingMessageHandler` to make its component consistent with all other `AbstractMessageProducingHandler` implementation. * Fix XSD docs and Reference Manual to say that default `send-timeout` for `AbstractMessageProducingHandler` components is `-1`, not `one second` * Make all Messaging Annotations attributes as `String` to allow to configure through the Property Placeholder options * Fix a couple of typos Polishing --- .../AbstractCorrelatingMessageHandler.java | 5 -- .../integration/annotation/Aggregator.java | 48 +++++++++++--- .../integration/annotation/Filter.java | 53 +++++++++++++--- .../integration/annotation/Router.java | 62 +++++++++++++++++-- .../annotation/ServiceActivator.java | 44 +++++++++++-- .../integration/annotation/Splitter.java | 46 ++++++++++++-- .../integration/annotation/Transformer.java | 36 +++++++++-- ...AbstractMethodAnnotationPostProcessor.java | 43 ++++++++++++- .../AggregatorAnnotationPostProcessor.java | 13 ++-- .../FilterAnnotationPostProcessor.java | 4 ++ .../RouterAnnotationPostProcessor.java | 13 ++-- ...rviceActivatorAnnotationPostProcessor.java | 3 + .../SplitterAnnotationPostProcessor.java | 4 +- .../TransformerAnnotationPostProcessor.java | 7 ++- .../config/xml/spring-integration-4.2.xsd | 9 ++- .../CorrelatingMessageHandlerTests.java | 9 ++- .../config/ResequencerParserTests.java | 4 +- .../annotation/AggregatorAnnotationTests.java | 6 +- ...ingAnnotationsWithBeanAnnotationTests.java | 37 ++++++++++- ...tatedEndpointWithCustomizedAggregator.java | 4 +- .../configuration/EnableIntegrationTests.java | 44 ++++++------- .../ip/tcp/TcpOutboundGateway.java | 4 +- src/reference/asciidoc/aggregator.adoc | 2 +- src/reference/asciidoc/claim-check.adoc | 4 +- src/reference/asciidoc/resequencer.adoc | 2 +- src/reference/asciidoc/syslog.adoc | 2 +- 26 files changed, 396 insertions(+), 112 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java index 5cb6243d19..45babced22 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java @@ -29,7 +29,6 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.aop.framework.ProxyFactory; -import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.beans.factory.DisposableBean; @@ -53,7 +52,6 @@ import org.springframework.integration.util.UUIDConverter; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageDeliveryException; -import org.springframework.messaging.core.DestinationResolutionException; import org.springframework.scheduling.TaskScheduler; import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; @@ -88,8 +86,6 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP private static final Log logger = LogFactory.getLog(AbstractCorrelatingMessageHandler.class); - public static final long DEFAULT_SEND_TIMEOUT = 1000L; - private final Comparator> sequenceNumberComparator = new SequenceNumberComparator(); private final Map> expireGroupScheduledFutures = new HashMap>(); @@ -141,7 +137,6 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP ? new HeaderAttributeCorrelationStrategy(IntegrationMessageHeaderAccessor.CORRELATION_ID) : correlationStrategy); this.releaseStrategy = releaseStrategy == null ? new SequenceSizeReleaseStrategy() : releaseStrategy; - setSendTimeout(DEFAULT_SEND_TIMEOUT); sequenceAware = this.releaseStrategy instanceof SequenceSizeReleaseStrategy; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/annotation/Aggregator.java b/spring-integration-core/src/main/java/org/springframework/integration/annotation/Aggregator.java index 0493cd28fb..32ada0787f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/annotation/Aggregator.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/annotation/Aggregator.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 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. @@ -22,8 +22,6 @@ import java.lang.annotation.Retention; import java.lang.annotation.RetentionPolicy; import java.lang.annotation.Target; -import org.springframework.integration.aggregator.AbstractCorrelatingMessageHandler; - /** * Indicates that a method is capable of aggregating messages. *

@@ -56,22 +54,51 @@ public @interface Aggregator { String discardChannel() default ""; /** + * Specify the maximum amount of time in milliseconds to wait when sending a reply + * {@link org.springframework.messaging.Message} to the {@link #outputChannel()}. + * Defaults to {@code -1} - blocking indefinitely. + * It is applied only if the output channel has some 'sending' limitations, e.g. + * {@link org.springframework.integration.channel.QueueChannel} with + * a fixed 'capacity' and is currently full. + * In this case a {@link org.springframework.messaging.MessageDeliveryException} is thrown. + * The 'sendTimeout' is ignored in case of + * {@link org.springframework.integration.channel.AbstractSubscribableChannel} implementations. + * Can be specified as 'property placeholder', e.g. {@code ${spring.integration.sendTimeout}}. * @return The timeout for sending results to the reply target (in milliseconds) */ - long sendTimeout() default AbstractCorrelatingMessageHandler.DEFAULT_SEND_TIMEOUT; + String sendTimeout() default ""; /** + * Specify whether messages that expired should be aggregated and sent to the {@link #outputChannel()} + * or {@code replyChannel} from message headers. Messages are expired when their containing + * {@link org.springframework.integration.store.MessageGroup} expires. One of the ways of expiring MessageGroups + * is by configuring a {@link org.springframework.integration.store.MessageGroupStoreReaper}. + * However MessageGroups can alternatively be expired by simply calling + * {@code MessageGroupStore.expireMessageGroup(groupId)}. That could be accomplished via a ControlBus operation + * or by simply invoking that method if you have a reference to the + * {@link org.springframework.integration.store.MessageGroupStore} instance. + * Defaults to {@code false}. + * * Can be specified as 'property placeholder', e.g. {@code ${spring.integration.sendPartialResultsOnExpiry}}. * @return Indicates whether to send an incomplete aggregate on expiry of the message group */ - boolean sendPartialResultsOnExpiry() default false; + String sendPartialResultsOnExpiry() default ""; - /* - {@code SmartLifecycle} options. - Can be specified as 'property placeholder', e.g. {@code ${foo.autoStartup}}. + /** + * The {@link org.springframework.context.SmartLifecycle} {@code autoStartup} option. + * Can be specified as 'property placeholder', e.g. {@code ${foo.autoStartup}}. + * Defaults to {@code true}. + * @return the auto startup {@code boolean} flag. */ - String autoStartup() default "true"; + String autoStartup() default ""; - String phase() default "0"; + /** + * Specify a {@link org.springframework.context.SmartLifecycle} {@code phase} option. + * Defaults {@code 0} for {@link org.springframework.integration.endpoint.PollingConsumer} + * and {@code Integer.MIN_VALUE} for {@link org.springframework.integration.endpoint.EventDrivenConsumer}. + * Can be specified as 'property placeholder', e.g. {@code ${foo.phase}}. + * @return the {@code SmartLifecycle} phase. + */ + String phase() default ""; /** * @return the {@link Poller} options for a polled endpoint @@ -80,4 +107,5 @@ public @interface Aggregator { * Only one {@link Poller} element is allowed. */ Poller[] poller() default {}; + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/annotation/Filter.java b/spring-integration-core/src/main/java/org/springframework/integration/annotation/Filter.java index bb0add30ac..06b744ebcd 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/annotation/Filter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/annotation/Filter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 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. @@ -51,19 +51,55 @@ public @interface Filter { String discardChannel() default ""; + /** + * Throw an exception if the filter rejects the message. + * Defaults to {@code false}. + * Can be specified as 'property placeholder', e.g. {@code ${spring.integration.throwExceptionOnRejection}}. + * @return the throw Exception on rejection flag. + */ String throwExceptionOnRejection() default ""; String[] adviceChain() default {}; - String discardWithinAdvice() default "true"; - - /* - {@code SmartLifecycle} options. - Can be specified as 'property placeholder', e.g. {@code ${foo.autoStartup}}. + /** + * When {@code true} (default) any discard action (and exception thrown) will occur + * within the scope of the advice class(es) in the chain. Otherwise, these actions + * will occur after the advice chain returns. + * Can be specified as 'property placeholder', e.g. {@code ${spring.integration.discardWithinAdvice}}. + * @return the discard within advice flag. */ - String autoStartup() default "true"; + String discardWithinAdvice() default ""; - String phase() default "0"; + /** + * Specify the maximum amount of time in milliseconds to wait when sending a reply + * {@link org.springframework.messaging.Message} to the {@link #outputChannel()}. + * Defaults to {@code -1} - blocking indefinitely. + * It is applied only if the output channel has some 'sending' limitations, e.g. + * {@link org.springframework.integration.channel.QueueChannel} with + * fixed a 'capacity'. In this case a {@link org.springframework.messaging.MessageDeliveryException} is thrown. + * The 'sendTimeout' is ignored in case of + * {@link org.springframework.integration.channel.AbstractSubscribableChannel} implementations. + * Can be specified as 'property placeholder', e.g. {@code ${spring.integration.sendTimeout}}. + * @return The timeout for sending results to the reply target (in milliseconds) + */ + String sendTimeout() default ""; + + /** + * The {@link org.springframework.context.SmartLifecycle} {@code autoStartup} option. + * Can be specified as 'property placeholder', e.g. {@code ${foo.autoStartup}}. + * Defaults to {@code true}. + * @return the auto startup {@code boolean} flag. + */ + String autoStartup() default ""; + + /** + * Specify a {@link org.springframework.context.SmartLifecycle} {@code phase} option. + * Defaults {@code 0} for {@link org.springframework.integration.endpoint.PollingConsumer} + * and {@code Integer.MIN_VALUE} for {@link org.springframework.integration.endpoint.EventDrivenConsumer}. + * Can be specified as 'property placeholder', e.g. {@code ${foo.phase}}. + * @return the {@code SmartLifecycle} phase. + */ + String phase() default ""; /** * @return the {@link Poller} options for a polled endpoint @@ -72,4 +108,5 @@ public @interface Filter { * Only one {@link Poller} element is allowed. */ Poller[] poller() default {}; + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/annotation/Router.java b/spring-integration-core/src/main/java/org/springframework/integration/annotation/Router.java index c688382ff2..db725b66a2 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/annotation/Router.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/annotation/Router.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 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. @@ -66,19 +66,68 @@ public @interface Router { String suffix() default ""; + /** + * Specify whether channel names must always be successfully resolved + * to existing channel instances. + *

If set to {@code true} (default), a {@link org.springframework.messaging.MessagingException} + * will be raised in case the channel cannot be resolved. Setting this attribute to {@code false}, + * will cause any unresolvable channels to be ignored. + * Can be specified as 'property placeholder', e.g. {@code ${spring.integration.resolutionRequired}}. + * @return the resolution required flag. + */ String resolutionRequired() default ""; + /** + * Specify whether sequence number and size headers should be added to each + * Message. Defaults to {@code false}. + * Can be specified as 'property placeholder', e.g. {@code ${spring.integration.applySequence}}. + * @return the apply sequence flag. + */ String applySequence() default ""; + /** + * If set to {@code true} , failures to send to a message channel will + * be ignored. If set to {@code false} (default), a {@link org.springframework.messaging.MessageDeliveryException} + * will be thrown instead, and if the router resolves more than one channel, + * any subsequent channels will not receive the message. + * Please be aware that when using direct channels (single threaded), + * send-failures can be caused by exceptions thrown by components + * much further down-stream. + * Can be specified as 'property placeholder', e.g. {@code ${spring.integration.ignoreSendFailures}}. + * @return the ignore send failures flag. + */ String ignoreSendFailures() default ""; - /* - {@code SmartLifecycle} options. - Can be specified as 'property placeholder', e.g. {@code ${foo.autoStartup}}. + /** + * Specify the maximum amount of time in milliseconds to wait when sending a reply + * {@link org.springframework.messaging.Message} to the {@code outputChannel}. + * Defaults to {@code -1} - blocking indefinitely. + * It is applied only if the output channel has some 'sending' limitations, e.g. + * {@link org.springframework.integration.channel.QueueChannel} with + * fixed a 'capacity'. In this case a {@link org.springframework.messaging.MessageDeliveryException} is thrown. + * The 'sendTimeout' is ignored in case of + * {@link org.springframework.integration.channel.AbstractSubscribableChannel} implementations. + * Can be specified as 'property placeholder', e.g. {@code ${spring.integration.sendTimeout}}. + * @return The timeout for sending results to the reply target (in milliseconds) */ - String autoStartup() default "true"; + String sendTimeout() default ""; - String phase() default "0"; + /** + * The {@link org.springframework.context.SmartLifecycle} {@code autoStartup} option. + * Can be specified as 'property placeholder', e.g. {@code ${foo.autoStartup}}. + * Defaults to {@code true}. + * @return the auto startup {@code boolean} flag. + */ + String autoStartup() default ""; + + /** + * Specify a {@link org.springframework.context.SmartLifecycle} {@code phase} option. + * Defaults {@code 0} for {@link org.springframework.integration.endpoint.PollingConsumer} + * and {@code Integer.MIN_VALUE} for {@link org.springframework.integration.endpoint.EventDrivenConsumer}. + * Can be specified as 'property placeholder', e.g. {@code ${foo.phase}}. + * @return the {@code SmartLifecycle} phase. + */ + String phase() default ""; /** * @return the {@link Poller} options for a polled endpoint @@ -87,4 +136,5 @@ public @interface Router { * Only one {@link Poller} element is allowed. */ Poller[] poller() default {}; + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/annotation/ServiceActivator.java b/spring-integration-core/src/main/java/org/springframework/integration/annotation/ServiceActivator.java index 4a612832ac..23b4a0b34f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/annotation/ServiceActivator.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/annotation/ServiceActivator.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 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. @@ -52,17 +52,48 @@ public @interface ServiceActivator { String outputChannel() default ""; + /** + * Specify whether the service method must return a non-null value. This value is + * {@code false} by default, but if set to {@code true}, a + * {@link org.springframework.integration.handler.ReplyRequiredException} will is thrown when + * the underlying service method (or expression) returns a null value. + * Can be specified as 'property placeholder', e.g. {@code ${spring.integration.requiresReply}}. + * @return the requires reply flag. + */ String requiresReply() default ""; String[] adviceChain() default {}; - /* - {@code SmartLifecycle} options. - Can be specified as 'property placeholder', e.g. {@code ${foo.autoStartup}}. + /** + * Specify the maximum amount of time in milliseconds to wait when sending a reply + * {@link org.springframework.messaging.Message} to the {@code outputChannel}. + * Defaults to {@code -1} - blocking indefinitely. + * It is applied only if the output channel has some 'sending' limitations, e.g. + * {@link org.springframework.integration.channel.QueueChannel} with + * fixed a 'capacity'. In this case a {@link org.springframework.messaging.MessageDeliveryException} is thrown. + * The 'sendTimeout' is ignored in case of + * {@link org.springframework.integration.channel.AbstractSubscribableChannel} implementations. + * Can be specified as 'property placeholder', e.g. {@code ${spring.integration.sendTimeout}}. + * @return The timeout for sending results to the reply target (in milliseconds) */ - String autoStartup() default "true"; + String sendTimeout() default ""; - String phase() default "0"; + /** + * The {@link org.springframework.context.SmartLifecycle} {@code autoStartup} option. + * Can be specified as 'property placeholder', e.g. {@code ${foo.autoStartup}}. + * Defaults to {@code true}. + * @return the auto startup {@code boolean} flag. + */ + String autoStartup() default ""; + + /** + * Specify a {@link org.springframework.context.SmartLifecycle} {@code phase} option. + * Defaults {@code 0} for {@link org.springframework.integration.endpoint.PollingConsumer} + * and {@code Integer.MIN_VALUE} for {@link org.springframework.integration.endpoint.EventDrivenConsumer}. + * Can be specified as 'property placeholder', e.g. {@code ${foo.phase}}. + * @return the {@code SmartLifecycle} phase. + */ + String phase() default ""; /** * @return the {@link Poller} options for a polled endpoint @@ -71,4 +102,5 @@ public @interface ServiceActivator { * Only one {@link Poller} element is allowed. */ Poller[] poller() default {}; + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/annotation/Splitter.java b/spring-integration-core/src/main/java/org/springframework/integration/annotation/Splitter.java index 3c5e4080f3..2b4b15efb5 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/annotation/Splitter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/annotation/Splitter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 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. @@ -51,17 +51,50 @@ public @interface Splitter { String outputChannel() default ""; + /** + * Set this flag to {@code false} to prevent adding sequence related headers in this splitter. + * This can be convenient in cases where the set sequence numbers conflict with downstream + * custom aggregations. When {@code true}, existing correlation and sequence related headers + * are pushed onto a stack; downstream components, such as aggregators may pop + * the stack to revert the existing headers after aggregation. + * Defaults to {@code true}. + * Can be specified as 'property placeholder', e.g. {@code ${spring.integration.applySequence}}. + * @return the apply sequence flag. + */ String applySequence() default ""; String[] adviceChain() default {}; - /* - {@code SmartLifecycle} options. - Can be specified as 'property placeholder', e.g. {@code ${foo.autoStartup}}. + /** + * Specify the maximum amount of time in milliseconds to wait when sending a reply + * {@link org.springframework.messaging.Message} to the {@code outputChannel}. + * Defaults to {@code -1} - blocking indefinitely. + * It is applied only if the output channel has some 'sending' limitations, e.g. + * {@link org.springframework.integration.channel.QueueChannel} with + * fixed a 'capacity'. In this case a {@link org.springframework.messaging.MessageDeliveryException} is thrown. + * The 'sendTimeout' is ignored in case of + * {@link org.springframework.integration.channel.AbstractSubscribableChannel} implementations. + * Can be specified as 'property placeholder', e.g. {@code ${spring.integration.sendTimeout}}. + * @return The timeout for sending results to the reply target (in milliseconds) */ - String autoStartup() default "true"; + String sendTimeout() default ""; - String phase() default "0"; + /** + * The {@link org.springframework.context.SmartLifecycle} {@code autoStartup} option. + * Can be specified as 'property placeholder', e.g. {@code ${foo.autoStartup}}. + * Defaults to {@code true}. + * @return the auto startup {@code boolean} flag. + */ + String autoStartup() default ""; + + /** + * Specify a {@link org.springframework.context.SmartLifecycle} {@code phase} option. + * Defaults {@code 0} for {@link org.springframework.integration.endpoint.PollingConsumer} + * and {@code Integer.MIN_VALUE} for {@link org.springframework.integration.endpoint.EventDrivenConsumer}. + * Can be specified as 'property placeholder', e.g. {@code ${foo.phase}}. + * @return the {@code SmartLifecycle} phase. + */ + String phase() default ""; /** * @return the {@link Poller} options for a polled endpoint @@ -70,4 +103,5 @@ public @interface Splitter { * Only one {@link Poller} element is allowed. */ Poller[] poller() default {}; + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/annotation/Transformer.java b/spring-integration-core/src/main/java/org/springframework/integration/annotation/Transformer.java index b0c8bd17e4..9940690a83 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/annotation/Transformer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/annotation/Transformer.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 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. @@ -43,13 +43,36 @@ public @interface Transformer { String[] adviceChain() default {}; - /* - {@code SmartLifecycle} options. - Can be specified as 'property placeholder', e.g. {@code ${foo.autoStartup}}. + /** + * Specify the maximum amount of time in milliseconds to wait when sending a reply + * {@link org.springframework.messaging.Message} to the {@code outputChannel}. + * Defaults to {@code -1} - blocking indefinitely. + * It is applied only if the output channel has some 'sending' limitations, e.g. + * {@link org.springframework.integration.channel.QueueChannel} with + * fixed a 'capacity'. In this case a {@link org.springframework.messaging.MessageDeliveryException} is thrown. + * The 'sendTimeout' is ignored in case of + * {@link org.springframework.integration.channel.AbstractSubscribableChannel} implementations. + * Can be specified as 'property placeholder', e.g. {@code ${spring.integration.sendTimeout}}. + * @return The timeout for sending results to the reply target (in milliseconds) */ - String autoStartup() default "true"; + String sendTimeout() default ""; - String phase() default "0"; + /** + * The {@link org.springframework.context.SmartLifecycle} {@code autoStartup} option. + * Can be specified as 'property placeholder', e.g. {@code ${foo.autoStartup}}. + * Defaults to {@code true}. + * @return the auto startup {@code boolean} flag. + */ + String autoStartup() default ""; + + /** + * Specify a {@link org.springframework.context.SmartLifecycle} {@code phase} option. + * Defaults {@code 0} for {@link org.springframework.integration.endpoint.PollingConsumer} + * and {@code Integer.MIN_VALUE} for {@link org.springframework.integration.endpoint.EventDrivenConsumer}. + * Can be specified as 'property placeholder', e.g. {@code ${foo.phase}}. + * @return the {@code SmartLifecycle} phase. + */ + String phase() default ""; /** * @return the {@link Poller} options for a polled endpoint @@ -58,4 +81,5 @@ public @interface Transformer { * Only one {@link Poller} element is allowed. */ Poller[] poller() default {}; + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java index 707da67b59..60302199e6 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java @@ -32,6 +32,7 @@ import org.springframework.aop.support.DefaultBeanFactoryPointcutAdvisor; import org.springframework.aop.support.NameMatchMethodPointcut; import org.springframework.beans.factory.ListableBeanFactory; import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; +import org.springframework.beans.factory.support.BeanDefinitionValidationException; import org.springframework.context.annotation.Bean; import org.springframework.core.GenericTypeResolver; import org.springframework.core.annotation.AnnotatedElementUtils; @@ -51,7 +52,9 @@ import org.springframework.integration.endpoint.AbstractPollingEndpoint; import org.springframework.integration.endpoint.EventDrivenConsumer; import org.springframework.integration.endpoint.PollingConsumer; import org.springframework.integration.endpoint.SourcePollingChannelAdapter; +import org.springframework.integration.handler.AbstractMessageProducingHandler; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; +import org.springframework.integration.router.AbstractMessageRouter; import org.springframework.integration.scheduling.PollerMetadata; import org.springframework.integration.support.channel.BeanFactoryChannelResolver; import org.springframework.integration.util.MessagingAnnotationUtils; @@ -82,6 +85,9 @@ public abstract class AbstractMethodAnnotationPostProcessor messageHandlerAttributes = new ArrayList(); protected final ConfigurableListableBeanFactory beanFactory; @@ -98,6 +104,7 @@ public abstract class AbstractMethodAnnotationPostProcessor annotations) { + for (String attribute : this.messageHandlerAttributes) { + for (Annotation annotation : annotations) { + Object value = AnnotationUtils.getValue(annotation, attribute); + if (MessagingAnnotationUtils.hasValue(value)) { + throw new BeanDefinitionValidationException("The MessageHandler [" + handlerBeanName + "] can not" + + " be populated because of ambiguity with annotation attributes " + + this.messageHandlerAttributes + " which are not allowed when an integration annotation " + + "is used with a @Bean definition for a MessageHandler." + + "\nThe attribute causing the ambiguity is: [" + attribute + "]." + + "\nUse the appropriate setter on the MessageHandler directly when configfuring an " + + "endpoint this way."); + } + } + } + } + /** * Subclasses must implement this method to create the MessageHandler. * diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AggregatorAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AggregatorAnnotationPostProcessor.java index fa2bfd4dbb..d78c905bb4 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AggregatorAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AggregatorAnnotationPostProcessor.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 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. @@ -77,14 +77,11 @@ public class AggregatorAnnotationPostProcessor extends AbstractMethodAnnotationP if (StringUtils.hasText(outputChannelName)) { handler.setOutputChannelName(outputChannelName); } - Long sendTimeout = MessagingAnnotationUtils.resolveAttribute(annotations, "sendTimeout", Long.class); - if (sendTimeout != null) { - handler.setSendTimeout(sendTimeout); - } - Boolean sendPartialResultsOnExpiry = MessagingAnnotationUtils.resolveAttribute(annotations, - "sendPartialResultsOnExpiry", Boolean.class); + String sendPartialResultsOnExpiry = MessagingAnnotationUtils.resolveAttribute(annotations, + "sendPartialResultsOnExpiry", String.class); if (sendPartialResultsOnExpiry != null) { - handler.setSendPartialResultOnExpiry(sendPartialResultsOnExpiry); + handler.setSendPartialResultOnExpiry( + Boolean.parseBoolean(this.environment.resolvePlaceholders(sendPartialResultsOnExpiry))); } handler.setBeanFactory(this.beanFactory); return handler; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/FilterAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/FilterAnnotationPostProcessor.java index cd7566b710..22726bdde6 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/FilterAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/FilterAnnotationPostProcessor.java @@ -18,6 +18,7 @@ package org.springframework.integration.config.annotation; import java.lang.annotation.Annotation; import java.lang.reflect.Method; +import java.util.Arrays; import java.util.List; import org.springframework.beans.factory.ListableBeanFactory; @@ -45,6 +46,8 @@ public class FilterAnnotationPostProcessor extends AbstractMethodAnnotationPostP public FilterAnnotationPostProcessor(ListableBeanFactory beanFactory, Environment environment) { super(beanFactory, environment); + this.messageHandlerAttributes.addAll(Arrays.asList("discardChannel", "throwExceptionOnRejection", + "adviceChain", "discardWithinAdvice")); } @@ -57,6 +60,7 @@ public class FilterAnnotationPostProcessor extends AbstractMethodAnnotationPostP selector = (MessageSelector) target; } else if (this.extractTypeIfPossible(target, MessageFilter.class) != null) { + checkMessageHandlerAttributes(resolveTargetBeanName(method), annotations); return (MessageHandler) target; } else { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/RouterAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/RouterAnnotationPostProcessor.java index bb20537b33..755e905d63 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/RouterAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/RouterAnnotationPostProcessor.java @@ -18,6 +18,7 @@ package org.springframework.integration.config.annotation; import java.lang.annotation.Annotation; import java.lang.reflect.Method; +import java.util.Arrays; import java.util.List; import java.util.Properties; @@ -46,9 +47,10 @@ public class RouterAnnotationPostProcessor extends AbstractMethodAnnotationPostP public RouterAnnotationPostProcessor(ListableBeanFactory beanFactory, Environment environment) { super(beanFactory, environment); + this.messageHandlerAttributes.addAll(Arrays.asList("defaultOutputChannel", "applySequence", + "ignoreSendFailures", "resolutionRequired", "channelMappings", "prefix", "suffix")); } - @Override protected MessageHandler createHandler(Object bean, Method method, List annotations) { AbstractMessageRouter router; @@ -57,10 +59,10 @@ public class RouterAnnotationPostProcessor extends AbstractMethodAnnotationPostP router = this.extractTypeIfPossible(target, AbstractMessageRouter.class); if (router == null) { if (target instanceof MessageHandler) { - Assert.isTrue(this.routerAttributesProvided(annotations), "'defaultOutputChannel', " - + "'applySequence', 'ignoreSendFailures', 'resolutionRequired' and 'channelMappings' " - + "can be applied to 'AbstractMessageRouter' implementations, but target handler is: " - + target.getClass()); + Assert.isTrue(this.routerAttributesProvided(annotations), "'defaultOutputChannel', 'applySequence', " + + "'ignoreSendFailures', 'resolutionRequired', 'channelMappings', 'prefix' and 'suffix' " + + "can be applied to 'AbstractMessageRouter' implementations, but target handler is: " + + target.getClass()); return (MessageHandler) target; } else { @@ -68,6 +70,7 @@ public class RouterAnnotationPostProcessor extends AbstractMethodAnnotationPostP } } else { + checkMessageHandlerAttributes(resolveTargetBeanName(method), annotations); return router; } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/ServiceActivatorAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/ServiceActivatorAnnotationPostProcessor.java index 4abaf436d7..4663505c40 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/ServiceActivatorAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/ServiceActivatorAnnotationPostProcessor.java @@ -18,6 +18,7 @@ package org.springframework.integration.config.annotation; import java.lang.annotation.Annotation; import java.lang.reflect.Method; +import java.util.Arrays; import java.util.List; import org.springframework.beans.factory.ListableBeanFactory; @@ -44,6 +45,7 @@ public class ServiceActivatorAnnotationPostProcessor extends AbstractMethodAnnot public ServiceActivatorAnnotationPostProcessor(ListableBeanFactory beanFactory, Environment environment) { super(beanFactory, environment); + this.messageHandlerAttributes.addAll(Arrays.asList("outputChannel", "requiresReply", "adviceChain")); } @@ -66,6 +68,7 @@ public class ServiceActivatorAnnotationPostProcessor extends AbstractMethodAnnot } } else { + checkMessageHandlerAttributes(resolveTargetBeanName(method), annotations); return (MessageHandler) target; } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/SplitterAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/SplitterAnnotationPostProcessor.java index 67e8965da6..ce565ea956 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/SplitterAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/SplitterAnnotationPostProcessor.java @@ -18,6 +18,7 @@ package org.springframework.integration.config.annotation; import java.lang.annotation.Annotation; import java.lang.reflect.Method; +import java.util.Arrays; import java.util.List; import org.springframework.beans.factory.ListableBeanFactory; @@ -43,9 +44,9 @@ public class SplitterAnnotationPostProcessor extends AbstractMethodAnnotationPos public SplitterAnnotationPostProcessor(ListableBeanFactory beanFactory, Environment environment) { super(beanFactory, environment); + this.messageHandlerAttributes.addAll(Arrays.asList("outputChannel", "applySequence", "adviceChain")); } - @Override protected MessageHandler createHandler(Object bean, Method method, List annotations) { String applySequence = MessagingAnnotationUtils.resolveAttribute(annotations, "applySequence", String.class); @@ -65,6 +66,7 @@ public class SplitterAnnotationPostProcessor extends AbstractMethodAnnotationPos } } else { + checkMessageHandlerAttributes(resolveTargetBeanName(method), annotations); return splitter; } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/TransformerAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/TransformerAnnotationPostProcessor.java index d1f431595c..cd93963700 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/TransformerAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/TransformerAnnotationPostProcessor.java @@ -18,6 +18,7 @@ package org.springframework.integration.config.annotation; import java.lang.annotation.Annotation; import java.lang.reflect.Method; +import java.util.Arrays; import java.util.List; import org.springframework.beans.factory.ListableBeanFactory; @@ -41,17 +42,19 @@ public class TransformerAnnotationPostProcessor extends AbstractMethodAnnotation public TransformerAnnotationPostProcessor(ListableBeanFactory beanFactory, Environment environment) { super(beanFactory, environment); + this.messageHandlerAttributes.addAll(Arrays.asList("outputChannel", "adviceChain")); } - @Override protected MessageHandler createHandler(Object bean, Method method, List annotations) { org.springframework.integration.transformer.Transformer transformer; if (AnnotatedElementUtils.isAnnotated(method, Bean.class.getName())) { Object target = this.resolveTargetBeanFromMethodWithBeanAnnotation(method); - transformer = this.extractTypeIfPossible(target, org.springframework.integration.transformer.Transformer.class); + transformer = this.extractTypeIfPossible(target, + org.springframework.integration.transformer.Transformer.class); if (transformer == null) { if (this.extractTypeIfPossible(target, AbstractReplyProducingMessageHandler.class) != null) { + checkMessageHandlerAttributes(resolveTargetBeanName(method), annotations); return (MessageHandler) target; } transformer = new MethodInvokingTransformer(target); diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.2.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.2.xsd index f6a39186ea..3c8df9712f 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.2.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.2.xsd @@ -933,8 +933,11 @@ - Maximum amount of time in milliseconds to wait when sending a message to the channel if such channel may block. - For example, a Queue Channel can block until space is available if its maximum capacity has been reached. + Maximum amount of time in milliseconds to wait when sending a message to the channel + if such channel may block. + For example, a Queue Channel can block until space is available if its maximum capacity + has been reached. + Defaults to '-1' - blocking indefinitely. @@ -4596,7 +4599,7 @@ The list of component name patterns you want to track (e.g., tracked-components Specify the maximum amount of time in milliseconds to wait when sending a reply - Message to the output channel. By default the send will block for one second. + Message to the output channel. Defaults to '-1' - blocking indefinitely. It is applied only if the output channel has some 'sending' limitations, e.g. QueueChannel with fixed a 'capacity'. In this case a MessageDeliveryException is thrown. The 'send-timeout' is ignored in case of AbstractSubscribableChannel implementations. diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java index 59032e1f44..420ab45d35 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2012 the original author or authors. + * Copyright 2002-2014 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. @@ -20,7 +20,6 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; import static org.mockito.Matchers.any; -import static org.mockito.Matchers.anyLong; import static org.mockito.Matchers.isA; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.verify; @@ -39,7 +38,6 @@ import org.mockito.Mock; import org.mockito.internal.stubbing.answers.ThrowsException; import org.mockito.runners.MockitoJUnitRunner; -import org.springframework.messaging.MessageHandlingException; import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.MessageGroupStore; import org.springframework.integration.store.SimpleMessageGroup; @@ -47,6 +45,7 @@ import org.springframework.integration.store.SimpleMessageStore; import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHandlingException; /** * @author Iwein Fuld @@ -86,7 +85,7 @@ public class CorrelatingMessageHandlerTests { when(correlationStrategy.getCorrelationKey(isA(Message.class))).thenReturn(correlationKey); when(processor.processMessageGroup(any(MessageGroup.class))).thenReturn(MessageBuilder.withPayload("grouped").build()); - when(outputChannel.send(any(Message.class), anyLong())).thenReturn(true); + when(outputChannel.send(any(Message.class))).thenReturn(true); handler.handleMessage(message1); @@ -142,7 +141,7 @@ public class CorrelatingMessageHandlerTests { when(correlationStrategy.getCorrelationKey(isA(Message.class))).thenReturn(correlationKey); when(processor.processMessageGroup(any(MessageGroup.class))).thenReturn(MessageBuilder.withPayload("grouped").build()); - when(outputChannel.send(any(Message.class), anyLong())).thenReturn(true); + when(outputChannel.send(any(Message.class))).thenReturn(true); handler.handleMessage(message1); bothMessagesHandled.countDown(); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/ResequencerParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/ResequencerParserTests.java index ad97168df4..a105d27a4f 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/ResequencerParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/ResequencerParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 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. You may obtain a copy of the License at @@ -65,7 +65,7 @@ public class ResequencerParserTests { ResequencingMessageHandler.class); assertNull(getPropertyValue(resequencer, "outputChannel")); assertTrue(getPropertyValue(resequencer, "discardChannel") instanceof NullChannel); - assertEquals("The ResequencerEndpoint is not set with the appropriate timeout value", 1000l, getPropertyValue( + assertEquals("The ResequencerEndpoint is not set with the appropriate timeout value", -1L, getPropertyValue( resequencer, "messagingTemplate.sendTimeout")); assertEquals( "The ResequencerEndpoint is not configured with the appropriate 'send partial results on timeout' flag", diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java index b482fd6f55..89d1b0b650 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 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. @@ -31,7 +31,6 @@ import org.junit.Test; import org.springframework.beans.DirectFieldAccessor; import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; -import org.springframework.integration.aggregator.AggregatingMessageHandler; import org.springframework.integration.aggregator.MethodInvokingCorrelationStrategy; import org.springframework.integration.aggregator.MethodInvokingReleaseStrategy; import org.springframework.integration.aggregator.SequenceSizeReleaseStrategy; @@ -56,8 +55,7 @@ public class AggregatorAnnotationTests { assertTrue(getPropertyValue(aggregator, "releaseStrategy") instanceof SequenceSizeReleaseStrategy); assertNull(getPropertyValue(aggregator, "outputChannel")); assertTrue(getPropertyValue(aggregator, "discardChannel") instanceof NullChannel); - assertEquals(AggregatingMessageHandler.DEFAULT_SEND_TIMEOUT, getPropertyValue(aggregator, - "messagingTemplate.sendTimeout")); + assertEquals(-1L, getPropertyValue(aggregator, "messagingTemplate.sendTimeout")); assertEquals(false, getPropertyValue(aggregator, "sendPartialResultOnExpiry")); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationsWithBeanAnnotationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationsWithBeanAnnotationTests.java index 879d84c3cf..6404513f4b 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationsWithBeanAnnotationTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationsWithBeanAnnotationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2014 the original author or authors. + * Copyright 2014-2015 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. @@ -16,10 +16,13 @@ package org.springframework.integration.config.annotation; +import static org.hamcrest.Matchers.containsString; +import static org.hamcrest.Matchers.instanceOf; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; import java.util.ArrayList; import java.util.List; @@ -31,7 +34,10 @@ import org.hamcrest.Matchers; import org.junit.Test; import org.junit.runner.RunWith; +import org.springframework.beans.factory.BeanCreationException; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.support.BeanDefinitionValidationException; +import org.springframework.context.annotation.AnnotationConfigApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.expression.ExpressionParser; @@ -67,13 +73,12 @@ import org.springframework.messaging.support.GenericMessage; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; -import org.springframework.test.context.support.AnnotationConfigContextLoader; /** * @author Artem Bilan * @since 4.0 */ -@ContextConfiguration(loader = AnnotationConfigContextLoader.class) +@ContextConfiguration(classes = MessagingAnnotationsWithBeanAnnotationTests.ContextConfiguration.class) @RunWith(SpringJUnit4ClassRunner.class) @DirtiesContext public class MessagingAnnotationsWithBeanAnnotationTests { @@ -109,6 +114,19 @@ public class MessagingAnnotationsWithBeanAnnotationTests { } } + @Test + public void testInvalidMessagingAnnotationsConfig() { + try { + new AnnotationConfigApplicationContext(InvalidContextConfiguration.class); + fail("BeanCreationException expected"); + } + catch (Exception e) { + assertThat(e, instanceOf(BeanCreationException.class)); + assertThat(e.getCause(), instanceOf(BeanDefinitionValidationException.class)); + assertThat(e.getMessage(), containsString("The attribute causing the ambiguity is: [applySequence].")); + } + } + @Configuration @EnableIntegration @EnableMessageHistory @@ -220,4 +238,17 @@ public class MessagingAnnotationsWithBeanAnnotationTests { } + @Configuration + @EnableIntegration + static class InvalidContextConfiguration { + + @Bean + @Splitter(inputChannel = "splitterChannel", applySequence = "false") + public MessageHandler splitter() { + return new DefaultMessageSplitter(); + } + + } + + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/TestAnnotatedEndpointWithCustomizedAggregator.java b/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/TestAnnotatedEndpointWithCustomizedAggregator.java index b48d47db45..8c8bea0713 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/TestAnnotatedEndpointWithCustomizedAggregator.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/TestAnnotatedEndpointWithCustomizedAggregator.java @@ -41,8 +41,8 @@ public class TestAnnotatedEndpointWithCustomizedAggregator { inputChannel = "inputChannel", outputChannel = "outputChannel", discardChannel = "discardChannel", - sendPartialResultsOnExpiry = true, - sendTimeout = 98765432) + sendPartialResultsOnExpiry = "true", + sendTimeout = "98765432") public Message aggregatingMethod(List> messages) { List> sortableList = new ArrayList>(messages); Collections.sort(sortableList, new MessageSequenceComparator()); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java index a494618d80..c6a23f810b 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java @@ -531,7 +531,7 @@ public class EnableIntegrationTests { assertEquals("annOutput", TestUtils.getPropertyValue(consumer, "handler.outputChannelName")); assertEquals("annOutput", TestUtils.getPropertyValue(consumer, "handler.discardChannelName")); assertEquals(1000L, TestUtils.getPropertyValue(consumer, "trigger.period")); - assertEquals(1000L, TestUtils.getPropertyValue(consumer, "handler.messagingTemplate.sendTimeout")); + assertEquals(-1L, TestUtils.getPropertyValue(consumer, "handler.messagingTemplate.sendTimeout")); assertFalse(TestUtils.getPropertyValue(consumer, "handler.sendPartialResultOnExpiry", Boolean.class)); consumer = this.context.getBean("annotationTestService.annAgg2.aggregator", PollingConsumer.class); @@ -1356,7 +1356,7 @@ public class EnableIntegrationTests { outputChannel = "annOutput", adviceChain = {"annAdvice"}, poller = @Poller(fixedDelay = "1000")) - public static @interface MyServiceActivator { + public @interface MyServiceActivator { String inputChannel() default ""; @@ -1374,7 +1374,7 @@ public class EnableIntegrationTests { @Target({ElementType.METHOD, ElementType.ANNOTATION_TYPE}) @Retention(RetentionPolicy.RUNTIME) @MyServiceActivator - public static @interface MyServiceActivator1 { + public @interface MyServiceActivator1 { String inputChannel() default ""; @@ -1392,7 +1392,7 @@ public class EnableIntegrationTests { @Target({ElementType.METHOD, ElementType.ANNOTATION_TYPE}) @Retention(RetentionPolicy.RUNTIME) @MyServiceActivator1 - public static @interface MyServiceActivator2 { + public @interface MyServiceActivator2 { String inputChannel() default ""; @@ -1401,7 +1401,7 @@ public class EnableIntegrationTests { @Target({ElementType.METHOD, ElementType.ANNOTATION_TYPE}) @Retention(RetentionPolicy.RUNTIME) @MyServiceActivator2 - public static @interface MyServiceActivator3 { + public @interface MyServiceActivator3 { String inputChannel() default ""; @@ -1410,7 +1410,7 @@ public class EnableIntegrationTests { @Target({ElementType.METHOD, ElementType.ANNOTATION_TYPE}) @Retention(RetentionPolicy.RUNTIME) @MyServiceActivator3(inputChannel = "annInput3") - public static @interface MyServiceActivator4 { + public @interface MyServiceActivator4 { String inputChannel() default ""; @@ -1419,7 +1419,7 @@ public class EnableIntegrationTests { @Target({ElementType.METHOD, ElementType.ANNOTATION_TYPE}) @Retention(RetentionPolicy.RUNTIME) @MyServiceActivator4 - public static @interface MyServiceActivator5 { + public @interface MyServiceActivator5 { String inputChannel() default ""; @@ -1430,7 +1430,7 @@ public class EnableIntegrationTests { @Target({ElementType.METHOD, ElementType.ANNOTATION_TYPE}) @Retention(RetentionPolicy.RUNTIME) @MyServiceActivator5 - public static @interface MyServiceActivator6 { + public @interface MyServiceActivator6 { String inputChannel() default ""; @@ -1439,7 +1439,7 @@ public class EnableIntegrationTests { @Target({ElementType.METHOD, ElementType.ANNOTATION_TYPE}) @Retention(RetentionPolicy.RUNTIME) @MyServiceActivator8 - public static @interface MyServiceActivator7 { + public @interface MyServiceActivator7 { String inputChannel() default ""; @@ -1448,7 +1448,7 @@ public class EnableIntegrationTests { @Target({ElementType.METHOD, ElementType.ANNOTATION_TYPE}) @Retention(RetentionPolicy.RUNTIME) @MyServiceActivator7 - public static @interface MyServiceActivator8 { + public @interface MyServiceActivator8 { String inputChannel() default ""; @@ -1463,7 +1463,7 @@ public class EnableIntegrationTests { outputChannel = "annOutput", adviceChain = {"annAdvice"}, poller = @Poller(fixedDelay = "1000")) - public static @interface MyServiceActivatorNoLocalAtts { + public @interface MyServiceActivatorNoLocalAtts { } @Target(ElementType.METHOD) @@ -1474,7 +1474,7 @@ public class EnableIntegrationTests { outputChannel = "annOutput", discardChannel = "annOutput", poller = @Poller(fixedDelay = "1000")) - public static @interface MyAggregator { + public @interface MyAggregator { String inputChannel() default ""; @@ -1482,7 +1482,7 @@ public class EnableIntegrationTests { String discardChannel() default ""; - long sendTimeout() default AbstractCorrelatingMessageHandler.DEFAULT_SEND_TIMEOUT; + long sendTimeout() default 1000L; boolean sendPartialResultsOnExpiry() default false; @@ -1500,21 +1500,21 @@ public class EnableIntegrationTests { inputChannel = "annInput", outputChannel = "annOutput", discardChannel = "annOutput", - sendPartialResultsOnExpiry = false, - sendTimeout = 1000L, + sendPartialResultsOnExpiry = "false", + sendTimeout = "1000", poller = @Poller(fixedDelay = "1000")) - public static @interface MyAggregatorDefaultOverrideDefaults { + public @interface MyAggregatorDefaultOverrideDefaults { - boolean sendPartialResultsOnExpiry() default true; + String sendPartialResultsOnExpiry() default "true"; - long sendTimeout() default 75; + String sendTimeout() default "75"; } @Target({ElementType.METHOD, ElementType.ANNOTATION_TYPE}) @Retention(RetentionPolicy.RUNTIME) @InboundChannelAdapter(value = "counterChannel", autoStartup = "false", phase = "23") - public static @interface MyInboundChannelAdapter { + public @interface MyInboundChannelAdapter { String value() default ""; @@ -1529,14 +1529,14 @@ public class EnableIntegrationTests { @Target(ElementType.METHOD) @Retention(RetentionPolicy.RUNTIME) @MyInboundChannelAdapter - public static @interface MyInboundChannelAdapter1 { + public @interface MyInboundChannelAdapter1 { } @Target(ElementType.METHOD) @Retention(RetentionPolicy.RUNTIME) @BridgeFrom(value = "metaBridgeInput", autoStartup = "false") - public static @interface MyBridgeFrom { + public @interface MyBridgeFrom { String value() default ""; } @@ -1544,7 +1544,7 @@ public class EnableIntegrationTests { @Target(ElementType.METHOD) @Retention(RetentionPolicy.RUNTIME) @BridgeTo(autoStartup = "false") - public static @interface MyBridgeTo { + public @interface MyBridgeTo { } // Error because the annotation is on a class; it must be on an interface diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java index 7b8ef93105..56acaeb2e1 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java @@ -134,7 +134,7 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler throw new MessageTimeoutException(requestMessage, "Timed out waiting for response"); } if (logger.isDebugEnabled()) { - logger.debug("Respose " + replyMessage); + logger.debug("Response " + replyMessage); } return replyMessage; } @@ -301,7 +301,7 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler /** * We have a race condition when a socket is closed right after the reply is received. The close "error" * might arrive before the actual reply. Overwrite an error with a good reply, but not vice-versa. - * @param reply + * @param reply the reply message. */ public void setReply(Message reply) { if (this.reply == null) { diff --git a/src/reference/asciidoc/aggregator.adoc b/src/reference/asciidoc/aggregator.adoc index 9eb340fd89..88578084e6 100644 --- a/src/reference/asciidoc/aggregator.adoc +++ b/src/reference/asciidoc/aggregator.adoc @@ -328,7 +328,7 @@ _Default - 'false'_. <9> The timeout interval to wait when sending a reply `Message` to the `output-channel` or `discard-channel`. -By default the send will block for one second. +Defaults to `-1` - blocking indefinitely. It is applied only if the output channel has some 'sending' limitations, e.g. `QueueChannel` with a fixed 'capacity'. In this case a `MessageDeliveryException` is thrown. diff --git a/src/reference/asciidoc/claim-check.adoc b/src/reference/asciidoc/claim-check.adoc index eb47e355e6..5b5f7c4d14 100644 --- a/src/reference/asciidoc/claim-check.adoc +++ b/src/reference/asciidoc/claim-check.adoc @@ -93,7 +93,7 @@ _Optional_. <7> Specify the maximum amount of time in milliseconds to wait when sending a reply Message to the output channel. -By default the send will block for one second. +Defaults to `-1` - blocking indefinitely. Attribute is not available inside a `Chain` element. _Optional_. @@ -174,7 +174,7 @@ _Optional_. <8> Specify the maximum amount of time in milliseconds to wait when sending a reply Message to the output channel. -By default the send will block for one second. +Defaults to `-1` - blocking indefinitely. Attribute is not available inside a `Chain` element. _Optional_. diff --git a/src/reference/asciidoc/resequencer.adoc b/src/reference/asciidoc/resequencer.adoc index e3dba9eb33..c870502280 100644 --- a/src/reference/asciidoc/resequencer.adoc +++ b/src/reference/asciidoc/resequencer.adoc @@ -83,7 +83,7 @@ See <>. <8> The timeout interval to wait when sending a reply `Message` to the `output-channel` or `discard-channel`. -By default the send will block for one second. +Defaults to `-1` - blocking indefinitely. It is applied only if the output channel has some 'sending' limitations, e.g. `QueueChannel` with a fixed 'capacity'. In this case a `MessageDeliveryException` is thrown. diff --git a/src/reference/asciidoc/syslog.adoc b/src/reference/asciidoc/syslog.adoc index 4cea2fe714..a1af1810d9 100644 --- a/src/reference/asciidoc/syslog.adoc +++ b/src/reference/asciidoc/syslog.adoc @@ -77,7 +77,7 @@ This attribute can contain `udp` or `tcp`; it defaults to `udp`. ---- A `UDP` adapter that sends messages to channel `fromSyslog`. -It also shows the `SmartLifecyle` attributes `auto-startup` and `phase`. +It also shows the `SmartLifecycle` attributes `auto-startup` and `phase`. It has a reference to a custom `org.springframework.integration.syslog.MessageConverter` with id `converter` and an `error-channel`. Also notice the `udp-attributes` child element. You can set various UDP attributes here, as defined in <>.