diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/PointToPointChannelParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/PointToPointChannelParser.java index 07386c92aa..218c623d94 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/PointToPointChannelParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/PointToPointChannelParser.java @@ -39,6 +39,7 @@ import org.springframework.util.xml.DomUtils; * @author Oleg Zhurakousky * @author Gary Russell * @author Artem Bilan + * @author Manuel Jordan */ public class PointToPointChannelParser extends AbstractChannelParser { @@ -55,23 +56,29 @@ public class PointToPointChannelParser extends AbstractChannelParser { builder = BeanDefinitionBuilder.genericBeanDefinition(QueueChannel.class); boolean hasStoreRef = this.parseStoreRef(builder, queueElement, channel, false); boolean hasQueueRef = this.parseQueueRef(builder, queueElement); - if (!hasStoreRef) { + if (!hasStoreRef || !hasQueueRef) { boolean hasCapacity = this.parseQueueCapacity(builder, queueElement); if (hasCapacity && hasQueueRef) { parserContext.getReaderContext().error( - "The 'capacity' attribute is not allowed" + " when providing a 'ref' to a custom queue.", + "The 'capacity' attribute is not allowed when providing a 'ref' to a custom queue.", + element); + } + if (hasCapacity && hasStoreRef) { + parserContext.getReaderContext().error( + "The 'capacity' attribute is not allowed" + + " when providing a 'message-store' to a custom MessageGroupStore.", element); } } if (hasStoreRef && hasQueueRef) { parserContext.getReaderContext().error( - "The 'message-store' attribute is not allowed" + " when providing a 'ref' to a custom queue.", + "The 'message-store' attribute is not allowed when providing a 'ref' to a custom queue.", element); } } else if ((queueElement = DomUtils.getChildElementByTagName(element, "priority-queue")) != null) { builder = BeanDefinitionBuilder.genericBeanDefinition(PriorityChannel.class); - this.parseQueueCapacity(builder, queueElement); + boolean hasCapacity = this.parseQueueCapacity(builder, queueElement); String comparatorRef = queueElement.getAttribute("comparator"); if (StringUtils.hasText(comparatorRef)) { builder.addConstructorArgReference(comparatorRef); @@ -79,9 +86,14 @@ public class PointToPointChannelParser extends AbstractChannelParser { if (parseStoreRef(builder, queueElement, channel, true)) { if (StringUtils.hasText(comparatorRef)) { parserContext.getReaderContext().error( - "The 'message-store' attribute is not allowed" + " when providing a 'comparator' to a priority queue.", + "The 'message-store' attribute is not allowed" + + " when providing a 'comparator' to a priority queue.", element); } + if (hasCapacity) { + parserContext.getReaderContext().error("The 'capacity' attribute is not allowed" + + " when providing a 'message-store' to a custom MessageGroupStore.", element); + } builder.getRawBeanDefinition().setBeanClass(QueueChannel.class); } @@ -116,7 +128,8 @@ public class PointToPointChannelParser extends AbstractChannelParser { else { if (isFixedSubscriber) { parserContext.getReaderContext().error( - "The 'fixed-subscriber' attribute is not allowed when a child element is present.", + "The 'fixed-subscriber' attribute is not allowed" + + " when a child element is present.", element); } // configure either an ExecutorChannel or DirectChannel based on existence of 'task-executor' @@ -128,12 +141,14 @@ public class PointToPointChannelParser extends AbstractChannelParser { else { builder = BeanDefinitionBuilder.genericBeanDefinition(DirectChannel.class); } - // unless the 'load-balancer' attribute is explicitly set to 'none' or 'load-balancer-ref' is explicitly configured, + // unless the 'load-balancer' attribute is explicitly set to 'none' + // or 'load-balancer-ref' is explicitly configured, // configure the default RoundRobinLoadBalancingStrategy String loadBalancer = dispatcherElement.getAttribute("load-balancer"); String loadBalancerRef = dispatcherElement.getAttribute("load-balancer-ref"); if (StringUtils.hasText(loadBalancer) && StringUtils.hasText(loadBalancerRef)) { - parserContext.getReaderContext().error("'load-balancer' and 'load-balancer-ref' are mutually exclusive", element); + parserContext.getReaderContext().error("'load-balancer' and 'load-balancer-ref' are mutually exclusive", + element); } if (StringUtils.hasText(loadBalancerRef)) { builder.addConstructorArgReference(loadBalancerRef); @@ -168,7 +183,8 @@ public class PointToPointChannelParser extends AbstractChannelParser { return false; } - private boolean parseStoreRef(BeanDefinitionBuilder builder, Element queueElement, String channel, boolean priority) { + private boolean parseStoreRef(BeanDefinitionBuilder builder, Element queueElement, String channel, + boolean priority) { String storeRef = queueElement.getAttribute("message-store"); if (StringUtils.hasText(storeRef)) { BeanDefinitionBuilder queueBuilder = BeanDefinitionBuilder diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-4.3.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-4.3.xsd index d3635ff66b..4e3e897b85 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-4.3.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-4.3.xsd @@ -202,10 +202,7 @@ Defines a queue for messages. If 'capacity' is specified, it will be a - bounded queue. - A custom Queue - implementation can be injected using the 'ref' - attribute. + bounded queue. A custom Queue implementation can be injected using the 'ref' attribute. @@ -214,6 +211,8 @@ Capacity for this queue. Default capacity is 0 which means this queue will accumulate as many messages as available resources allow. + This attribute is mutually exclusive with the "message-store" + and "ref" attributes (only one can be specified). @@ -230,7 +229,7 @@ message store has a way to specify a region or similar additional tag for messages. This attribute is mutually - exclusive with the "ref" attribute (only one can be specified). + exclusive with the "capacity" and "ref" attributes (only one can be specified). @@ -243,7 +242,7 @@ Reference to a Queue that can be used to buffer the messages. This attribute is - mutually exclusive with the "message-store" attribute (only one can be specified). + mutually exclusive with the "capacity" and "message-store" attributes (only one can be specified). diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/ChannelWithMessageStoreParserTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/ChannelWithMessageStoreParserTests-context.xml index 6abffbf024..0d41613d60 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/ChannelWithMessageStoreParserTests-context.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/ChannelWithMessageStoreParserTests-context.xml @@ -10,7 +10,7 @@ - + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/InvalidChannelWithMessageStoreParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/InvalidPriorityChannelParserTests.java similarity index 53% rename from spring-integration-core/src/test/java/org/springframework/integration/config/InvalidChannelWithMessageStoreParserTests.java rename to spring-integration-core/src/test/java/org/springframework/integration/config/InvalidPriorityChannelParserTests.java index d38461cf00..be726b4bbe 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/InvalidChannelWithMessageStoreParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/InvalidPriorityChannelParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * Copyright 2002-2016 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. @@ -26,17 +26,28 @@ import org.springframework.context.support.ClassPathXmlApplicationContext; /** * @author Dave Syer + * @author Manuel Jordan + * @since 4.3 */ -public class InvalidChannelWithMessageStoreParserTests { +public class InvalidPriorityChannelParserTests { @Rule public ExpectedException exception = ExpectedException.none(); @Test - public void testRefAndStoreIllegal() throws Exception { - exception.expect(BeanDefinitionParsingException.class); - exception.expectMessage(Matchers.containsString("'message-store' attribute is not allowed")); - new ClassPathXmlApplicationContext(getClass().getSimpleName() + "-context.xml", getClass()).close(); + public void testMessageStoreAndCapacityIllegal() throws Exception { + this.exception.expect(BeanDefinitionParsingException.class); + this.exception.expectMessage(Matchers.containsString("'capacity' attribute is not allowed")); + new ClassPathXmlApplicationContext("InvalidPriorityChannelWithMessageStoreAndCapacityParserTests.xml", + getClass()).close(); + } + + @Test + public void testComparatorAndMessageStoreIllegal() throws Exception { + this.exception.expect(BeanDefinitionParsingException.class); + this.exception.expectMessage(Matchers.containsString("The 'message-store' attribute is not allowed")); + new ClassPathXmlApplicationContext("InvalidPriorityChannelWithComparatorAndMessageStoreParserTests.xml", + getClass()).close(); } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/InvalidPriorityChannelWithComparatorAndMessageStoreParserTests.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/InvalidPriorityChannelWithComparatorAndMessageStoreParserTests.xml new file mode 100644 index 0000000000..125dc08b17 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/InvalidPriorityChannelWithComparatorAndMessageStoreParserTests.xml @@ -0,0 +1,17 @@ + + + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/InvalidPriorityChannelWithMessageStoreAndCapacityParserTests.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/InvalidPriorityChannelWithMessageStoreAndCapacityParserTests.xml new file mode 100644 index 0000000000..ec5f6e99ae --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/InvalidPriorityChannelWithMessageStoreAndCapacityParserTests.xml @@ -0,0 +1,15 @@ + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/InvalidQueueChannelParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/InvalidQueueChannelParserTests.java new file mode 100644 index 0000000000..b4bb15fe49 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/InvalidQueueChannelParserTests.java @@ -0,0 +1,61 @@ +/* + * Copyright 2002-2016 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; + +import org.hamcrest.Matchers; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.ExpectedException; + +import org.springframework.beans.factory.parsing.BeanDefinitionParsingException; +import org.springframework.context.support.ClassPathXmlApplicationContext; + +/** + * @author Dave Syer + * @author Manuel Jordan + * @since 4.3 + */ +public class InvalidQueueChannelParserTests { + + @Rule + public ExpectedException exception = ExpectedException.none(); + + @Test + public void testMessageStoreAndCapacityIllegal() throws Exception { + this.exception.expect(BeanDefinitionParsingException.class); + this.exception.expectMessage(Matchers.containsString("'capacity' attribute is not allowed")); + new ClassPathXmlApplicationContext("InvalidQueueChannelWithMessageStoreAndCapacityParserTests.xml", + getClass()).close(); + } + + @Test + public void testRefAndCapacityIllegal() throws Exception { + this.exception.expect(BeanDefinitionParsingException.class); + this.exception.expectMessage(Matchers.containsString("'capacity' attribute is not allowed")); + new ClassPathXmlApplicationContext("InvalidQueueChannelWithRefAndCapacityParserTests.xml", getClass()) + .close(); + } + + @Test + public void testRefAndMessageStoreIllegal() throws Exception { + this.exception.expect(BeanDefinitionParsingException.class); + this.exception.expectMessage(Matchers.containsString("'message-store' attribute is not allowed")); + new ClassPathXmlApplicationContext("InvalidQueueChannelWithRefAndMessageStoreParserTests.xml", getClass()) + .close(); + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/InvalidChannelWithMessageStoreParserTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/InvalidQueueChannelWithMessageStoreAndCapacityParserTests.xml similarity index 92% rename from spring-integration-core/src/test/java/org/springframework/integration/config/InvalidChannelWithMessageStoreParserTests-context.xml rename to spring-integration-core/src/test/java/org/springframework/integration/config/InvalidQueueChannelWithMessageStoreAndCapacityParserTests.xml index 0ef3e25c57..ecfb537ba8 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/InvalidChannelWithMessageStoreParserTests-context.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/InvalidQueueChannelWithMessageStoreAndCapacityParserTests.xml @@ -7,7 +7,7 @@ http://www.springframework.org/schema/integration/spring-integration.xsd"> - + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/InvalidQueueChannelWithRefAndCapacityParserTests.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/InvalidQueueChannelWithRefAndCapacityParserTests.xml new file mode 100644 index 0000000000..af1d71566c --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/InvalidQueueChannelWithRefAndCapacityParserTests.xml @@ -0,0 +1,15 @@ + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/InvalidQueueChannelWithRefAndMessageStoreParserTests.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/InvalidQueueChannelWithRefAndMessageStoreParserTests.xml new file mode 100644 index 0000000000..dc44de5880 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/InvalidQueueChannelWithRefAndMessageStoreParserTests.xml @@ -0,0 +1,17 @@ + + + + + + + + + + + + diff --git a/src/reference/asciidoc/channel.adoc b/src/reference/asciidoc/channel.adoc index fb67d46ad1..4d38b363d9 100644 --- a/src/reference/asciidoc/channel.adoc +++ b/src/reference/asciidoc/channel.adoc @@ -474,6 +474,8 @@ Since a `QueueChannel` provides the capability to buffer Messages, but does so i To mitigate this risk, a `QueueChannel` may be backed by a persistent implementation of the `MessageGroupStore` strategy interface. For more details on `MessageGroupStore` and `MessageStore` see <>. +IMPORTANT: The `capacity` attribute is not allowed when the `message-store` attribute is used. + When a `QueueChannel` receives a Message, it will add it to the Message Store, and when a Message is polled from a `QueueChannel`, it is removed from the Message Store. By default, a `QueueChannel` stores its Messages in an in-memory Queue and can therefore lead to the lost message scenario mentioned above. @@ -629,7 +631,7 @@ The following example demonstrates all of these: ---- -Since _version 4.0_, the `priority-channel` child element supports the `message-store` option (`comparator` is not allowed in that case). +Since _version 4.0_, the `priority-channel` child element supports the `message-store` option (`comparator` and `capacity` are not allowed in that case). The message store must be a `PriorityCapableChannelMessageStore` and, in this case, the namespace parser will declare a `QueueChannel` instead of a `PriorityChannel`. Implementations of the `PriorityCapableChannelMessageStore` are currently provided for `Redis`, `JDBC` and `MongoDB`. See <>.