Added namespace support for a 'bridge' element that simply connects any input-channel to any output-channel. This enables the conversion between PollableChannels and SubscribableChannels and also provides throttling capabilities based on the polling interval (or cron-expression) and 'max-messages-per-poll' settings (INT-193).

2008-11-22 22:03:36 +00:00
parent 25766bc7be
commit 55c62a06e6
6 changed files with 178 additions and 8 deletions
--- a/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/BridgeParser.java
+++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/BridgeParser.java
@@ -0,0 +1,37 @@
+/*
+ * 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.xml;
+
+import org.w3c.dom.Element;
+
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.handler.BridgeHandler;
+
+/**
+ * Parser for the &lt;bridge&gt; element.
+ * 
+ * @author Mark Fisher
+ */
+public class BridgeParser extends AbstractConsumerEndpointParser {
+
+	@Override
+	protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) {
+		return BeanDefinitionBuilder.genericBeanDefinition(BridgeHandler.class);
+	}
+
+}
--- a/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceHandler.java
+++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceHandler.java
@@ -109,6 +109,7 @@ public class IntegrationNamespaceHandler implements NamespaceHandler {
 			registerBeanDefinitionParser("inbound-channel-adapter", new MethodInvokingInboundChannelAdapterParser());
 			registerBeanDefinitionParser("outbound-channel-adapter", new MethodInvokingOutboundChannelAdapterParser());
 			registerBeanDefinitionParser("gateway", new GatewayParser());
+			registerBeanDefinitionParser("bridge", new BridgeParser());
 			registerBeanDefinitionParser("chain", new ChainParser());
 			registerBeanDefinitionParser("selector-chain", new SelectorChainParser());
 			registerBeanDefinitionParser("annotation-config", new AnnotationConfigParser());
--- a/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/spring-integration-1.0.xsd
+++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/spring-integration-1.0.xsd
@@ -173,7 +173,7 @@
 		<xsd:attribute name="auto-startup" type="xsd:string" default="true"/>
 	</xsd:complexType>

-	<xsd:element name="service-activator" type="inputOutputEndpointType">
+	<xsd:element name="service-activator" type="handlerEndpointType">
 		<xsd:annotation>
 			<xsd:documentation>
 	Defines an endpoint for exposing any bean reference as a service that receives
@@ -185,6 +185,20 @@
 		</xsd:annotation>
 	</xsd:element>

+	<xsd:complexType name="handlerEndpointType">
+		<xsd:annotation>
+			<xsd:documentation>
+	Base type for Message-handling endpoints.
+			</xsd:documentation>
+		</xsd:annotation>
+		<xsd:complexContent>
+			<xsd:extension base="inputOutputEndpointType">
+				<xsd:attribute name="ref" type="xsd:string"/>
+				<xsd:attribute name="method" type="xsd:string"/>
+			</xsd:extension>
+		</xsd:complexContent>
+	</xsd:complexType>
+
 	<xsd:complexType name="inputOutputEndpointType">
 		<xsd:annotation>
 			<xsd:documentation>
@@ -214,7 +228,7 @@
 			</xsd:documentation>
 		</xsd:annotation>
 		<xsd:complexContent>
-			<xsd:extension base="handlerType">
+			<xsd:extension base="beans:identifiedType">
 				<xsd:sequence>
 					<xsd:element ref="poller" minOccurs="0" maxOccurs="1"/>
 				</xsd:sequence>
@@ -246,6 +260,14 @@
 		</xsd:complexContent>
 	</xsd:complexType>

+	<xsd:element name="bridge" type="inputOutputEndpointType">
+		<xsd:annotation>
+			<xsd:documentation>
+	Defines an endpoint that passes a Message to the output-channel without modifying it.
+			</xsd:documentation>
+		</xsd:annotation>
+	</xsd:element>
+
 	<xsd:element name="chain">
 		<xsd:annotation>
 			<xsd:documentation>
@@ -475,7 +497,7 @@
 		</xsd:complexType>
 	</xsd:element>

-	<xsd:element name="transformer" type="inputOutputEndpointType">
+	<xsd:element name="transformer" type="handlerEndpointType">
 		<xsd:annotation>
 			<xsd:documentation>
 	Defines a Transformer.
@@ -491,7 +513,7 @@
 		</xsd:annotation>
 		<xsd:complexType>
 			<xsd:complexContent>
-				<xsd:extension base="inputOutputEndpointType">
+				<xsd:extension base="handlerEndpointType">
 					<xsd:attribute name="throw-exception-on-rejection" type="xsd:string" default="false"/>
 				</xsd:extension>
 			</xsd:complexContent>
@@ -508,12 +530,14 @@
 			<xsd:complexContent>
 				<xsd:extension base="inputEndpointType">
 					<xsd:attribute name="default-output-channel" type="xsd:string"/>
+					<xsd:attribute name="ref" type="xsd:string"/>
+					<xsd:attribute name="method" type="xsd:string"/>
 				</xsd:extension>
 			</xsd:complexContent>
 		</xsd:complexType>
 	</xsd:element>

-	<xsd:element name="splitter" type="inputOutputEndpointType">
+	<xsd:element name="splitter" type="handlerEndpointType">
 		<xsd:annotation>
 			<xsd:documentation>
 	Defines a Splitter.
@@ -529,7 +553,7 @@
 				</xsd:documentation>
 			</xsd:annotation>
 			<xsd:complexContent>
-				<xsd:extension base="inputOutputEndpointType">
+				<xsd:extension base="handlerEndpointType">
 					<xsd:attribute name="completion-strategy" type="xsd:string"/>
 					<xsd:attribute name="completion-strategy-method" type="xsd:string"/>
 					<xsd:attribute name="discard-channel" type="xsd:string"/>
@@ -551,7 +575,7 @@
 				</xsd:documentation>
 			</xsd:annotation>
 			<xsd:complexContent>
-				<xsd:extension base="inputOutputEndpointType">
+				<xsd:extension base="handlerEndpointType">
 					<xsd:attribute name="discard-channel" type="xsd:string"/>
 					<xsd:attribute name="send-timeout" type="xsd:string"/>
 					<xsd:attribute name="release-partial-sequences" type="xsd:string"/>
--- a/org.springframework.integration/src/main/java/org/springframework/integration/handler/MessagingBridge.java
+++ b/org.springframework.integration/src/main/java/org/springframework/integration/handler/MessagingBridge.java
@@ -19,9 +19,14 @@ package org.springframework.integration.handler;
 import org.springframework.integration.core.Message;

 /**
+ * A simple MessageHandler implementation that passes the request Message
+ * directly to the output channel without modifying it. The main purpose of
+ * this handler is to bridge a PollableChannel to a SubscribableChannel or
+ * vice-versa.
+ * 
 * @author Mark Fisher
 */
-public class MessagingBridge extends AbstractReplyProducingMessageHandler {
+public class BridgeHandler extends AbstractReplyProducingMessageHandler {

 	@Override
 	protected void handleRequestMessage(Message<?> requestMessage, ReplyMessageHolder replyMessageHolder) {
--- a/org.springframework.integration/src/test/java/org/springframework/integration/config/xml/BridgeParserTests-context.xml
+++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/xml/BridgeParserTests-context.xml
@@ -0,0 +1,32 @@
+<?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-2.5.xsd
+			http://www.springframework.org/schema/integration
+			http://www.springframework.org/schema/integration/spring-integration-1.0.xsd">
+
+	<channel id="pollableChannel">
+		<queue/>
+	</channel>
+
+	<channel id="output1">
+		<queue/>
+	</channel>
+
+	<channel id="subscribableChannel"/>
+
+	<channel id="output2">
+		<queue/>
+	</channel>
+
+	<bridge input-channel="pollableChannel" output-channel="output1">
+		<poller max-messages-per-poll="2">
+			<interval-trigger interval="3000"/>
+		</poller>
+	</bridge>
+
+	<bridge input-channel="subscribableChannel" output-channel="output2"/>
+
+</beans:beans>
--- a/org.springframework.integration/src/test/java/org/springframework/integration/config/xml/BridgeParserTests.java
+++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/xml/BridgeParserTests.java
@@ -0,0 +1,71 @@
+/*
+ * 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.xml;
+
+import static org.junit.Assert.assertEquals;
+
+import org.junit.Test;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.integration.channel.PollableChannel;
+import org.springframework.integration.core.Message;
+import org.springframework.integration.core.MessageChannel;
+import org.springframework.integration.message.StringMessage;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.AbstractJUnit4SpringContextTests;
+
+/**
+ * @author Mark Fisher
+ */
+@ContextConfiguration
+public class BridgeParserTests extends AbstractJUnit4SpringContextTests {
+
+	@Autowired
+	@Qualifier("pollableChannel")
+	private PollableChannel pollableChannel;
+
+	@Autowired
+	@Qualifier("subscribableChannel")
+	private MessageChannel subscribableChannel;
+
+	@Autowired
+	@Qualifier("output1")
+	private PollableChannel output1;
+
+	@Autowired
+	@Qualifier("output2")
+	private PollableChannel output2;
+
+
+	@Test
+	public void pollableChannel() {
+		Message<?> message = new StringMessage("test1");
+		this.pollableChannel.send(message);
+		Message<?> reply = this.output1.receive(1000);
+		assertEquals(message, reply);
+	}
+
+	@Test
+	public void subscribableChannel() {
+		Message<?> message = new StringMessage("test2");
+		this.subscribableChannel.send(message);
+		Message<?> reply = this.output2.receive(0);
+		assertEquals(message, reply);
+	}
+
+}