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

INT-4005: Fix channels' capacity with store

Browse Source 
JIRA: https://jira.spring.io/browse/INT-4005

* The `<int:queue>` lets declare the `capacity` attribute together
with either `message-store` or `ref` attributes, which it is not
correct

* The `<int:queue>` lets declare the `capacity` attribute together
with either `message-store` or `ref` attributes, which it is not
correct
* The `priority-queue` lets declare the `capacity` attribute
together with the `message-store` attribute, which it is not
correct
* Reference documentation fixed for `channel.adoc` and `jdbc.adoc`,
doing mention about these restrictions

* Minor changes about conventions and formats applied.

Code polishing
This commit is contained in:
Manuel Jordan
2016-04-27 16:11:14 -05:00
committed by Artem Bilan
parent ca0fa1b182
commit 8cfb7414ad
11 changed files with 177 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
spring-integration-core/src/main/java/org/springframework/integration/config/xml/PointToPointChannelParser.java
View File

@@ -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 <dispatcher/> child element is present.",
"The 'fixed-subscriber' attribute is not allowed" +
" when a <dispatcher/> 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

11
spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-4.3.xsd
View File

@@ -202,10 +202,7 @@
<xsd:annotation>
<xsd:documentation>
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.
</xsd:documentation>
</xsd:annotation>
<xsd:attribute name="capacity" type="xsd:string">
@@ -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).
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
@@ -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).
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
@@ -243,7 +242,7 @@
<xsd:annotation>
<xsd:documentation>
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).
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">

2
spring-integration-core/src/test/java/org/springframework/integration/config/ChannelWithMessageStoreParserTests-context.xml
View File

@@ -10,7 +10,7 @@
<context:property-placeholder location="classpath:org/springframework/integration/config/ms.properties"/>
<channel id="output">
<queue capacity="5" message-store="${ms}" />
<queue message-store="${ms}" />
</channel>
<channel id="input" />

23
spring-integration-core/src/test/java/org/springframework/integration/config/InvalidChannelWithMessageStoreParserTests.java → spring-integration-core/src/test/java/org/springframework/integration/config/InvalidPriorityChannelParserTests.java
View File

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

17
spring-integration-core/src/test/java/org/springframework/integration/config/InvalidPriorityChannelWithComparatorAndMessageStoreParserTests.xml Normal file
View File

@@ -0,0 +1,17 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/integration" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:beans="http://www.springframework.org/schema/beans"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration.xsd">
<channel id="output">
<priority-queue comparator="comparator" message-store="messageStore" />
</channel>
<beans:bean id="messageStore" class="org.springframework.integration.store.SimpleMessageStore" />
<beans:bean id="comparator" class="org.springframework.integration.aggregator.MessageSequenceComparator" />
</beans:beans>

15
spring-integration-core/src/test/java/org/springframework/integration/config/InvalidPriorityChannelWithMessageStoreAndCapacityParserTests.xml Normal file
View File

@@ -0,0 +1,15 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/integration" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:beans="http://www.springframework.org/schema/beans"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration.xsd">
<channel id="output">
<priority-queue message-store="messageStore" capacity="10" />
</channel>
<beans:bean id="messageStore" class="org.springframework.integration.store.SimpleMessageStore" />
</beans:beans>

61
spring-integration-core/src/test/java/org/springframework/integration/config/InvalidQueueChannelParserTests.java Normal file
View File

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

2
spring-integration-core/src/test/java/org/springframework/integration/config/InvalidChannelWithMessageStoreParserTests-context.xml → spring-integration-core/src/test/java/org/springframework/integration/config/InvalidQueueChannelWithMessageStoreAndCapacityParserTests.xml
View File

@@ -7,7 +7,7 @@
http://www.springframework.org/schema/integration/spring-integration.xsd">
<channel id="output">
<queue message-store="messageStore" ref="queue" />
<queue message-store="messageStore" capacity="10" />
</channel>
<beans:bean id="messageStore" class="org.springframework.integration.store.SimpleMessageStore" />

15
spring-integration-core/src/test/java/org/springframework/integration/config/InvalidQueueChannelWithRefAndCapacityParserTests.xml Normal file
View File

@@ -0,0 +1,15 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/integration" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:beans="http://www.springframework.org/schema/beans"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration.xsd">
<channel id="output">
<queue ref="queue" capacity="10" />
</channel>
<beans:bean id="queue" class="java.util.concurrent.LinkedBlockingQueue" />
</beans:beans>

17
spring-integration-core/src/test/java/org/springframework/integration/config/InvalidQueueChannelWithRefAndMessageStoreParserTests.xml Normal file
View File

@@ -0,0 +1,17 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/integration" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:beans="http://www.springframework.org/schema/beans"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration.xsd">
<channel id="output">
<queue ref="queue" message-store="messageStore" />
</channel>
<beans:bean id="queue" class="java.util.concurrent.LinkedBlockingQueue" />
<beans:bean id="messageStore" class="org.springframework.integration.store.SimpleMessageStore" />
</beans:beans>

4
src/reference/asciidoc/channel.adoc
View File

@@ -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 <<message-store>>.
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 <<channel-configuration-queuechannel>>.