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

INT-911, Created MessageHandling infrastructure around Gateway so it could be included in the <chain>

Browse Source
This commit is contained in:
Oleg Zhurakousky
2009-12-24 10:42:55 +00:00
parent 34d7f0c63c
commit 572b7c0656
7 changed files with 227 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
org.springframework.integration/src/main/java/org/springframework/integration/config/xml/ChainParser.java
View File

@@ -27,12 +27,14 @@ import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.support.ManagedList;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.gateway.GatewayInvokingMessageHandler;
/**
* Parser for the &lt;chain&gt; element.
*
* @author Mark Fisher
* @author Iwein Fuld
* @author Oleg Zhurakousky
*/
public class ChainParser extends AbstractConsumerEndpointParser {
@@ -47,7 +49,14 @@ public class ChainParser extends AbstractConsumerEndpointParser {
Node child = children.item(i);
if (child.getNodeType() == Node.ELEMENT_NODE && !"poller".equals(child.getLocalName())) {
String childBeanName = this.parseChild((Element) child, parserContext, builder.getBeanDefinition());
handlerList.add(new RuntimeBeanReference(childBeanName));
// INT-911 will create Gateway invoking MessageHandler, allowing 'gateway' to be included in the chain
if ("gateway".equals(child.getLocalName())){
BeanDefinitionBuilder gwBuilder = BeanDefinitionBuilder.genericBeanDefinition(GatewayInvokingMessageHandler.class);
gwBuilder.addConstructorArgValue(new RuntimeBeanReference(childBeanName));
handlerList.add(gwBuilder.getBeanDefinition());
} else {
handlerList.add(new RuntimeBeanReference(childBeanName));
}
}
}
builder.addPropertyValue("handlers", handlerList);

47
org.springframework.integration/src/main/java/org/springframework/integration/gateway/GatewayInvokingMessageHandler.java Normal file
View File

@@ -0,0 +1,47 @@
/*
* 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.gateway;
import org.springframework.integration.core.Message;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.message.MessageHandler;
import org.springframework.util.Assert;
/**
* Will decorate 'gateway' as {@link MessageHandler} so it could be included in the chain.
*
* @author Oleg Zhurakousky
* @since 2.0
*/
public class GatewayInvokingMessageHandler extends
AbstractReplyProducingMessageHandler {
private GenericSendAndRecieveGateway gateway;
/**
*
* @param gateway
*/
public GatewayInvokingMessageHandler(GenericSendAndRecieveGateway gateway){
Assert.notNull(gateway, "gateway must not be null");
this.gateway = gateway;
}
/**
* Will simply delegate to the original gateway
*/
protected Object handleRequestMessage(Message<?> requestMessage) {
return gateway.sendAndRecieve(requestMessage);
}
}

10
org.springframework.integration/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java
View File

@@ -24,7 +24,6 @@ import java.util.Map;
import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.SimpleTypeConverter;
@@ -74,7 +73,14 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint implements Factory
private Map<String, GatewayMethodDefinition> methodToChannelMap;
/**
* Will initialize this Factory with he default instance of the 'gateway' interface
* {@link GenericSendAndRecieveGateway} which will be used by this proxy if
* 'service-interface' attribute is not set.
*/
public GatewayProxyFactoryBean(){
this.serviceInterface = GenericSendAndRecieveGateway.class;
}
/**
* Set the interface class that the generated proxy should implement.
* If none is provided explicitly, the default is MessageHandler.

30
org.springframework.integration/src/main/java/org/springframework/integration/gateway/GenericSendAndRecieveGateway.java Normal file
View File

@@ -0,0 +1,30 @@
/*
* 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.gateway;
import org.springframework.integration.core.Message;
/**
* Generic definition of the 'gateway' which will be used by {@link GatewayProxyFactoryBean}
* if 'service-interface' property is not provided
*
* @author Oleg Zhurakousky
* @since 2.0
*/
interface GenericSendAndRecieveGateway {
public Message<?> sendAndRecieve(Message<?> message);
}

5
org.springframework.integration/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd
View File

@@ -315,8 +315,8 @@
</xsd:complexType>
</xsd:element>
</xsd:sequence>
<xsd:attribute name="id" type="xsd:ID" use="required" />
<xsd:attribute name="service-interface" type="xsd:string" use="required">
<xsd:attribute name="id" type="xsd:ID" use="optional" />
<xsd:attribute name="service-interface" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="direct">
@@ -706,6 +706,7 @@
<xsd:element ref="router" />
<xsd:element ref="delayer" />
<xsd:element ref="chain" />
<xsd:element ref="gateway" />
<xsd:element name="poller" type="innerPollerType" />
</xsd:choice>
</xsd:extension>

38
org.springframework.integration/src/test/java/org/springframework/integration/gateway/GatewayInvokingMessaheHandlerTests-context.xml Normal file
View File

@@ -0,0 +1,38 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.0.xsd"
xmlns:si="http://www.springframework.org/schema/integration">
<si:gateway id="simpleGateway"
default-request-channel="inputA"
default-reply-channel="inputB"
service-interface="org.springframework.integration.gateway.GatewayInvokingMessaheHandlerTests$SimpleGateway"/>
<si:publish-subscribe-channel id="inputA" />
<si:publish-subscribe-channel id="inputB" />
<si:publish-subscribe-channel id="inputC" />
<si:chain input-channel="inputA" output-channel="inputB">
<si:header-enricher>
<si:header name="foo" value="foo" />
</si:header-enricher>
<si:service-activator>
<bean class="org.springframework.integration.gateway.GatewayInvokingMessaheHandlerTests$SimpleService" />
</si:service-activator>
<si:gateway default-request-channel="inputC"/>
</si:chain>
<si:chain input-channel="inputC">
<si:header-enricher>
<si:header name="name" value="oleg" />
</si:header-enricher>
<si:service-activator>
<bean class="org.springframework.integration.gateway.GatewayInvokingMessaheHandlerTests$SimpleService" />
</si:service-activator>
</si:chain>
</beans>

91
org.springframework.integration/src/test/java/org/springframework/integration/gateway/GatewayInvokingMessaheHandlerTests.java Normal file
View File

@@ -0,0 +1,91 @@
/*
* 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.gateway;
import junit.framework.Assert;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.integration.channel.SubscribableChannel;
import org.springframework.integration.core.Message;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.MessageHandler;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.integration.message.MessageRejectedException;
import org.springframework.integration.message.StringMessage;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* TODO - insert COMMENT
* @author Oleg Zhurakousky
* @since 2.0
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
public class GatewayInvokingMessaheHandlerTests {
@Autowired
@Qualifier("inputA")
SubscribableChannel channel;
@Autowired
@Qualifier("simpleGateway")
SimpleGateway gateway;
@Autowired
@Qualifier("inputB")
SubscribableChannel output;
@Test
public void validateGatewayInTheChainViaChannel(){
output.subscribe(new MessageHandler() {
public void handleMessage(Message<?> message)
throws MessageRejectedException, MessageHandlingException,
MessageDeliveryException {
Assert.assertEquals("echo:echo:hello", message.getPayload());
Assert.assertEquals("foo", message.getHeaders().get("foo"));
Assert.assertEquals("oleg", message.getHeaders().get("name"));
}
});
channel.send(new StringMessage("hello"));
}
@Test
public void validateGatewayInTheChainViaAnotherGateway(){
output.subscribe(new MessageHandler() {
public void handleMessage(Message<?> message)
throws MessageRejectedException, MessageHandlingException,
MessageDeliveryException {
Assert.assertEquals("echo:echo:hello", message.getPayload());
Assert.assertEquals("foo", message.getHeaders().get("foo"));
Assert.assertEquals("oleg", message.getHeaders().get("name"));
}
});
String result = gateway.sendRecieve("hello");
Assert.assertEquals("echo:echo:hello", result);
}
public static interface SimpleGateway{
public String sendRecieve(String str);
}
public static class SimpleService {
public String echo(String str){
return "echo:" + str;
}
}
}