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 c9c8c4fa57..cb59ad23f8 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 @@ -304,7 +304,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP return "aggregator"; } - protected MessageGroupStore getMessageStore() { + public MessageGroupStore getMessageStore() { return messageStore; } 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 2c4fd7f799..4abaf436d7 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 @@ -21,6 +21,7 @@ import java.lang.reflect.Method; import java.util.List; import org.springframework.beans.factory.ListableBeanFactory; +import org.springframework.context.Lifecycle; import org.springframework.context.annotation.Bean; import org.springframework.core.annotation.AnnotatedElementUtils; import org.springframework.core.env.Environment; @@ -58,15 +59,7 @@ public class ServiceActivatorAnnotationPostProcessor extends AbstractMethodAnnot * Return a reply-producing message handler so that we still get 'produced no reply' messages * and the super class will inject the advice chain to advise the handler method if needed. */ - return new AbstractReplyProducingMessageHandler() { - - @Override - protected Object handleRequestMessage(Message requestMessage) { - - ((MessageHandler) target).handleMessage(requestMessage); - return null; - } - }; + return new ReplyProducingMessageHandlerWrapper((MessageHandler) target); } else { serviceActivator = new ServiceActivatingHandler(target); @@ -89,4 +82,41 @@ public class ServiceActivatorAnnotationPostProcessor extends AbstractMethodAnnot return serviceActivator; } + private class ReplyProducingMessageHandlerWrapper extends AbstractReplyProducingMessageHandler + implements Lifecycle { + + private final MessageHandler target; + + private ReplyProducingMessageHandlerWrapper(MessageHandler target) { + this.target = target; + } + + @Override + protected Object handleRequestMessage(Message requestMessage) { + this.target.handleMessage(requestMessage); + return null; + } + + @Override + public void start() { + if (this.target instanceof Lifecycle) { + ((Lifecycle) this.target).start(); + } + + } + + @Override + public void stop() { + if (this.target instanceof Lifecycle) { + ((Lifecycle) this.target).stop(); + } + } + + @Override + public boolean isRunning() { + return !(this.target instanceof Lifecycle) || ((Lifecycle) this.target).isRunning(); + } + + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceHandler.java index b8a13d9b5e..59f9b4d1fd 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceHandler.java @@ -81,6 +81,7 @@ public class IntegrationNamespaceHandler extends AbstractIntegrationNamespaceHan RetryAdviceParser retryParser = new RetryAdviceParser(); registerBeanDefinitionParser("handler-retry-advice", retryParser); registerBeanDefinitionParser("retry-advice", retryParser); + registerBeanDefinitionParser("scatter-gather", new ScatterGatherParser()); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ScatterGatherParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ScatterGatherParser.java new file mode 100644 index 0000000000..a1e05f201e --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ScatterGatherParser.java @@ -0,0 +1,118 @@ +/* + * Copyright 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.config.xml; + +import javax.xml.parsers.DocumentBuilderFactory; +import javax.xml.parsers.ParserConfigurationException; + +import org.w3c.dom.Element; + +import org.springframework.beans.factory.config.BeanDefinition; +import org.springframework.beans.factory.config.RuntimeBeanReference; +import org.springframework.beans.factory.support.AbstractBeanDefinition; +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.support.RootBeanDefinition; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.handler.ScatterGatherHandler; +import org.springframework.integration.router.RecipientListRouter; +import org.springframework.util.StringUtils; +import org.springframework.util.xml.DomUtils; + +/** + * Parser for the <scatter-gather> element. + * + * @author Artem Bilan + * @since 4.1 + */ +public class ScatterGatherParser extends AbstractConsumerEndpointParser { + + private static final RecipientListRouterParser SCATTERER_PARSER = new RecipientListRouterParser(); + + private static final AggregatorParser GATHERER_PARSER = new AggregatorParser(); + + private static final DocumentBuilderFactory documentBuilderFactory = DocumentBuilderFactory.newInstance(); + + @Override + protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) { + String scatterChannel = element.getAttribute("scatter-channel"); + boolean hasScatterChannel = StringUtils.hasText(scatterChannel); + Element scatterer = DomUtils.getChildElementByTagName(element, "scatterer"); + boolean hasScatterer = scatterer != null; + + if (hasScatterChannel & hasScatterer) { + parserContext.getReaderContext() + .error("'scatter-channel' attribute and 'scatterer' sub-element are mutually exclusive", element); + } + + if (!hasScatterChannel & !hasScatterer) { + parserContext.getReaderContext() + .error("The 'scatter-channel' attribute or 'scatterer' sub-element must be specified", element); + } + + BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(ScatterGatherHandler.class); + AbstractBeanDefinition scatterGatherDefinition = builder.getRawBeanDefinition(); + String id = resolveId(element, scatterGatherDefinition, parserContext); + + if (hasScatterChannel) { + builder.addConstructorArgReference(scatterChannel); + } + else { + BeanDefinition scattererDefinition = null; + if (!hasScatterer) { + scattererDefinition = new RootBeanDefinition(RecipientListRouter.class); + } + else { + scattererDefinition = SCATTERER_PARSER.parse(scatterer, + new ParserContext(parserContext.getReaderContext(), parserContext.getDelegate(), + scatterGatherDefinition)); + } + + String scattererId = id + ".scatterer"; + if (hasScatterer && scatterer.hasAttribute(ID_ATTRIBUTE)) { + scattererId = scatterer.getAttribute(ID_ATTRIBUTE); + } + parserContext.getRegistry().registerBeanDefinition(scattererId, scattererDefinition); + builder.addConstructorArgValue(new RuntimeBeanReference(scattererId)); + } + + Element gatherer = DomUtils.getChildElementByTagName(element, "gatherer"); + + BeanDefinition gathererDefinition = null; + if (gatherer == null) { + try { + gatherer = documentBuilderFactory.newDocumentBuilder().newDocument().createElement("aggregator"); + } + catch (ParserConfigurationException e) { + parserContext.getReaderContext().error(e.getMessage(), element); + } + } + gathererDefinition = GATHERER_PARSER.parse(gatherer, new ParserContext(parserContext.getReaderContext(), + parserContext.getDelegate(), scatterGatherDefinition)); + String gathererId = id + ".gatherer"; + if (gatherer != null && gatherer.hasAttribute(ID_ATTRIBUTE)) { + gathererId = gatherer.getAttribute(ID_ATTRIBUTE); + } + parserContext.getRegistry().registerBeanDefinition(gathererId, gathererDefinition); + builder.addConstructorArgValue(new RuntimeBeanReference(gathererId)); + + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "gather-channel"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "gather-timeout"); + + return builder; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/ScatterGatherHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/ScatterGatherHandler.java new file mode 100644 index 0000000000..a863b93923 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/ScatterGatherHandler.java @@ -0,0 +1,180 @@ +/* + * Copyright 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.handler; + +import org.springframework.aop.support.AopUtils; +import org.springframework.context.Lifecycle; +import org.springframework.integration.aggregator.AggregatingMessageHandler; +import org.springframework.integration.channel.FixedSubscriberChannel; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.context.IntegrationContextUtils; +import org.springframework.integration.core.MessageProducer; +import org.springframework.integration.endpoint.AbstractEndpoint; +import org.springframework.integration.endpoint.EventDrivenConsumer; +import org.springframework.integration.endpoint.PollingConsumer; +import org.springframework.integration.router.RecipientListRouter; +import org.springframework.integration.support.channel.HeaderChannelRegistry; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageDeliveryException; +import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.MessagingException; +import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.SubscribableChannel; +import org.springframework.util.Assert; + +/** + * The {@link MessageHandler} implementation for the + * Scatter-Gather EIP pattern. + * + * @author Artem Bilan + * @since 4.1 + */ +public class ScatterGatherHandler extends AbstractReplyProducingMessageHandler implements Lifecycle { + + private static final String GATHER_RESULT_CHANNEL = "gatherResultChannel"; + + private final MessageChannel scatterChannel; + + private final MessageHandler gatherer; + + private MessageChannel gatherChannel; + + private long gatherTimeout = -1; + + private AbstractEndpoint gatherEndpoint; + + private HeaderChannelRegistry replyChannelRegistry; + + + public ScatterGatherHandler(MessageChannel scatterChannel, MessageHandler gatherer) { + Assert.notNull(scatterChannel); + Assert.notNull(gatherer); + Class gathererClass = AopUtils.getTargetClass(gatherer); + Assert.isAssignable(AggregatingMessageHandler.class, gathererClass, + "the 'gatherer' must be an AggregatingMessageHandler instance"); + this.scatterChannel = scatterChannel; + this.gatherer = gatherer; + } + + public ScatterGatherHandler(MessageHandler scatterer, MessageHandler gatherer) { + this(new FixedSubscriberChannel(scatterer), gatherer); + Assert.notNull(scatterer); + Class scatterClass = AopUtils.getTargetClass(scatterer); + Assert.isAssignable(RecipientListRouter.class, scatterClass, + "the 'scatterer' must be a RecipientListRouter instance"); + } + + public void setGatherChannel(MessageChannel gatherChannel) { + this.gatherChannel = gatherChannel; + } + + public void setGatherTimeout(long gatherTimeout) { + this.gatherTimeout = gatherTimeout; + } + + @Override + protected void doInit() { + if (this.gatherChannel == null) { + this.gatherChannel = new FixedSubscriberChannel(this.gatherer); + } + else { + if (this.gatherChannel instanceof SubscribableChannel) { + this.gatherEndpoint = new EventDrivenConsumer((SubscribableChannel) this.gatherChannel, this.gatherer); + } + else if (this.gatherChannel instanceof PollableChannel) { + this.gatherEndpoint = new PollingConsumer((PollableChannel) this.gatherChannel, this.gatherer); + ((PollingConsumer) this.gatherEndpoint).setReceiveTimeout(this.gatherTimeout); + } + else { + throw new MessagingException("Unsupported 'replyChannel' type [" + this.gatherChannel.getClass() + "]." + + "SubscribableChannel or PollableChannel type are supported."); + } + this.gatherEndpoint.setBeanFactory(this.getBeanFactory()); + this.gatherEndpoint.afterPropertiesSet(); + } + + ((MessageProducer) this.gatherer).setOutputChannel(new FixedSubscriberChannel(new MessageHandler() { + + @Override + public void handleMessage(Message message) throws MessagingException { + MessageHeaders headers = message.getHeaders(); + if (headers.containsKey(GATHER_RESULT_CHANNEL)) { + Object gatherResultChannel = headers.get(GATHER_RESULT_CHANNEL); + if (gatherResultChannel instanceof MessageChannel) { + messagingTemplate.send((MessageChannel) gatherResultChannel, message); + } + else if (gatherResultChannel instanceof String) { + messagingTemplate.send((String) gatherResultChannel, message); + } + } + else { + throw new MessageDeliveryException(message, + "The 'gatherResultChannel' header is required to delivery gather result."); + } + } + + })); + + this.replyChannelRegistry = getBeanFactory() + .getBean(IntegrationContextUtils.INTEGRATION_HEADER_CHANNEL_REGISTRY_BEAN_NAME, + HeaderChannelRegistry.class); + } + + @Override + protected Object handleRequestMessage(Message requestMessage) { + PollableChannel gatherResultChannel = new QueueChannel(); + + Object gatherResultChannelName = this.replyChannelRegistry.channelToChannelName(gatherResultChannel); + + Message scatterMessage = getMessageBuilderFactory() + .fromMessage(requestMessage) + .setHeader(GATHER_RESULT_CHANNEL, gatherResultChannelName) + .setReplyChannel(this.gatherChannel) + .build(); + + this.messagingTemplate.send(this.scatterChannel, scatterMessage); + + Message gatherResult = gatherResultChannel.receive(this.gatherTimeout); + if (gatherResult != null) { + return gatherResult.getPayload(); + } + + return null; + } + + @Override + public void start() { + if (this.gatherEndpoint != null) { + gatherEndpoint.start(); + } + } + + @Override + public void stop() { + if (this.gatherEndpoint != null) { + gatherEndpoint.start(); + } + } + + @Override + public boolean isRunning() { + return this.gatherEndpoint == null || this.gatherEndpoint.isRunning(); + } + +} diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.1.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.1.xsd index f21f7aa3b1..dd888a5cac 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.1.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.1.xsd @@ -680,8 +680,8 @@ @@ -762,7 +762,7 @@ but a custom executor can return any Future. If the method return type is not compatible with the executor the flow will run on the caller's thread and the flow must return an appropriate Future. Finally, you can disable the - gateway ansync handling by setting this attribute to "". This allows the downstream + gateway async handling by setting this attribute to "". This allows the downstream flow to return a Future that would otherwise have been compatible with the default executor. ]]> @@ -1446,7 +1446,7 @@ Boolean value indicating whether any payload that implements Cloneable should be cloned - prior to sending the Message to the request chanenl for acquiring the enriching data. + prior to sending the Message to the request channel for acquiring the enriching data. The cloned version would be used as the target payload for the ultimate reply. If the payload does NOT implement 'Cloneable', then setting this @@ -1800,7 +1800,7 @@ The java.util.concurrent.TimeUnit enum value. This can ONLY be used in combination with the 'fixed-delay' or 'fixed-rate' attributes. If combined with either 'cron' or a 'trigger' reference attribute, it will cause a failure. The minimal supported - granularity for a PeriodicTrigger is MILLISEONDS. Therefore, the only available options + granularity for a PeriodicTrigger is MILLISECONDS. Therefore, the only available options are MILLISECONDS and SECONDS. If this value is not provided, then any 'fixed-delay' or 'fixed-rate' value will be interpreted as MILLISECONDS by default. Basically this enum provides a convenience for SECONDS-based interval trigger values. For hourly, daily, @@ -2885,7 +2885,7 @@ - + - + - + @@ -3319,7 +3319,7 @@ If set to 'true', a MessagingException will be raised in case the channel cannot be resolved. Setting this attribute to 'false', - will cause any unresovable channels to be ignored. + will cause any unresolvable channels to be ignored. If not explicitly set, 'resolution-required' will default to 'true'. @@ -3699,7 +3699,7 @@ - Alows you to configure a Wire Tap interceptor that will send a copy of the message to a + Allows you to configure a Wire Tap interceptor that will send a copy of the message to a channel identified by 'channel' attribute. @@ -4210,6 +4210,80 @@ The list of component name patterns you want to track (e.g., tracked-components + + + + + + + + + + + + + + + + + + + + Base type for 'scatter-gather' elements. + + + + + + + + + + + + + + + + + + 'id' value: + - Identifies the underlying Spring bean definition (AbstractEndpoint) + - as MessageHandler bean alias together with suffix '.handler' + - as a RecipientListRouter bean together with suffix '.scatterer' + - as a AggregatingMessageHandler bean together with suffix '.gatherer' + + + + + + + Identifies the channel to send a message for 'auction' Scatter-Gather pattern variant. + Mutually exclusive with 'scatterer' sub-element. + + + + + + + Identifies the channel to receive reply Messages for gathering. + + + + + + + Allows to specify how long the Scatter-Gather will wait for reply Messages for gathering. + By default it will wait indefinitely. Value is specified in milliseconds. + It will be applied only if the 'gather-channel' is specified and it is some blocking channel, + e.g. 'QueueChannel'. + + + + + @@ -4374,7 +4448,7 @@ endpoint itself is a Polling Consumer for a channel with a queue. or poller + Attributes provided in either a or poller sub element. Used to take action after the transaction completes () or after the channel.send() is complete (). @@ -4431,7 +4505,7 @@ endpoint itself is a Polling Consumer for a channel with a queue. diff --git a/spring-integration-core/src/test/java/org/springframework/integration/scattergather/config/ScatterGatherParserTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/scattergather/config/ScatterGatherParserTests-context.xml new file mode 100644 index 0000000000..307deffc0c --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/scattergather/config/ScatterGatherParserTests-context.xml @@ -0,0 +1,39 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/scattergather/config/ScatterGatherParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/scattergather/config/ScatterGatherParserTests.java new file mode 100644 index 0000000000..ad5f876bcc --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/scattergather/config/ScatterGatherParserTests.java @@ -0,0 +1,101 @@ +/* + * Copyright 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.scattergather.config; + +import static org.hamcrest.Matchers.instanceOf; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; + +import java.util.Collection; + +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.BeanFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.integration.aggregator.AggregatingMessageHandler; +import org.springframework.integration.channel.FixedSubscriberChannel; +import org.springframework.integration.endpoint.EventDrivenConsumer; +import org.springframework.integration.handler.ScatterGatherHandler; +import org.springframework.integration.router.RecipientListRouter; +import org.springframework.integration.test.util.TestUtils; +import org.springframework.messaging.MessageHandler; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Artem Bilan + * @since 4.1 + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +public class ScatterGatherParserTests { + + @Autowired + private BeanFactory beanFactory; + + @Test + public void testAuction() { + MessageHandler scatterGather = this.beanFactory.getBean("scatterGather1.handler", MessageHandler.class); + assertThat(scatterGather, instanceOf(ScatterGatherHandler.class)); + assertSame(this.beanFactory.getBean("scatterChannel"), + TestUtils.getPropertyValue(scatterGather, "scatterChannel")); + assertTrue(this.beanFactory.containsBean("scatterGather1.gatherer")); + AggregatingMessageHandler gatherer = + this.beanFactory.getBean("scatterGather1.gatherer", AggregatingMessageHandler.class); + assertSame(gatherer, TestUtils.getPropertyValue(scatterGather, "gatherer")); + + Object reaper = this.beanFactory.getBean("reaper"); + assertSame(gatherer.getMessageStore(), TestUtils.getPropertyValue(reaper, "messageGroupStore")); + } + + @Test + @SuppressWarnings("unchecked") + public void testDistribution() { + MessageHandler scatterGather = this.beanFactory.getBean("scatterGather2.handler", MessageHandler.class); + assertSame(this.beanFactory.getBean("gatherChannel"), + TestUtils.getPropertyValue(scatterGather, "gatherChannel")); + assertNotNull(TestUtils.getPropertyValue(scatterGather, "gatherEndpoint")); + assertThat(TestUtils.getPropertyValue(scatterGather, "gatherEndpoint"), instanceOf(EventDrivenConsumer.class)); + assertTrue(TestUtils.getPropertyValue(scatterGather, "gatherEndpoint.running", Boolean.class)); + assertEquals(100L, TestUtils.getPropertyValue(scatterGather, "gatherTimeout")); + + assertTrue(this.beanFactory.containsBean("myGatherer")); + Object gatherer = this.beanFactory.getBean("myGatherer"); + assertSame(gatherer, TestUtils.getPropertyValue(scatterGather, "gatherer")); + assertSame(this.beanFactory.getBean("messageStore"), TestUtils.getPropertyValue(gatherer, "messageStore")); + assertSame(gatherer, TestUtils.getPropertyValue(scatterGather, "gatherEndpoint.handler")); + + assertTrue(this.beanFactory.containsBean("myScatterer")); + Object scatterer = this.beanFactory.getBean("myScatterer"); + assertTrue(TestUtils.getPropertyValue(scatterer, "applySequence", Boolean.class)); + + Collection recipients = TestUtils.getPropertyValue(scatterer, "recipients", + Collection.class); + assertEquals(1, recipients.size()); + assertSame(this.beanFactory.getBean("distributionChannel"), recipients.iterator().next().getChannel()); + + Object scatterChannel = TestUtils.getPropertyValue(scatterGather, "scatterChannel"); + assertThat(scatterChannel, instanceOf(FixedSubscriberChannel.class)); + assertSame(scatterer, TestUtils.getPropertyValue(scatterChannel, "handler")); + + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/scattergather/config/ScatterGatherTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/scattergather/config/ScatterGatherTests-context.xml new file mode 100644 index 0000000000..94b70f9f48 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/scattergather/config/ScatterGatherTests-context.xml @@ -0,0 +1,61 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/scattergather/config/ScatterGatherTests.java b/spring-integration-core/src/test/java/org/springframework/integration/scattergather/config/ScatterGatherTests.java new file mode 100644 index 0000000000..3190e0e257 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/scattergather/config/ScatterGatherTests.java @@ -0,0 +1,86 @@ +/* + * Copyright 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.scattergather.config; + +import static org.hamcrest.Matchers.greaterThanOrEqualTo; +import static org.hamcrest.Matchers.instanceOf; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertThat; + +import java.util.List; + +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.integration.gateway.RequestReplyExchanger; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.support.GenericMessage; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Artem Bilan + * @since 4.1 + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +public class ScatterGatherTests { + + @Autowired + private PollableChannel output; + + @Autowired + private MessageChannel inputAuction; + + @Autowired + private MessageChannel inputDistribution; + + @Autowired + private RequestReplyExchanger gateway; + + @Test + public void testAuction() { + this.inputAuction.send(new GenericMessage("foo")); + Message bestQuoteMessage = this.output.receive(10000); + assertNotNull(bestQuoteMessage); + Object payload = bestQuoteMessage.getPayload(); + assertThat(payload, instanceOf(List.class)); + assertThat(((List) payload).size(), greaterThanOrEqualTo(1)); + } + + @Test + public void testDistribution() { + this.inputDistribution.send(new GenericMessage("foo")); + Message bestQuoteMessage = this.output.receive(10000); + assertNotNull(bestQuoteMessage); + Object payload = bestQuoteMessage.getPayload(); + assertThat(payload, instanceOf(List.class)); + assertThat(((List) payload).size(), greaterThanOrEqualTo(1)); + } + + @Test + public void testGatewayScatterGather() { + Message bestQuoteMessage = this.gateway.exchange(new GenericMessage("foo")); + Object payload = bestQuoteMessage.getPayload(); + assertThat(payload, instanceOf(List.class)); + assertThat(((List) payload).size(), greaterThanOrEqualTo(1)); + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpOutboundGatewayParser.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpOutboundGatewayParser.java index c1b778b2b0..e50b736ae8 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpOutboundGatewayParser.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpOutboundGatewayParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 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. @@ -22,6 +22,7 @@ import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.xml.ParserContext; import org.springframework.integration.config.xml.AbstractConsumerEndpointParser; import org.springframework.integration.config.xml.IntegrationNamespaceUtils; +import org.springframework.integration.ip.tcp.TcpOutboundGateway; /** * Parser for the <outbound-gateway> element of the integration 'jms' namespace. @@ -31,8 +32,6 @@ import org.springframework.integration.config.xml.IntegrationNamespaceUtils; */ public class TcpOutboundGatewayParser extends AbstractConsumerEndpointParser { - private static final String BASE_PACKAGE = "org.springframework.integration.ip.tcp"; - @Override protected String getInputChannelAttributeName() { return "request-channel"; @@ -40,8 +39,7 @@ public class TcpOutboundGatewayParser extends AbstractConsumerEndpointParser { @Override protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) { - BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(BASE_PACKAGE + - ".TcpOutboundGateway"); + BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(TcpOutboundGateway.class); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, IpAdapterParserUtils.TCP_CONNECTION_FACTORY); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, @@ -52,10 +50,6 @@ public class TcpOutboundGatewayParser extends AbstractConsumerEndpointParser { IpAdapterParserUtils.REMOTE_TIMEOUT); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IpAdapterParserUtils.REPLY_TIMEOUT, "sendTimeout"); - IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, - IntegrationNamespaceUtils.AUTO_STARTUP); - IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, - IntegrationNamespaceUtils.PHASE); return builder; } 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 5e1e093928..e5ed9691ed 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 @@ -22,6 +22,7 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; +import org.springframework.context.Lifecycle; import org.springframework.context.SmartLifecycle; import org.springframework.integration.MessageTimeoutException; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; @@ -50,7 +51,8 @@ import org.springframework.util.Assert; * @author Gary Russell * @since 2.0 */ -public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler implements TcpSender, TcpListener, SmartLifecycle { +public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler + implements TcpSender, TcpListener, Lifecycle { private volatile AbstractClientConnectionFactory connectionFactory; @@ -64,10 +66,6 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler imp private volatile long requestTimeout = 10000; - private volatile boolean autoStartup = true; - - private volatile int phase; - /** * @param requestTimeout the requestTimeout to set */ @@ -232,29 +230,6 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler imp return this.connectionFactory.isRunning(); } - @Override - public int getPhase() { - return this.phase; - } - - @Override - public boolean isAutoStartup() { - return this.autoStartup; - } - - @Override - public void stop(Runnable callback) { - this.connectionFactory.stop(callback); - } - - public void setAutoStartup(boolean autoStartup) { - this.autoStartup = autoStartup; - } - - public void setPhase(int phase) { - this.phase = phase; - } - /** * @return the connectionFactory */ diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java index 81398e8be2..c951164452 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java @@ -496,8 +496,6 @@ public class ParserUnitTests { assertEquals("ip:tcp-outbound-gateway", tcpOutboundGateway.getComponentType()); assertTrue(cfC2.isLookupHost()); assertEquals(24, dfa.getPropertyValue("order")); - assertFalse(tcpOutboundGateway.isAutoStartup()); - assertEquals(127, tcpOutboundGateway.getPhase()); } @Test diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java index 582602eedb..e249bb8195 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java @@ -42,7 +42,7 @@ import javax.jms.TemporaryQueue; import javax.jms.TemporaryTopic; import javax.jms.Topic; -import org.springframework.context.SmartLifecycle; +import org.springframework.context.Lifecycle; import org.springframework.expression.Expression; import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.MessageTimeoutException; @@ -74,7 +74,7 @@ import org.springframework.util.StringUtils; * @author Gary Russell * @author Artem Bilan */ -public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler implements SmartLifecycle, MessageListener { +public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler implements Lifecycle, MessageListener { private volatile Destination requestDestination; @@ -126,8 +126,6 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp private final Object initializationMonitor = new Object(); - private volatile boolean autoStartup; - private volatile boolean active; private final AtomicLong correlationId = new AtomicLong(); @@ -484,20 +482,6 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp session, replyDestinationName, this.replyPubSubDomain); } - @Override - public int getPhase() { - return Integer.MAX_VALUE; - } - - @Override - public boolean isAutoStartup() { - return this.autoStartup; - } - - public void setAutoStartup(boolean autoStartup) { - this.autoStartup = autoStartup; - } - @Override protected void doInit() { synchronized (this.initializationMonitor) { @@ -663,12 +647,6 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp return this.active; } - @Override - public void stop(Runnable callback) { - this.stop(); - callback.run(); - } - @Override protected Object handleRequestMessage(final Message message) { if (!this.initialized) { diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ScatterGatherHandlerIntegrationTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ScatterGatherHandlerIntegrationTests.java new file mode 100644 index 0000000000..5219a2f69d --- /dev/null +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ScatterGatherHandlerIntegrationTests.java @@ -0,0 +1,361 @@ +/* + * Copyright 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.monitor; + +import static org.hamcrest.Matchers.instanceOf; +import static org.hamcrest.Matchers.lessThan; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertThat; + +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.Executors; + +import javax.management.MBeanServer; + +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.integration.IntegrationMessageHeaderAccessor; +import org.springframework.integration.aggregator.AggregatingMessageHandler; +import org.springframework.integration.aggregator.DefaultAggregatingMessageGroupProcessor; +import org.springframework.integration.aggregator.ExpressionEvaluatingMessageGroupProcessor; +import org.springframework.integration.aggregator.ExpressionEvaluatingReleaseStrategy; +import org.springframework.integration.aggregator.HeaderAttributeCorrelationStrategy; +import org.springframework.integration.aggregator.MessageCountReleaseStrategy; +import org.springframework.integration.annotation.ServiceActivator; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.channel.PublishSubscribeChannel; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.config.EnableIntegration; +import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; +import org.springframework.integration.handler.BridgeHandler; +import org.springframework.integration.handler.ScatterGatherHandler; +import org.springframework.integration.jmx.config.EnableIntegrationMBeanExport; +import org.springframework.integration.router.RecipientListRouter; +import org.springframework.integration.store.SimpleMessageStore; +import org.springframework.jmx.support.MBeanServerFactoryBean; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.SubscribableChannel; +import org.springframework.messaging.support.MessageBuilder; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Artem Bilan + * @since 4.1 + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +@DirtiesContext +public class ScatterGatherHandlerIntegrationTests { + + @Autowired + private PollableChannel output; + + @Autowired + private MessageChannel inputAuctionWithoutGatherChannel; + + @Autowired + private MessageChannel inputAuctionWithGatherChannel; + + @Autowired + private MessageChannel distributionChannel; + + @Test + public void testSimpleAuction() { + Message quoteMessage = MessageBuilder.withPayload("testQuote").build(); + + this.inputAuctionWithoutGatherChannel.send(quoteMessage); + Message bestQuoteMessage = this.output.receive(10000); + assertNotNull(bestQuoteMessage); + Object payload = bestQuoteMessage.getPayload(); + assertThat(payload, instanceOf(Double.class)); + assertThat((Double) payload, lessThan(10D)); + } + + @Test + public void testAuctionWithGatherChannel() { + Message quoteMessage = MessageBuilder.withPayload("testQuote").build(); + + this.inputAuctionWithGatherChannel.send(quoteMessage); + Message bestQuoteMessage = this.output.receive(10000); + assertNotNull(bestQuoteMessage); + Object payload = bestQuoteMessage.getPayload(); + assertThat(payload, instanceOf(List.class)); + assertEquals(3, ((List) payload).size()); + } + + @Test + public void testDistribution() { + Message quoteMessage = MessageBuilder.withPayload("testQuote").build(); + + this.distributionChannel.send(quoteMessage); + Message bestQuoteMessage = this.output.receive(10000); + assertNotNull(bestQuoteMessage); + Object payload = bestQuoteMessage.getPayload(); + assertThat(payload, instanceOf(Double.class)); + assertThat((Double) payload, lessThan(10D)); + } + + + @Configuration + @EnableIntegration + @EnableIntegrationMBeanExport(server = "mBeanServer") + public static class ContextConfiguration { + + @Bean + public static MBeanServerFactoryBean mBeanServer() { + return new MBeanServerFactoryBean(); + } + + @Bean + public PollableChannel output() { + return new QueueChannel(); + } + + @Bean + public SubscribableChannel scatterAuctionWithoutGatherChannel() { + PublishSubscribeChannel channel = new PublishSubscribeChannel(); + channel.setApplySequence(true); + return channel; + } + + @Bean + public MessageHandler gatherer1() { + return new AggregatingMessageHandler( + new ExpressionEvaluatingMessageGroupProcessor("^[payload gt 5] ?: -1D"), + new SimpleMessageStore(), + new HeaderAttributeCorrelationStrategy(IntegrationMessageHeaderAccessor.CORRELATION_ID), + new ExpressionEvaluatingReleaseStrategy("size() == 2")); + } + + @Bean + public MessageChannel inputAuctionWithoutGatherChannel() { + return new DirectChannel(); + } + + @Bean + @ServiceActivator(inputChannel = "inputAuctionWithoutGatherChannel") + public MessageHandler scatterGatherAuctionWithoutGatherChannel() { + ScatterGatherHandler handler = new ScatterGatherHandler(scatterAuctionWithoutGatherChannel(), gatherer1()); + handler.setOutputChannel(output()); + return handler; + } + + @Bean + @ServiceActivator(inputChannel = "scatterAuctionWithoutGatherChannel") + public MessageHandler auctionWithoutGatherChannelBridge1() { + BridgeHandler handler = new BridgeHandler(); + handler.setOutputChannel(serviceChannel1()); + return handler; + } + + @Bean + @ServiceActivator(inputChannel = "scatterAuctionWithoutGatherChannel") + public MessageHandler auctionWithoutGatherChannelBridge2() { + BridgeHandler handler = new BridgeHandler(); + handler.setOutputChannel(serviceChannel1()); + return handler; + } + + @Bean + @ServiceActivator(inputChannel = "scatterAuctionWithoutGatherChannel") + public MessageHandler auctionWithoutGatherChannelBridge3() { + BridgeHandler handler = new BridgeHandler(); + handler.setOutputChannel(serviceChannel1()); + return handler; + } + + @Bean + public MessageChannel serviceChannel1() { + return new DirectChannel(); + } + + @Bean + @ServiceActivator(inputChannel = "serviceChannel1") + public MessageHandler service1() { + return new AbstractReplyProducingMessageHandler() { + + @Override + protected Object handleRequestMessage(Message requestMessage) { + return Math.random() * 10; + } + + }; + } + + @Bean + public MessageHandler distributor() { + RecipientListRouter router = new RecipientListRouter(); + router.setApplySequence(true); + router.setChannels(Arrays.asList(distributionChannel1(), distributionChannel2(), distributionChannel3())); + return router; + } + + @Bean + public MessageChannel distributionChannel1() { + return new DirectChannel(); + } + + @Bean + public MessageChannel distributionChannel2() { + return new DirectChannel(); + } + + @Bean + public MessageChannel distributionChannel3() { + return new DirectChannel(); + } + + @Bean + public MessageChannel distributionChannel() { + return new DirectChannel(); + } + + @Bean + @ServiceActivator(inputChannel = "distributionChannel") + public MessageHandler scatterGatherDistribution() { + ScatterGatherHandler handler = new ScatterGatherHandler(distributor(), gatherer1()); + handler.setOutputChannel(output()); + return handler; + } + + @Bean + @ServiceActivator(inputChannel = "distributionChannel1") + public MessageHandler distributionBridge1() { + BridgeHandler handler = new BridgeHandler(); + handler.setOutputChannel(serviceChannel1()); + return handler; + } + + @Bean + @ServiceActivator(inputChannel = "distributionChannel2") + public MessageHandler distributionBridge2() { + BridgeHandler handler = new BridgeHandler(); + handler.setOutputChannel(serviceChannel1()); + return handler; + } + + @Bean + @ServiceActivator(inputChannel = "distributionChannel3") + public MessageHandler distributionBridge3() { + BridgeHandler handler = new BridgeHandler(); + handler.setOutputChannel(serviceChannel1()); + return handler; + } + + @Bean + public MessageChannel inputAuctionWithGatherChannel() { + return new DirectChannel(); + } + + @Bean + public MessageChannel gatherChannel() { + return new DirectChannel(); + } + + @Bean + public SubscribableChannel scatterAuctionWithGatherChannel() { + PublishSubscribeChannel channel = new PublishSubscribeChannel(Executors.newCachedThreadPool()); + channel.setApplySequence(true); + return channel; + } + + @Bean + public MessageHandler gatherer2() { + return new AggregatingMessageHandler(new DefaultAggregatingMessageGroupProcessor(), + new SimpleMessageStore(), + new HeaderAttributeCorrelationStrategy(IntegrationMessageHeaderAccessor.CORRELATION_ID), + new MessageCountReleaseStrategy(3)); + } + + + @Bean + @ServiceActivator(inputChannel = "inputAuctionWithGatherChannel") + public MessageHandler scatterGatherAuctionWithGatherChannel() { + ScatterGatherHandler handler = new ScatterGatherHandler(scatterAuctionWithGatherChannel(), gatherer2()); + handler.setGatherChannel(gatherChannel()); + handler.setOutputChannel(output()); + return handler; + } + + @Bean + @ServiceActivator(inputChannel = "scatterAuctionWithGatherChannel") + public MessageHandler auctionWithGatherChannelBridge1() { + BridgeHandler handler = new BridgeHandler(); + handler.setOutputChannel(serviceChannel2()); + return handler; + } + + @Bean + @ServiceActivator(inputChannel = "scatterAuctionWithGatherChannel") + public MessageHandler auctionWithGatherChannelBridge2() { + BridgeHandler handler = new BridgeHandler(); + handler.setOutputChannel(serviceChannel2()); + return handler; + } + + @Bean + @ServiceActivator(inputChannel = "scatterAuctionWithGatherChannel") + public MessageHandler auctionWithGatherChannelBridge3() { + BridgeHandler handler = new BridgeHandler(); + handler.setOutputChannel(serviceChannel2()); + return handler; + } + + @Bean + @ServiceActivator(inputChannel = "scatterAuctionWithGatherChannel") + public MessageHandler auctionWithGatherChannelBridge4() { + BridgeHandler handler = new BridgeHandler(); + handler.setOutputChannel(serviceChannel2()); + return handler; + } + + @Bean + public MessageChannel serviceChannel2() { + return new DirectChannel(); + } + + @Bean + @ServiceActivator(inputChannel = "serviceChannel2") + public MessageHandler service2() { + return new AbstractReplyProducingMessageHandler() { + + { + setOutputChannel(gatherChannel()); + } + + @Override + protected Object handleRequestMessage(Message requestMessage) { + return Math.random(); + } + + }; + } + + } + +} diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/config/xml/MqttParserUtils.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/config/xml/MqttParserUtils.java index 7489554e35..41388831b8 100644 --- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/config/xml/MqttParserUtils.java +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/config/xml/MqttParserUtils.java @@ -57,8 +57,6 @@ public final class MqttParserUtils { } } IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "converter"); - IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup"); - IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "phase"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "send-timeout"); } diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/AbstractMqttMessageHandler.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/AbstractMqttMessageHandler.java index e8a508d827..23ce4c4512 100644 --- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/AbstractMqttMessageHandler.java +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/AbstractMqttMessageHandler.java @@ -16,7 +16,7 @@ package org.springframework.integration.mqtt.outbound; -import org.springframework.context.SmartLifecycle; +import org.springframework.context.Lifecycle; import org.springframework.integration.handler.AbstractMessageHandler; import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter; import org.springframework.integration.mqtt.support.MqttHeaders; @@ -32,7 +32,7 @@ import org.springframework.util.Assert; * @since 4.0 * */ -public abstract class AbstractMqttMessageHandler extends AbstractMessageHandler implements SmartLifecycle { +public abstract class AbstractMqttMessageHandler extends AbstractMessageHandler implements Lifecycle { private final String url; @@ -48,10 +48,6 @@ public abstract class AbstractMqttMessageHandler extends AbstractMessageHandler private boolean running; - private volatile int phase; - - private volatile boolean autoStartup; - private volatile int clientInstance; public AbstractMqttMessageHandler(String url, String clientId) { @@ -118,6 +114,7 @@ public abstract class AbstractMqttMessageHandler extends AbstractMessageHandler @Override public final void start() { this.doStart(); + this.running = true; } protected abstract void doStart(); @@ -125,6 +122,7 @@ public abstract class AbstractMqttMessageHandler extends AbstractMessageHandler @Override public final void stop() { this.doStop(); + this.running = false; } protected abstract void doStop(); @@ -134,30 +132,6 @@ public abstract class AbstractMqttMessageHandler extends AbstractMessageHandler return this.running; } - @Override - public int getPhase() { - return this.phase; - } - - public void setPhase(int phase) { - this.phase = phase; - } - - public void setAutoStartup(boolean autoStartup) { - this.autoStartup = autoStartup; - } - - @Override - public boolean isAutoStartup() { - return this.autoStartup; - } - - @Override - public void stop(Runnable callback) { - this.stop(); - callback.run(); - } - @Override protected void handleMessageInternal(Message message) throws Exception { this.connectIfNeeded(); diff --git a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/config/xml/MqttOutboundChannelAdapterParserTests.java b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/config/xml/MqttOutboundChannelAdapterParserTests.java index 0b075c18b1..89d494825c 100644 --- a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/config/xml/MqttOutboundChannelAdapterParserTests.java +++ b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/config/xml/MqttOutboundChannelAdapterParserTests.java @@ -68,8 +68,6 @@ public class MqttOutboundChannelAdapterParserTests { @Test public void testWithConverter() throws Exception { assertEquals("tcp://localhost:1883", TestUtils.getPropertyValue(withConverterHandler, "url")); - assertFalse(TestUtils.getPropertyValue(withConverterHandler, "autoStartup", Boolean.class)); - assertEquals(25, TestUtils.getPropertyValue(withConverterHandler, "phase")); assertEquals("foo", TestUtils.getPropertyValue(withConverterHandler, "clientId")); assertEquals("bar", TestUtils.getPropertyValue(withConverterHandler, "defaultTopic")); assertSame(converter, TestUtils.getPropertyValue(withConverterHandler, "converter")); @@ -90,8 +88,6 @@ public class MqttOutboundChannelAdapterParserTests { @Test public void testWithDefaultConverter() { assertEquals("tcp://localhost:1883", TestUtils.getPropertyValue(withDefaultConverterHandler, "url")); - assertFalse(TestUtils.getPropertyValue(withDefaultConverterHandler, "autoStartup", Boolean.class)); - assertEquals(25, TestUtils.getPropertyValue(withDefaultConverterHandler, "phase")); assertEquals("foo", TestUtils.getPropertyValue(withDefaultConverterHandler, "clientId")); assertEquals("bar", TestUtils.getPropertyValue(withDefaultConverterHandler, "defaultTopic")); assertEquals(1, TestUtils.getPropertyValue(withDefaultConverterHandler, "defaultQos")); diff --git a/src/reference/docbook/message-routing.xml b/src/reference/docbook/message-routing.xml index b71b571da3..62c84c00f6 100644 --- a/src/reference/docbook/message-routing.xml +++ b/src/reference/docbook/message-routing.xml @@ -10,5 +10,6 @@ + diff --git a/src/reference/docbook/scatter-gather.xml b/src/reference/docbook/scatter-gather.xml new file mode 100644 index 0000000000..52b603d3d5 --- /dev/null +++ b/src/reference/docbook/scatter-gather.xml @@ -0,0 +1,26 @@ + +
+ Scatter-Gather + +
+ Introduction + + TBD + +
+ +
+ Functionality + + TBD + +
+ +
+ Configuring a Scatter-Gather + + TBD + +
+ +
diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml index d0bd1573cb..d0baec2766 100644 --- a/src/reference/docbook/whats-new.xml +++ b/src/reference/docbook/whats-new.xml @@ -27,6 +27,13 @@ See for more information. +
+ Scatter-Gather EIP pattern + + The Scatter-Gather EIP pattern is now implemented. + See for more information. + +
BoonJsonObjectMapper