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

INT-3705: Improve Messaging Annotations handling

Browse Source 
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
This commit is contained in:
Artem Bilan
2015-04-21 20:05:23 +03:00
committed by Gary Russell
parent 5bf4d97b38
commit 183941ec07
26 changed files with 396 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java
View File

@@ -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<Message<?>> sequenceNumberComparator = new SequenceNumberComparator();
private final Map<UUID, ScheduledFuture<?>> expireGroupScheduledFutures = new HashMap<UUID, ScheduledFuture<?>>();
@@ -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;
}

48
spring-integration-core/src/main/java/org/springframework/integration/annotation/Aggregator.java
View File

@@ -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.
* <p>
@@ -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 {};
}

53
spring-integration-core/src/main/java/org/springframework/integration/annotation/Filter.java
View File

@@ -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 {};
}

62
spring-integration-core/src/main/java/org/springframework/integration/annotation/Router.java
View File

@@ -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.
* <p> 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 {};
}

44
spring-integration-core/src/main/java/org/springframework/integration/annotation/ServiceActivator.java
View File

@@ -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 {};
}

46
spring-integration-core/src/main/java/org/springframework/integration/annotation/Splitter.java
View File

@@ -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 {};
}

36
spring-integration-core/src/main/java/org/springframework/integration/annotation/Transformer.java
View File

@@ -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 {};
}

43
spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java
View File

@@ -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<T extends Annotation
private static final String ADVICE_CHAIN_ATTRIBUTE = "adviceChain";
protected static final String SEND_TIMEOUT_ATTRIBUTE = "sendTimeout";
protected final List<String> messageHandlerAttributes = new ArrayList<String>();
protected final ConfigurableListableBeanFactory beanFactory;
@@ -98,6 +104,7 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
Assert.notNull(beanFactory, "'beanFactory' must not be null");
Assert.isInstanceOf(ConfigurableListableBeanFactory.class, beanFactory,
"'beanFactory' must be instanceOf ConfigurableListableBeanFactory");
this.messageHandlerAttributes.add(SEND_TIMEOUT_ATTRIBUTE);
this.beanFactory = (ConfigurableListableBeanFactory) beanFactory;
ConversionService conversionService = this.beanFactory.getConversionService();
if (conversionService != null) {
@@ -123,6 +130,18 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
((Orderable) handler).setOrder(orderAnnotation.value());
}
}
if (handler instanceof AbstractMessageProducingHandler || handler instanceof AbstractMessageRouter) {
String sendTimeout = MessagingAnnotationUtils.resolveAttribute(annotations, "sendTimeout", String.class);
if (sendTimeout != null) {
Long value = Long.valueOf(this.environment.resolvePlaceholders(sendTimeout));
if (handler instanceof AbstractMessageProducingHandler) {
((AbstractMessageProducingHandler) handler).setSendTimeout(value);
}
else {
((AbstractMessageRouter) handler).setTimeout(value);
}
}
}
boolean handlerExists = false;
if (this.beanAnnotationAware() && AnnotatedElementUtils.isAnnotated(method, Bean.class.getName())) {
@@ -356,6 +375,11 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
}
protected Object resolveTargetBeanFromMethodWithBeanAnnotation(Method method) {
String id = resolveTargetBeanName(method);
return this.beanFactory.getBean(id);
}
protected String resolveTargetBeanName(Method method) {
String id = null;
String[] names = AnnotationUtils.getAnnotation(method, Bean.class).name();
if (!ObjectUtils.isEmpty(names)) {
@@ -364,7 +388,7 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
if (!StringUtils.hasText(id)) {
id = method.getName();
}
return this.beanFactory.getBean(id);
return id;
}
@SuppressWarnings("unchecked")
@@ -389,6 +413,23 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
return null;
}
protected void checkMessageHandlerAttributes(String handlerBeanName, List<Annotation> 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.
*

13
spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AggregatorAnnotationPostProcessor.java
View File

@@ -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;

4
spring-integration-core/src/main/java/org/springframework/integration/config/annotation/FilterAnnotationPostProcessor.java
View File

@@ -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.<String>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 {

13
spring-integration-core/src/main/java/org/springframework/integration/config/annotation/RouterAnnotationPostProcessor.java
View File

@@ -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.<String>asList("defaultOutputChannel", "applySequence",
"ignoreSendFailures", "resolutionRequired", "channelMappings", "prefix", "suffix"));
}
@Override
protected MessageHandler createHandler(Object bean, Method method, List<Annotation> 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;
}
}

3
spring-integration-core/src/main/java/org/springframework/integration/config/annotation/ServiceActivatorAnnotationPostProcessor.java
View File

@@ -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.<String>asList("outputChannel", "requiresReply", "adviceChain"));
}
@@ -66,6 +68,7 @@ public class ServiceActivatorAnnotationPostProcessor extends AbstractMethodAnnot
}
}
else {
checkMessageHandlerAttributes(resolveTargetBeanName(method), annotations);
return (MessageHandler) target;
}
}

4
spring-integration-core/src/main/java/org/springframework/integration/config/annotation/SplitterAnnotationPostProcessor.java
View File

@@ -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.<String>asList("outputChannel", "applySequence", "adviceChain"));
}
@Override
protected MessageHandler createHandler(Object bean, Method method, List<Annotation> 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;
}
}

7
spring-integration-core/src/main/java/org/springframework/integration/config/annotation/TransformerAnnotationPostProcessor.java
View File

@@ -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.<String>asList("outputChannel", "adviceChain"));
}
@Override
protected MessageHandler createHandler(Object bean, Method method, List<Annotation> 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);

9
spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.2.xsd
View File

@@ -933,8 +933,11 @@
<xsd:attribute name="send-timeout" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
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.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
@@ -4596,7 +4599,7 @@ The list of component name patterns you want to track (e.g., tracked-components
<xsd:annotation>
<xsd:documentation>
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.

9
spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java
View File

@@ -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();

4
spring-integration-core/src/test/java/org/springframework/integration/config/ResequencerParserTests.java
View File

@@ -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",

6
spring-integration-core/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java
View File

@@ -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"));
}

37
spring-integration-core/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationsWithBeanAnnotationTests.java
View File

@@ -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();
}
}
}

4
spring-integration-core/src/test/java/org/springframework/integration/config/annotation/TestAnnotatedEndpointWithCustomizedAggregator.java
View File

@@ -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<Message<?>> messages) {
List<Message<?>> sortableList = new ArrayList<Message<?>>(messages);
Collections.sort(sortableList, new MessageSequenceComparator());

44
spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java
View File

@@ -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

4
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java
View File

@@ -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) {

2
src/reference/asciidoc/aggregator.adoc
View File

@@ -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.

4
src/reference/asciidoc/claim-check.adoc
View File

@@ -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_.

2
src/reference/asciidoc/resequencer.adoc
View File

@@ -83,7 +83,7 @@ See <<reaper>>.
<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.

2
src/reference/asciidoc/syslog.adoc
View File

@@ -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 <<ip-ib-adapter-attributes>>.