The <channel/> element is now used for creating all Point-to-Point channel types. It accepts a queue sub-element (options are: <queue/>, <priority-queue/>, or <rendezvous-queue/>). If no queue sub-element is provided, the channel type will be a DirectChannel.
This commit is contained in:
@@ -32,15 +32,18 @@ import org.springframework.integration.message.MessagePriority;
|
||||
*/
|
||||
public class PriorityChannel extends QueueChannel {
|
||||
|
||||
private static final int DEFAULT_MAX_CAPACITY = Integer.MAX_VALUE;
|
||||
|
||||
|
||||
private final Semaphore semaphore;
|
||||
|
||||
|
||||
/**
|
||||
* Create a channel with the specified queue capacity.
|
||||
* Priority will be based upon the provided {@link Comparator}.
|
||||
* Priority will be determined by the provided {@link Comparator}.
|
||||
*/
|
||||
public PriorityChannel(int capacity, Comparator<Message<?>> comparator) {
|
||||
super(new PriorityBlockingQueue<Message<?>>(capacity, comparator));
|
||||
super(new PriorityBlockingQueue<Message<?>>(11, comparator));
|
||||
this.semaphore = new Semaphore(capacity, true);
|
||||
}
|
||||
|
||||
@@ -53,11 +56,19 @@ public class PriorityChannel extends QueueChannel {
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a channel with the default queue capacity and dispatcher policy.
|
||||
* Create a channel with the default queue capacity of {@link Integer#MAX_VALUE}.
|
||||
* Priority will be determined by the provided {@link Comparator}.
|
||||
*/
|
||||
public PriorityChannel(Comparator<Message<?>> comparator) {
|
||||
this(DEFAULT_MAX_CAPACITY, comparator);
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a channel with the default queue capacity of {@link Integer#MAX_VALUE}.
|
||||
* Priority will be based on the value of {@link MessageHeader#getPriority()}.
|
||||
*/
|
||||
public PriorityChannel() {
|
||||
this(DEFAULT_CAPACITY, new MessagePriorityComparator());
|
||||
this(DEFAULT_MAX_CAPACITY, new MessagePriorityComparator());
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -37,9 +37,6 @@ import org.springframework.util.Assert;
|
||||
*/
|
||||
public class QueueChannel extends AbstractPollableChannel {
|
||||
|
||||
public static final int DEFAULT_CAPACITY = 100;
|
||||
|
||||
|
||||
private final BlockingQueue<Message<?>> queue;
|
||||
|
||||
|
||||
@@ -61,11 +58,12 @@ public class QueueChannel extends AbstractPollableChannel {
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a channel with the default queue capacity.
|
||||
* @see #DEFAULT_CAPACITY
|
||||
* Create a channel with "unbounded" queue capacity. The actual capacity value is
|
||||
* {@link Integer#MAX_VALUE}. Note that a bounded queue is recommended, since an
|
||||
* unbounded queue may lead to OutOfMemoryErrors.
|
||||
*/
|
||||
public QueueChannel() {
|
||||
this(DEFAULT_CAPACITY);
|
||||
this(new LinkedBlockingQueue<Message<?>>());
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -17,44 +17,36 @@
|
||||
package org.springframework.integration.config;
|
||||
|
||||
import org.w3c.dom.Element;
|
||||
import org.w3c.dom.Node;
|
||||
import org.w3c.dom.NodeList;
|
||||
|
||||
import org.springframework.beans.factory.config.RuntimeBeanReference;
|
||||
import org.springframework.beans.factory.parsing.BeanComponentDefinition;
|
||||
import org.springframework.beans.factory.support.AbstractBeanDefinition;
|
||||
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
|
||||
import org.springframework.beans.factory.support.ManagedList;
|
||||
import org.springframework.beans.factory.support.RootBeanDefinition;
|
||||
import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
|
||||
import org.springframework.beans.factory.xml.AbstractBeanDefinitionParser;
|
||||
import org.springframework.beans.factory.xml.ParserContext;
|
||||
import org.springframework.integration.channel.interceptor.MessageSelectingInterceptor;
|
||||
import org.springframework.integration.message.selector.PayloadTypeSelector;
|
||||
import org.springframework.util.StringUtils;
|
||||
import org.springframework.util.xml.DomUtils;
|
||||
|
||||
/**
|
||||
* Base class for channel parsers.
|
||||
*
|
||||
* @author Mark Fisher
|
||||
*/
|
||||
public abstract class AbstractChannelParser extends AbstractSingleBeanDefinitionParser {
|
||||
|
||||
@Override
|
||||
protected abstract Class<?> getBeanClass(Element element);
|
||||
|
||||
protected void postProcess(BeanDefinitionBuilder builder, Element element) {
|
||||
}
|
||||
public abstract class AbstractChannelParser extends AbstractBeanDefinitionParser {
|
||||
|
||||
@Override
|
||||
@SuppressWarnings("unchecked")
|
||||
protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
|
||||
protected AbstractBeanDefinition parseInternal(Element element, ParserContext parserContext) {
|
||||
BeanDefinitionBuilder builder = this.buildBeanDefinition(element, parserContext);
|
||||
ManagedList interceptors = null;
|
||||
NodeList childNodes = element.getChildNodes();
|
||||
for (int i = 0; i < childNodes.getLength(); i++) {
|
||||
Node child = childNodes.item(i);
|
||||
if (child.getNodeType() == Node.ELEMENT_NODE && child.getLocalName().equals("interceptors")) {
|
||||
ChannelInterceptorParser interceptorParser = new ChannelInterceptorParser();
|
||||
interceptors = interceptorParser.parseInterceptors((Element) child, parserContext);
|
||||
}
|
||||
Element interceptorsElement = DomUtils.getChildElementByTagName(element, "interceptors");
|
||||
if (interceptorsElement != null) {
|
||||
ChannelInterceptorParser interceptorParser = new ChannelInterceptorParser();
|
||||
interceptors = interceptorParser.parseInterceptors(interceptorsElement, parserContext);
|
||||
}
|
||||
if (interceptors == null) {
|
||||
interceptors = new ManagedList();
|
||||
@@ -75,7 +67,16 @@ public abstract class AbstractChannelParser extends AbstractSingleBeanDefinition
|
||||
interceptors.add(new RuntimeBeanReference(interceptorBeanName));
|
||||
}
|
||||
builder.addPropertyValue("interceptors", interceptors);
|
||||
this.postProcess(builder, element);
|
||||
return builder.getBeanDefinition();
|
||||
}
|
||||
|
||||
/**
|
||||
* Subclasses must implement this method to create the bean definition.
|
||||
* The class must be defined, and any implementation-specific constructor
|
||||
* arguments or properties should be configured. This base class will
|
||||
* configure the interceptors including the 'datatype' interceptor if
|
||||
* the 'datatype' attribute is defined on the channel element.
|
||||
*/
|
||||
protected abstract BeanDefinitionBuilder buildBeanDefinition(Element element, ParserContext parserContext);
|
||||
|
||||
}
|
||||
|
||||
@@ -1,35 +0,0 @@
|
||||
/*
|
||||
* Copyright 2002-2008 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.w3c.dom.Element;
|
||||
|
||||
import org.springframework.integration.channel.DirectChannel;
|
||||
|
||||
/**
|
||||
* Parser for the <direct-channel> element.
|
||||
*
|
||||
* @author Mark Fisher
|
||||
*/
|
||||
public class DirectChannelParser extends AbstractChannelParser {
|
||||
|
||||
@Override
|
||||
protected Class<?> getBeanClass(Element element) {
|
||||
return DirectChannel.class;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -49,13 +49,13 @@ public class IntegrationNamespaceHandler extends NamespaceHandlerSupport {
|
||||
public void init() {
|
||||
registerBeanDefinitionParser("message-bus", new MessageBusParser());
|
||||
registerBeanDefinitionParser("annotation-driven", new AnnotationDrivenParser());
|
||||
registerBeanDefinitionParser("channel", new QueueChannelParser());
|
||||
registerBeanDefinitionParser("queue-channel", new QueueChannelParser());
|
||||
registerBeanDefinitionParser("publish-subscribe-channel", new PublishSubscribeChannelParser());
|
||||
registerBeanDefinitionParser("direct-channel", new DirectChannelParser());
|
||||
registerBeanDefinitionParser("priority-channel", new PriorityChannelParser());
|
||||
registerBeanDefinitionParser("rendezvous-channel", new RendezvousChannelParser());
|
||||
registerBeanDefinitionParser("channel", new PointToPointChannelParser());
|
||||
registerBeanDefinitionParser("queue-channel", new PointToPointChannelParser());
|
||||
registerBeanDefinitionParser("direct-channel", new PointToPointChannelParser());
|
||||
registerBeanDefinitionParser("priority-channel", new PointToPointChannelParser());
|
||||
registerBeanDefinitionParser("rendezvous-channel", new PointToPointChannelParser());
|
||||
registerBeanDefinitionParser("thread-local-channel", new ThreadLocalChannelParser());
|
||||
registerBeanDefinitionParser("publish-subscribe-channel", new PublishSubscribeChannelParser());
|
||||
registerBeanDefinitionParser("service-activator", new ServiceActivatorParser());
|
||||
registerBeanDefinitionParser("channel-adapter", new ChannelAdapterParser());
|
||||
registerBeanDefinitionParser("gateway", new GatewayParser());
|
||||
|
||||
@@ -0,0 +1,74 @@
|
||||
/*
|
||||
* Copyright 2002-2008 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.w3c.dom.Element;
|
||||
|
||||
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
|
||||
import org.springframework.beans.factory.xml.ParserContext;
|
||||
import org.springframework.integration.channel.DirectChannel;
|
||||
import org.springframework.integration.channel.PriorityChannel;
|
||||
import org.springframework.integration.channel.QueueChannel;
|
||||
import org.springframework.integration.channel.RendezvousChannel;
|
||||
import org.springframework.util.StringUtils;
|
||||
import org.springframework.util.xml.DomUtils;
|
||||
|
||||
/**
|
||||
* Parser for the <channel> element.
|
||||
*
|
||||
* @author Mark Fisher
|
||||
*/
|
||||
public class PointToPointChannelParser extends AbstractChannelParser {
|
||||
|
||||
private static final String CAPACITY_ATTRIBUTE = "capacity";
|
||||
|
||||
|
||||
@Override
|
||||
protected BeanDefinitionBuilder buildBeanDefinition(Element element, ParserContext parserContext) {
|
||||
BeanDefinitionBuilder builder = null;
|
||||
Element queueElement = null;
|
||||
if ((queueElement = DomUtils.getChildElementByTagName(element, "queue")) != null) {
|
||||
builder = BeanDefinitionBuilder.genericBeanDefinition(QueueChannel.class);
|
||||
this.parseQueueCapacity(builder, queueElement);
|
||||
}
|
||||
else if ((queueElement = DomUtils.getChildElementByTagName(element, "priority-queue")) != null) {
|
||||
builder = BeanDefinitionBuilder.genericBeanDefinition(PriorityChannel.class);
|
||||
this.parseQueueCapacity(builder, queueElement);
|
||||
String comparatorRef = queueElement.getAttribute("comparator");
|
||||
if (StringUtils.hasText(comparatorRef)) {
|
||||
builder.addConstructorArgReference(comparatorRef);
|
||||
}
|
||||
}
|
||||
else if ((queueElement = DomUtils.getChildElementByTagName(element, "rendezvous-queue")) != null) {
|
||||
builder = BeanDefinitionBuilder.genericBeanDefinition(RendezvousChannel.class);
|
||||
}
|
||||
else {
|
||||
builder = BeanDefinitionBuilder.genericBeanDefinition(DirectChannel.class);
|
||||
}
|
||||
return builder;
|
||||
}
|
||||
|
||||
private void parseQueueCapacity(BeanDefinitionBuilder builder, Element queueElement) {
|
||||
String capacity = queueElement.getAttribute(CAPACITY_ATTRIBUTE);
|
||||
if (StringUtils.hasText(capacity)) {
|
||||
if (!capacity.equals("UNBOUNDED")) {
|
||||
builder.addConstructorArgValue(Integer.valueOf(capacity));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,46 +0,0 @@
|
||||
/*
|
||||
* Copyright 2002-2008 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.w3c.dom.Element;
|
||||
|
||||
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
|
||||
import org.springframework.integration.channel.PriorityChannel;
|
||||
import org.springframework.util.StringUtils;
|
||||
|
||||
/**
|
||||
* Parser for the <priority-channel> element.
|
||||
*
|
||||
* @author Mark Fisher
|
||||
*/
|
||||
public class PriorityChannelParser extends QueueChannelParser {
|
||||
|
||||
@Override
|
||||
protected Class<?> getBeanClass(Element element) {
|
||||
return PriorityChannel.class;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void postProcess(BeanDefinitionBuilder builder, Element element) {
|
||||
super.postProcess(builder, element);
|
||||
String comparator = element.getAttribute("comparator");
|
||||
if (StringUtils.hasText(comparator)) {
|
||||
builder.addConstructorArgReference(comparator);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -19,6 +19,7 @@ package org.springframework.integration.config;
|
||||
import org.w3c.dom.Element;
|
||||
|
||||
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
|
||||
import org.springframework.beans.factory.xml.ParserContext;
|
||||
import org.springframework.integration.channel.PublishSubscribeChannel;
|
||||
import org.springframework.util.StringUtils;
|
||||
|
||||
@@ -30,17 +31,14 @@ import org.springframework.util.StringUtils;
|
||||
public class PublishSubscribeChannelParser extends AbstractChannelParser {
|
||||
|
||||
@Override
|
||||
protected Class<?> getBeanClass(Element element) {
|
||||
return PublishSubscribeChannel.class;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void postProcess(BeanDefinitionBuilder builder, Element element) {
|
||||
protected BeanDefinitionBuilder buildBeanDefinition(Element element, ParserContext parserContext) {
|
||||
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(PublishSubscribeChannel.class);
|
||||
String taskExecutorRef = element.getAttribute("task-executor");
|
||||
if (StringUtils.hasText(taskExecutorRef)) {
|
||||
builder.addConstructorArgReference(taskExecutorRef);
|
||||
}
|
||||
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "apply-sequence");
|
||||
return builder;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -1,45 +0,0 @@
|
||||
/*
|
||||
* Copyright 2002-2008 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.w3c.dom.Element;
|
||||
|
||||
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
|
||||
import org.springframework.integration.channel.QueueChannel;
|
||||
import org.springframework.util.StringUtils;
|
||||
|
||||
/**
|
||||
* Parser for the <queue-channel> element.
|
||||
*
|
||||
* @author Mark Fisher
|
||||
*/
|
||||
public class QueueChannelParser extends AbstractChannelParser {
|
||||
|
||||
@Override
|
||||
protected Class<?> getBeanClass(Element element) {
|
||||
return QueueChannel.class;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void postProcess(BeanDefinitionBuilder builder, Element element) {
|
||||
String capacityAttribute = element.getAttribute("capacity");
|
||||
int capacity = (StringUtils.hasText(capacityAttribute)) ?
|
||||
Integer.parseInt(capacityAttribute) : QueueChannel.DEFAULT_CAPACITY;
|
||||
builder.addConstructorArgValue(capacity);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,35 +0,0 @@
|
||||
/*
|
||||
* Copyright 2002-2008 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.w3c.dom.Element;
|
||||
|
||||
import org.springframework.integration.channel.RendezvousChannel;
|
||||
|
||||
/**
|
||||
* Parser for the <rendezvous-channel> element.
|
||||
*
|
||||
* @author Mark Fisher
|
||||
*/
|
||||
public class RendezvousChannelParser extends AbstractChannelParser {
|
||||
|
||||
@Override
|
||||
protected Class<?> getBeanClass(Element element) {
|
||||
return RendezvousChannel.class;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -18,6 +18,8 @@ package org.springframework.integration.config;
|
||||
|
||||
import org.w3c.dom.Element;
|
||||
|
||||
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
|
||||
import org.springframework.beans.factory.xml.ParserContext;
|
||||
import org.springframework.integration.channel.ThreadLocalChannel;
|
||||
|
||||
/**
|
||||
@@ -28,8 +30,8 @@ import org.springframework.integration.channel.ThreadLocalChannel;
|
||||
public class ThreadLocalChannelParser extends AbstractChannelParser {
|
||||
|
||||
@Override
|
||||
protected Class<?> getBeanClass(Element element) {
|
||||
return ThreadLocalChannel.class;
|
||||
protected BeanDefinitionBuilder buildBeanDefinition(Element element, ParserContext parserContext) {
|
||||
return BeanDefinitionBuilder.genericBeanDefinition(ThreadLocalChannel.class);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -47,32 +47,86 @@
|
||||
</xsd:complexType>
|
||||
</xsd:element>
|
||||
|
||||
<xsd:element name="channel" type="channelType">
|
||||
<xsd:element name="channel">
|
||||
<xsd:annotation>
|
||||
<xsd:documentation>
|
||||
Defines a generic channel type. The actual channel type
|
||||
will be determined by the channel factory set on the message bus.
|
||||
Defines a Point-to-Point MessageChannel.
|
||||
</xsd:documentation>
|
||||
</xsd:annotation>
|
||||
<xsd:complexType>
|
||||
<xsd:complexContent>
|
||||
<xsd:extension base="channelType">
|
||||
<xsd:sequence>
|
||||
<xsd:choice minOccurs="0" maxOccurs="1">
|
||||
<xsd:element ref="queue"/>
|
||||
<xsd:element ref="priority-queue"/>
|
||||
<xsd:element ref="rendezvous-queue"/>
|
||||
</xsd:choice>
|
||||
<xsd:element name="interceptors" type="channelInterceptorsType" minOccurs="0" maxOccurs="1"/>
|
||||
</xsd:sequence>
|
||||
</xsd:extension>
|
||||
</xsd:complexContent>
|
||||
</xsd:complexType>
|
||||
</xsd:element>
|
||||
|
||||
<xsd:element name="queue-channel" type="capacityChannelType">
|
||||
<xsd:annotation>
|
||||
<xsd:documentation>
|
||||
Defines a channel that buffers messages in a queue.
|
||||
</xsd:documentation>
|
||||
</xsd:annotation>
|
||||
<xsd:element name="queue">
|
||||
<xsd:complexType>
|
||||
<xsd:annotation>
|
||||
<xsd:documentation>
|
||||
Defines a bounded queue for messages.
|
||||
</xsd:documentation>
|
||||
</xsd:annotation>
|
||||
<xsd:attribute name="capacity" use="required">
|
||||
<xsd:simpleType>
|
||||
<xsd:union>
|
||||
<xsd:simpleType>
|
||||
<xsd:restriction base="xsd:integer"/>
|
||||
</xsd:simpleType>
|
||||
<xsd:simpleType>
|
||||
<xsd:restriction base="xsd:string">
|
||||
<xsd:enumeration value="UNBOUNDED"/>
|
||||
</xsd:restriction>
|
||||
</xsd:simpleType>
|
||||
</xsd:union>
|
||||
</xsd:simpleType>
|
||||
</xsd:attribute>
|
||||
</xsd:complexType>
|
||||
</xsd:element>
|
||||
|
||||
<xsd:element name="priority-queue">
|
||||
<xsd:complexType>
|
||||
<xsd:annotation>
|
||||
<xsd:documentation>
|
||||
Defines a queue with priority-ordering for message reception.
|
||||
</xsd:documentation>
|
||||
</xsd:annotation>
|
||||
<xsd:attribute name="capacity" type="xsd:integer" use="required"/>
|
||||
<xsd:attribute name="comparator" type="xsd:string"/>
|
||||
</xsd:complexType>
|
||||
</xsd:element>
|
||||
|
||||
<xsd:element name="rendezvous-queue">
|
||||
<xsd:complexType>
|
||||
<xsd:annotation>
|
||||
<xsd:documentation>
|
||||
Defines a rendezvous queue where a sender will block until the receiver arrives or vice-versa.
|
||||
</xsd:documentation>
|
||||
</xsd:annotation>
|
||||
</xsd:complexType>
|
||||
</xsd:element>
|
||||
|
||||
<xsd:element name="publish-subscribe-channel">
|
||||
<xsd:complexType>
|
||||
<xsd:annotation>
|
||||
<xsd:documentation>
|
||||
Defines a publish-subscribe channel that broadcasts to its targets.
|
||||
Defines a Publish-Subscribe channel that broadcasts messages to its subscribers.
|
||||
</xsd:documentation>
|
||||
</xsd:annotation>
|
||||
<xsd:complexContent>
|
||||
<xsd:extension base="channelType">
|
||||
<xsd:sequence>
|
||||
<xsd:element name="interceptors" type="channelInterceptorsType" minOccurs="0" maxOccurs="1"/>
|
||||
</xsd:sequence>
|
||||
<xsd:attribute name="task-executor" type="xsd:string"/>
|
||||
<xsd:attribute name="apply-sequence" type="xsd:string"/>
|
||||
</xsd:extension>
|
||||
@@ -80,55 +134,14 @@
|
||||
</xsd:complexType>
|
||||
</xsd:element>
|
||||
|
||||
<xsd:element name="direct-channel" type="channelType">
|
||||
<xsd:annotation>
|
||||
<xsd:documentation>
|
||||
Defines a channel that invokes its handlers directly in the sender's thread.
|
||||
</xsd:documentation>
|
||||
</xsd:annotation>
|
||||
</xsd:element>
|
||||
|
||||
<xsd:element name="priority-channel">
|
||||
<xsd:complexType>
|
||||
<xsd:annotation>
|
||||
<xsd:documentation>
|
||||
Defines a channel with priority-ordering for message reception.
|
||||
</xsd:documentation>
|
||||
</xsd:annotation>
|
||||
<xsd:complexContent>
|
||||
<xsd:extension base="capacityChannelType">
|
||||
<xsd:attribute name="comparator" type="xsd:string"/>
|
||||
</xsd:extension>
|
||||
</xsd:complexContent>
|
||||
</xsd:complexType>
|
||||
</xsd:element>
|
||||
|
||||
<xsd:element name="rendezvous-channel" type="channelType"/>
|
||||
|
||||
<xsd:element name="thread-local-channel" type="channelType"/>
|
||||
|
||||
<xsd:complexType name="capacityChannelType">
|
||||
<xsd:annotation>
|
||||
<xsd:documentation>
|
||||
Defines a channel with a configurable capacity.
|
||||
</xsd:documentation>
|
||||
</xsd:annotation>
|
||||
<xsd:complexContent>
|
||||
<xsd:extension base="channelType">
|
||||
<xsd:attribute name="capacity" type="xsd:integer"/>
|
||||
</xsd:extension>
|
||||
</xsd:complexContent>
|
||||
</xsd:complexType>
|
||||
|
||||
<xsd:complexType name="channelType">
|
||||
<xsd:annotation>
|
||||
<xsd:documentation>
|
||||
Defines a message channel.
|
||||
</xsd:documentation>
|
||||
</xsd:annotation>
|
||||
<xsd:sequence>
|
||||
<xsd:element name="interceptors" type="channelInterceptorsType" minOccurs="0" maxOccurs="1"/>
|
||||
</xsd:sequence>
|
||||
<xsd:attribute name="id" type="xsd:ID" use="required"/>
|
||||
<xsd:attribute name="datatype" type="xsd:string"/>
|
||||
</xsd:complexType>
|
||||
|
||||
@@ -17,6 +17,7 @@
|
||||
package org.springframework.integration.dispatcher;
|
||||
|
||||
import org.springframework.integration.message.Message;
|
||||
import org.springframework.integration.message.MessageDeliveryException;
|
||||
import org.springframework.integration.message.MessageRejectedException;
|
||||
import org.springframework.integration.message.MessageTarget;
|
||||
|
||||
@@ -25,9 +26,10 @@ import org.springframework.integration.message.MessageTarget;
|
||||
* to send a {@link Message} to one of its targets. As soon as <em>one</em>
|
||||
* of the targets accepts the Message, the dispatcher will return 'true'.
|
||||
* <p>
|
||||
* If all targets reject the Message, the dispatcher will throw a
|
||||
* MessageRejectedException. If all targets return 'false' (e.g. due
|
||||
* to a timeout), the dispatcher will return 'false'.
|
||||
* If the dispatcher has no targets, a {@link MessageDeliveryException}
|
||||
* will be thrown. If all targets reject the Message, the dispatcher will
|
||||
* throw a MessageRejectedException. If all targets return 'false'
|
||||
* (e.g. due to a timeout), the dispatcher will return 'false'.
|
||||
*
|
||||
* @author Mark Fisher
|
||||
*/
|
||||
@@ -35,10 +37,7 @@ public class SimpleDispatcher extends AbstractDispatcher {
|
||||
|
||||
public boolean send(Message<?> message) {
|
||||
if (this.targets.size() == 0) {
|
||||
if (logger.isWarnEnabled()) {
|
||||
logger.warn("Dispatcher has no targets.");
|
||||
}
|
||||
return false;
|
||||
throw new MessageDeliveryException(message, "Dispatcher has no targets.");
|
||||
}
|
||||
int count = 0;
|
||||
int rejectedExceptionCount = 0;
|
||||
|
||||
@@ -36,7 +36,7 @@ public interface MessageTarget {
|
||||
* @throws MessageRejectedException if this particular Message is not accepted by the target
|
||||
* (e.g. after consulting a {@link org.springframework.integration.message.selector.MessageSelector})
|
||||
* @throws MessageDeliveryException if this target is unable to send the Message due
|
||||
* to a transport error
|
||||
* to a transport error.
|
||||
*/
|
||||
boolean send(Message<?> message) throws MessageRejectedException, MessageDeliveryException;
|
||||
|
||||
|
||||
Reference in New Issue
Block a user