diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/DispatcherPolicy.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/DispatcherPolicy.java index de31b3a0a5..1cdf01139e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/DispatcherPolicy.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/DispatcherPolicy.java @@ -37,15 +37,15 @@ public class DispatcherPolicy { private final boolean publishSubscribe; - private int maxMessagesPerTask = DEFAULT_MAX_MESSAGES_PER_TASK; + private volatile int maxMessagesPerTask = DEFAULT_MAX_MESSAGES_PER_TASK; - private long receiveTimeout = DEFAULT_RECEIVE_TIMEOUT; + private volatile long receiveTimeout = DEFAULT_RECEIVE_TIMEOUT; - private int rejectionLimit = DEFAULT_REJECTION_LIMIT; + private volatile int rejectionLimit = DEFAULT_REJECTION_LIMIT; - private long retryInterval = DEFAULT_RETRY_INTERVAL; + private volatile long retryInterval = DEFAULT_RETRY_INTERVAL; - private boolean shouldFailOnRejectionLimit = true; + private volatile boolean shouldFailOnRejectionLimit = true; public DispatcherPolicy() { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/ChannelAdapterParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/ChannelAdapterParser.java index 2545049120..d47021c055 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/ChannelAdapterParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/ChannelAdapterParser.java @@ -100,7 +100,7 @@ public class ChannelAdapterParser implements BeanDefinitionParser { if (!this.isInbound) { RootBeanDefinition endpointDef = new RootBeanDefinition(DefaultMessageEndpoint.class); RootBeanDefinition subscriptionDef = new RootBeanDefinition(Subscription.class); - subscriptionDef.getPropertyValues().addPropertyValue("channel", new RuntimeBeanReference(channel)); + subscriptionDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference(channel)); String subscriptionBeanName = parserContext.getReaderContext().generateBeanName(subscriptionDef); parserContext.registerBeanComponent(new BeanComponentDefinition(subscriptionDef, subscriptionBeanName)); endpointDef.getPropertyValues().addPropertyValue("subscription", new RuntimeBeanReference(subscriptionBeanName)); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/EndpointParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/EndpointParser.java index f3a2fafbb2..36e12932ee 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/EndpointParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/EndpointParser.java @@ -36,6 +36,7 @@ import org.springframework.integration.endpoint.DefaultMessageEndpoint; import org.springframework.integration.handler.DefaultMessageHandlerAdapter; import org.springframework.integration.handler.MessageHandlerChain; import org.springframework.integration.scheduling.PollingSchedule; +import org.springframework.integration.scheduling.Schedule; import org.springframework.integration.scheduling.Subscription; import org.springframework.util.StringUtils; @@ -52,8 +53,6 @@ public class EndpointParser implements BeanDefinitionParser { private static final String SUBSCRIPTION_PROPERTY = "subscription"; - private static final String CHANNEL_NAME_PROPERTY = "channelName"; - private static final String DEFAULT_OUTPUT_CHANNEL_ATTRIBUTE = "default-output-channel"; private static final String DEFAULT_OUTPUT_CHANNEL_PROPERTY = "defaultOutputChannelName"; @@ -92,8 +91,6 @@ public class EndpointParser implements BeanDefinitionParser { private static final String SCHEDULE_ELEMENT = "schedule"; - private static final String SCHEDULE_PROPERTY = "schedule"; - private static final String CONCURRENCY_ELEMENT = "concurrency"; private static final String CONCURRENCY_POLICY_PROPERTY = "concurrencyPolicy"; @@ -103,11 +100,8 @@ public class EndpointParser implements BeanDefinitionParser { RootBeanDefinition endpointDef = new RootBeanDefinition(DefaultMessageEndpoint.class); endpointDef.setSource(parserContext.extractSource(element)); String inputChannel = element.getAttribute(INPUT_CHANNEL_ATTRIBUTE); - RootBeanDefinition subscriptionDef = new RootBeanDefinition(Subscription.class); - if (StringUtils.hasText(inputChannel)) { - subscriptionDef.getPropertyValues().addPropertyValue(CHANNEL_NAME_PROPERTY, inputChannel); - } String defaultOutputChannel = element.getAttribute(DEFAULT_OUTPUT_CHANNEL_ATTRIBUTE); + Schedule schedule = null; if (StringUtils.hasText(defaultOutputChannel)) { endpointDef.getPropertyValues().addPropertyValue(DEFAULT_OUTPUT_CHANNEL_PROPERTY, defaultOutputChannel); } @@ -136,13 +130,20 @@ public class EndpointParser implements BeanDefinitionParser { } } else if (SCHEDULE_ELEMENT.equals(localName)) { - this.parseSchedule((Element) child, subscriptionDef); + schedule = this.parseSchedule((Element) child); } } } - String subscriptionBeanName = parserContext.getReaderContext().generateBeanName(subscriptionDef); - parserContext.registerBeanComponent(new BeanComponentDefinition(subscriptionDef, subscriptionBeanName)); - endpointDef.getPropertyValues().addPropertyValue(SUBSCRIPTION_PROPERTY, new RuntimeBeanReference(subscriptionBeanName)); + if (StringUtils.hasText(inputChannel)) { + RootBeanDefinition subscriptionDef = new RootBeanDefinition(Subscription.class); + subscriptionDef.getConstructorArgumentValues().addGenericArgumentValue(inputChannel); + if (schedule != null) { + subscriptionDef.getConstructorArgumentValues().addGenericArgumentValue(schedule); + } + String subscriptionBeanName = parserContext.getReaderContext().generateBeanName(subscriptionDef); + parserContext.registerBeanComponent(new BeanComponentDefinition(subscriptionDef, subscriptionBeanName)); + endpointDef.getPropertyValues().addPropertyValue(SUBSCRIPTION_PROPERTY, new RuntimeBeanReference(subscriptionBeanName)); + } if (selectors.size() > 0) { endpointDef.getPropertyValues().addPropertyValue(SELECTORS_PROPERTY, selectors); } @@ -199,13 +200,13 @@ public class EndpointParser implements BeanDefinitionParser { endpointDef.getPropertyValues().addPropertyValue(CONCURRENCY_POLICY_PROPERTY, policy); } - private void parseSchedule(Element scheduleElement, RootBeanDefinition subscriptionDef) { + private Schedule parseSchedule(Element scheduleElement) { PollingSchedule schedule = new PollingSchedule(5); String period = scheduleElement.getAttribute(PERIOD_ATTRIBUTE); if (StringUtils.hasText(period)) { schedule.setPeriod(Integer.parseInt(period)); } - subscriptionDef.getPropertyValues().addPropertyValue(SCHEDULE_PROPERTY, schedule); + return schedule; } private String parseHandlerAdapter(String handlerRef, String handlerMethod, ParserContext parserContext) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java index 9139dd2ddb..a467a03ac4 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java @@ -131,10 +131,8 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor final DefaultMessageEndpoint endpoint) { String channelName = annotation.input(); if (StringUtils.hasText(channelName)) { - Subscription subscription = new Subscription(); - subscription.setChannelName(channelName); Schedule schedule = new PollingSchedule(annotation.pollPeriod()); - subscription.setSchedule(schedule); + Subscription subscription = new Subscription(channelName, schedule); endpoint.setSubscription(subscription); } ReflectionUtils.doWithMethods(bean.getClass(), new ReflectionUtils.MethodCallback() { @@ -155,11 +153,10 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor String channelName = beanName + "-inputChannel"; messageBus.registerChannel(channelName, channel); messageBus.registerSourceAdapter(beanName + "-sourceAdapter", adapter); - Subscription subscription = new Subscription(channel); PollingSchedule schedule = new PollingSchedule(period); schedule.setInitialDelay(initialDelay); schedule.setFixedRate(fixedRate); - subscription.setSchedule(schedule); + Subscription subscription = new Subscription(channel, schedule); endpoint.setSubscription(subscription); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd b/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd index f0c129cafc..91634b179e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd @@ -123,7 +123,7 @@ - + @@ -209,11 +209,12 @@ - + - Defines an aggregating message handler + + Defines an aggregating message handler. diff --git a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/Subscription.java b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/Subscription.java index c2b64e7c32..80a9311659 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/Subscription.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/Subscription.java @@ -17,30 +17,42 @@ package org.springframework.integration.scheduling; import org.springframework.integration.channel.MessageChannel; +import org.springframework.util.Assert; /** - * Configuration metadata for activating a subscription. + * Configuration metadata for activating a subscription. Immutable. * * @author Mark Fisher */ public class Subscription { - private MessageChannel channel; + private final MessageChannel channel; - private String channelName; + private final String channelName; - private Schedule schedule; + private final Schedule schedule; - public Subscription() { - } - public Subscription(MessageChannel channel) { - this.channel = channel; + this(channel, null); } public Subscription(String channelName) { + this(channelName, null); + } + + public Subscription(MessageChannel channel, Schedule schedule) { + Assert.notNull(channel, "'channel' must not be null"); + this.channel = channel; + this.schedule = schedule; + this.channelName = this.channel.getName(); + } + + public Subscription(String channelName, Schedule schedule) { + Assert.notNull(channelName, "'channelName' must not be null"); this.channelName = channelName; + this.schedule = schedule; + this.channel = null; } @@ -48,24 +60,12 @@ public class Subscription { return this.channel; } - public void setChannel(MessageChannel channel) { - this.channel = channel; - } - public String getChannelName() { return (this.channel != null) ? this.channel.getName() : this.channelName; } - public void setChannelName(String channelName) { - this.channelName = channelName; - } - public Schedule getSchedule() { return this.schedule; } - public void setSchedule(Schedule schedule) { - this.schedule = schedule; - } - } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/DefaultTargetAdapterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/adapter/DefaultTargetAdapterTests.java index dc8f0ee2bd..a2051c1b27 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/adapter/DefaultTargetAdapterTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/DefaultTargetAdapterTests.java @@ -47,8 +47,7 @@ public class DefaultTargetAdapterTests { target.afterPropertiesSet(); DefaultTargetAdapter adapter = new DefaultTargetAdapter(target); SimpleChannel channel = new SimpleChannel(); - Subscription subscription = new Subscription(); - subscription.setChannel(channel); + Subscription subscription = new Subscription(channel); Message message = new GenericMessage("123", "testing"); channel.send(message); assertNull(queue.poll()); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/CorrelationIdTests.java b/spring-integration-core/src/test/java/org/springframework/integration/handler/CorrelationIdTests.java index add7575ee8..38fd456107 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/handler/CorrelationIdTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/handler/CorrelationIdTests.java @@ -64,7 +64,6 @@ public class CorrelationIdTests { assertEquals(message.getId(), reply.getHeader().getCorrelationId()); } - @Test public void testCorrelationIdCopiedFromMessageCorrelationIdIfAvailable() { Message message = new StringMessage("messageId","test"); @@ -77,7 +76,7 @@ public class CorrelationIdTests { assertEquals(message.getHeader().getCorrelationId(), reply.getHeader().getCorrelationId()); assertTrue(message.getHeader().getCorrelationId().equals(reply.getHeader().getCorrelationId())); } - + @Test public void testCorrelationNotPassedIfAlreadySetByHandler() throws Exception { Object correlationId = "123-ABC"; @@ -85,7 +84,7 @@ public class CorrelationIdTests { message.getHeader().setCorrelationId(correlationId); AbstractMessageHandlerAdapter adapter = new AbstractMessageHandlerAdapter() { @Override - protected Object doHandle(Message message, SimpleMethodInvoker invoker) { + protected Object doHandle(Message message, SimpleMethodInvoker invoker) { Object result = invoker.invokeMethod(message.getPayload()); Message resultMessage = new GenericMessage(result); resultMessage.getHeader().setCorrelationId("456-XYZ"); @@ -104,7 +103,7 @@ public class CorrelationIdTests { Message message = new StringMessage("test"); AbstractMessageHandlerAdapter adapter = new AbstractMessageHandlerAdapter() { @Override - protected Object doHandle(Message message, SimpleMethodInvoker invoker) { + protected Object doHandle(Message message, SimpleMethodInvoker invoker) { Object result = invoker.invokeMethod(message.getPayload()); Message resultMessage = new GenericMessage(result); resultMessage.getHeader().setCorrelationId("456-XYZ");