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

Merge pull request #14 from dturanski/INT-2049

Browse Source 
remove gemfire cfg files
  add namespace support
  add SpEL support to CacheWritingMessageHandler and general cleanup
This commit is contained in:
Mark Fisher
2011-08-16 16:43:51 -04:00
parent 0bba8827ca 3d9dc732f1
commit 254f0f1f50
31 changed files with 1042 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
build.gradle
View File

@@ -126,7 +126,7 @@ configure(javaprojects) {
springDataMongoVersion = '1.0.0.BUILD-SNAPSHOT'
springDataCommonsVersion = '1.2.0.BUILD-SNAPSHOT'
springDataRedisVersion = '1.0.0.BUILD-SNAPSHOT'
springGemfireVersion = '1.0.1.RELEASE'
springGemfireVersion = '1.1.0.BUILD-SNAPSHOT'
springSecurityVersion = '3.0.5.RELEASE'
springWsVersion = '2.0.2.RELEASE'
@@ -249,6 +249,7 @@ project('spring-integration-gemfire') {
compile "org.springframework:spring-context:$springVersion"
compile "org.springframework.data.gemfire:spring-gemfire:$springGemfireVersion"
testCompile project(":spring-integration-stream")
testCompile project(":spring-integration-test")
}
repositories {
mavenRepo urls: 'http://dist.gemstone.com/maven/release' // for gemfire
@@ -444,6 +445,7 @@ project('spring-integration-test') {
compile "junit:junit-dep:$junitVersion"
compile "org.mockito:mockito-all:$mockitoVersion"
compile "org.springframework:spring-context:$springVersion"
compile "org.springframework:spring-test:$springVersion"
}
}

15
spring-integration-gemfire/.springBeans Normal file
View File

@@ -0,0 +1,15 @@
<?xml version="1.0" encoding="UTF-8"?>
<beansProjectDescription>
<version>1</version>
<pluginVersion><![CDATA[2.7.1.201107082359-RELEASE]]></pluginVersion>
<configSuffixes>
<configSuffix><![CDATA[xml]]></configSuffix>
</configSuffixes>
<enableImports><![CDATA[false]]></enableImports>
<configs>
<config>src/test/java/org/springframework/integration/gemfire/inbound/GemfireInboundChannelAdapterTests-context.xml</config>
<config>src/test/java/org/springframework/integration/gemfire/outbound/GemfireOutboundChannelAdapterTests-context.xml</config>
</configs>
<configSets>
</configSets>
</beansProjectDescription>

68
spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/config/xml/GemfireInboundChannelAdapterParser.java Normal file
View File

@@ -0,0 +1,68 @@
/*
* Copyright 2002-2011 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.gemfire.config.xml;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.AbstractChannelAdapterParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.util.StringUtils;
import org.w3c.dom.Element;
/**
* @author David Turanski
* @since 2.1
*
*/
public class GemfireInboundChannelAdapterParser extends AbstractChannelAdapterParser {
private static final String ERROR_CHANNEL_ATTRIBUTE = "error-channel";
private static final String OUTPUT_CHANNEL_PROPERTY = "outputChannel";
private static final String REGION_ATTRIBUTE = "region";
private static final String PAYLOAD_EXPRESSION_PROPERTY = "payloadExpression";
private static final String EXPRESSION_ATTRIBUTE = "expression";
private static final String GEMFIRE_INBOUND_CACHE_LISTENING_MESSAGE_PRODUCER = "org.springframework.integration.gemfire.inbound.CacheListeningMessageProducer";
private static final String SUPPORTED_EVENT_TYPES_PROPERTY = "supportedEventTypes";
private static final String CACHE_EVENTS_ATTRIBUTE = "cache-events";
/* (non-Javadoc)
* @see org.springframework.integration.config.xml.AbstractChannelAdapterParser#doParse(org.w3c.dom.Element, org.springframework.beans.factory.xml.ParserContext, java.lang.String)
*/
@Override
protected AbstractBeanDefinition doParse(Element element, ParserContext parserContext, String channelName) {
BeanDefinitionBuilder listeningMessageProducer = BeanDefinitionBuilder.genericBeanDefinition(GEMFIRE_INBOUND_CACHE_LISTENING_MESSAGE_PRODUCER);
IntegrationNamespaceUtils.setValueIfAttributeDefined(listeningMessageProducer, element, EXPRESSION_ATTRIBUTE,PAYLOAD_EXPRESSION_PROPERTY);
IntegrationNamespaceUtils.setValueIfAttributeDefined(listeningMessageProducer, element, CACHE_EVENTS_ATTRIBUTE, SUPPORTED_EVENT_TYPES_PROPERTY);
if (!StringUtils.hasText(REGION_ATTRIBUTE)){
parserContext.getReaderContext().error("'region' attribute is required.",element);
}
listeningMessageProducer.addConstructorArgReference(element.getAttribute(REGION_ATTRIBUTE));
listeningMessageProducer.addPropertyReference(OUTPUT_CHANNEL_PROPERTY, channelName);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(listeningMessageProducer, element, ERROR_CHANNEL_ATTRIBUTE);
return listeningMessageProducer.getBeanDefinition();
}
}

32
spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/config/xml/GemfireIntegrationNamespaceHandler.java Normal file
View File

@@ -0,0 +1,32 @@
/*
* Copyright 2002-2011 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.gemfire.config.xml;
import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHandler;
/**
* @author David Turanski
*
*/
public class GemfireIntegrationNamespaceHandler extends AbstractIntegrationNamespaceHandler{
/* (non-Javadoc)
* @see org.springframework.beans.factory.xml.NamespaceHandler#init()
*/
public void init() {
registerBeanDefinitionParser("inbound-channel-adapter", new GemfireInboundChannelAdapterParser());
registerBeanDefinitionParser("outbound-channel-adapter", new GemfireOutboundChannelAdapterParser());
}
}

61
spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/config/xml/GemfireOutboundChannelAdapterParser.java Normal file
View File

@@ -0,0 +1,61 @@
/*
* Copyright 2002-2011 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.gemfire.config.xml;
import java.util.Map;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser;
import org.springframework.util.StringUtils;
import org.springframework.util.xml.DomUtils;
import org.w3c.dom.Element;
/**
* @author David Turanski
* @since 2.1
*
*/
public class GemfireOutboundChannelAdapterParser extends AbstractOutboundChannelAdapterParser {
private static final String CACHE_ENTRIES_PROPERTY = "cacheEntries";
private static final String CACHE_ENTRIES_ELEMENT = "cache-entries";
private static final String REGION_ATTRIBUTE = "region";
private static final String GEMFIRE_OUTBOUND_CACHE_WRITING_MESSAGE_HANDLER = "org.springframework.integration.gemfire.outbound.CacheWritingMessageHandler";
/* (non-Javadoc)
* @see org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser#parseConsumer(org.w3c.dom.Element, org.springframework.beans.factory.xml.ParserContext)
*/
@Override
protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) {
BeanDefinitionBuilder cacheWritingMessageHandler = BeanDefinitionBuilder.genericBeanDefinition(
GEMFIRE_OUTBOUND_CACHE_WRITING_MESSAGE_HANDLER);
if (!StringUtils.hasText(REGION_ATTRIBUTE)){
parserContext.getReaderContext().error("'region' attribute is required.",element);
}
cacheWritingMessageHandler.addConstructorArgReference(element.getAttribute(REGION_ATTRIBUTE));
Element cacheEntries = DomUtils.getChildElementByTagName(element,CACHE_ENTRIES_ELEMENT);
if (cacheEntries != null) {
Map<?,?> map = parserContext.getDelegate().parseMapElement(cacheEntries,cacheWritingMessageHandler.getBeanDefinition());
cacheWritingMessageHandler.addPropertyValue(CACHE_ENTRIES_PROPERTY, map);
}
return cacheWritingMessageHandler.getBeanDefinition();
}
}

18
spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/CacheListeningMessageProducer.java
View File

@@ -29,6 +29,7 @@ import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.util.Assert;
import com.gemstone.gemfire.cache.CacheClosedException;
import com.gemstone.gemfire.cache.CacheListener;
import com.gemstone.gemfire.cache.EntryEvent;
import com.gemstone.gemfire.cache.Region;
@@ -42,6 +43,7 @@ import com.gemstone.gemfire.cache.util.CacheListenerAdapter;
* payloadExpression is provided, the {@link EntryEvent} itself will be the payload.
*
* @author Mark Fisher
* @author David Turanski
* @since 2.1
*/
@SuppressWarnings({"rawtypes", "unchecked"})
@@ -60,11 +62,10 @@ public class CacheListeningMessageProducer extends MessageProducerSupport {
private final SpelExpressionParser parser = new SpelExpressionParser();
public CacheListeningMessageProducer(Region<?, ?> region) {
Assert.notNull(region, "region must not be null");
this.region = region;
this.listener = new MessageProducingCacheListener();
this.listener = new MessageProducingCacheListener();
}
@@ -95,9 +96,16 @@ public class CacheListeningMessageProducer extends MessageProducerSupport {
if (logger.isInfoEnabled()) {
logger.info("removing MessageProducingCacheListener from GemFire Region '" + this.region.getName() + "'");
}
this.region.getAttributesMutator().removeCacheListener(this.listener);
try {
this.region.getAttributesMutator().removeCacheListener(this.listener);
} catch (CacheClosedException e) {
if (logger.isDebugEnabled()){
logger.debug(e.getMessage(),e);
}
}
}
private class MessageProducingCacheListener extends CacheListenerAdapter {
@@ -143,5 +151,7 @@ public class CacheListeningMessageProducer extends MessageProducerSupport {
sendMessage(MessageBuilder.withPayload(payload).build());
}
}
}

63
spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/outbound/CacheWritingMessageHandler.java
View File

@@ -16,12 +16,18 @@
package org.springframework.integration.gemfire.outbound;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Map.Entry;
import org.springframework.data.gemfire.GemfireCallback;
import org.springframework.data.gemfire.GemfireTemplate;
import org.springframework.expression.Expression;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.integration.Message;
import org.springframework.integration.core.MessageHandler;
import org.springframework.integration.handler.AbstractMessageHandler;
import org.springframework.util.Assert;
import com.gemstone.gemfire.GemFireCheckedException;
@@ -29,30 +35,37 @@ import com.gemstone.gemfire.GemFireException;
import com.gemstone.gemfire.cache.Region;
/**
* A {@link MessageHandler} implementation that writes to a GemFire Region.
* The Message's payload must be an instance of java.util.Map.
* A {@link MessageHandler} implementation that writes to a GemFire Region. The
* Message's payload must be an instance of java.util.Map.
*
* @author Mark Fisher
* @author David Turanski
* @since 2.1
*/
public class CacheWritingMessageHandler implements MessageHandler {
public class CacheWritingMessageHandler extends AbstractMessageHandler {
private final Map<Expression, Expression> cacheEntryExpressions = new LinkedHashMap<Expression, Expression>();
private final GemfireTemplate gemfireTemplate = new GemfireTemplate();
@SuppressWarnings("rawtypes")
public CacheWritingMessageHandler(Region region) {
Assert.notNull(region, "region must not be null");
this.gemfireTemplate.setRegion(region);
this.gemfireTemplate.afterPropertiesSet();
this.gemfireTemplate.afterPropertiesSet();
}
public void handleMessage(Message<?> message) {
// TODO: add support for more options to get key/value (SpEL?)
@Override
public void handleMessageInternal(Message<?> message) {
Object payload = message.getPayload();
Assert.isTrue(payload instanceof Map, "only Map payloads are supported");
final Map<?, ?> map = (Map<?, ?>) payload;
Map<?, ?> cacheValues = (cacheEntryExpressions.size() > 0)?parseCacheEntries(message):null;
if (cacheValues == null) {
Assert.isTrue(payload instanceof Map, "If cache entry expressions are not configured, then payload must be a Map");
cacheValues = (Map<?, ?>) payload;
}
final Map<?, ?> map = cacheValues;
this.gemfireTemplate.execute(new GemfireCallback<Object>() {
@SuppressWarnings({ "rawtypes", "unchecked" })
public Object doInGemfire(Region region) throws GemFireCheckedException, GemFireException {
@@ -62,4 +75,32 @@ public class CacheWritingMessageHandler implements MessageHandler {
});
}
/**
* @param message
* @return
*/
private Map<Object, Object> parseCacheEntries(Message<?> message) {
if (cacheEntryExpressions.size() == 0) {
return null;
}
else {
Map<Object, Object> cacheValues = new HashMap<Object, Object>();
for (Entry<Expression, Expression> expressionEntry : cacheEntryExpressions.entrySet()) {
cacheValues.put(expressionEntry.getKey().getValue(message),expressionEntry.getValue().getValue(message));
}
return cacheValues;
}
}
public void setCacheEntries(Map<String, String> cacheEntries) {
if (cacheEntryExpressions.size() > 0) {
cacheEntryExpressions.clear();
}
for (Entry<String, String> cacheEntry : cacheEntries.entrySet()) {
this.cacheEntryExpressions.put(new SpelExpressionParser().parseExpression(cacheEntry.getKey()),
new SpelExpressionParser().parseExpression(cacheEntry.getValue()));
}
}
}

1
spring-integration-gemfire/src/main/resources/META-INF/spring.handlers Normal file
View File

@@ -0,0 +1 @@
http\://www.springframework.org/schema/integration/gemfire=org.springframework.integration.gemfire.config.xml.GemfireIntegrationNamespaceHandler

2
spring-integration-gemfire/src/main/resources/META-INF/spring.schemas Normal file
View File

@@ -0,0 +1,2 @@
http\://www.springframework.org/schema/integration/scripting/spring-integration-gemfire.xsd=org/springframework/integration/gemfire/config/xml/spring-integration-gemfire-2.1.xsd
http\://www.springframework.org/schema/integration/scripting/spring-integration-gemfire-2.1.xsd=org/springframework/integration/gemfire/config/xml/spring-integration-gemfire-2.1.xsd

4
spring-integration-gemfire/src/main/resources/META-INF/spring.tooling Normal file
View File

@@ -0,0 +1,4 @@
# Tooling related information for the integration gemfire namespace
http\://www.springframework.org/schema/integration/gemfire@name=integration gemfire Namespace
http\://www.springframework.org/schema/integration/gemfire@prefix=int-gfe
http\://www.springframework.org/schema/integration@icon=org/springframework/integration/gemfire/config/xml/spring-integration-gemfire.gif

140
spring-integration-gemfire/src/main/resources/org/springframework/integration/gemfire/config/xml/spring-integration-gemfire-2.1.xsd Normal file
View File

@@ -0,0 +1,140 @@
<?xml version="1.0" encoding="UTF-8"?>
<xsd:schema xmlns="http://www.springframework.org/schema/integration/gemfire"
xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:tool="http://www.springframework.org/schema/tool"
xmlns:integration="http://www.springframework.org/schema/integration"
targetNamespace="http://www.springframework.org/schema/integration/gemfire"
elementFormDefault="qualified" attributeFormDefault="unqualified">
<xsd:import namespace="http://www.springframework.org/schema/beans"
schemaLocation="http://www.springframework.org/schema/beans/spring-beans-3.1.xsd"/>
<xsd:import namespace="http://www.springframework.org/schema/tool" />
<xsd:import namespace="http://www.springframework.org/schema/integration"
schemaLocation="http://www.springframework.org/schema/integration/spring-integration-2.1.xsd" />
<xsd:annotation>
<xsd:documentation><![CDATA[
Defines the core configuration elements for Spring Integration Gemfire Support.
]]></xsd:documentation>
</xsd:annotation>
<xsd:element name="inbound-channel-adapter">
<xsd:annotation>
<xsd:documentation>
Configures an inbound Channel Adapter backed by a
Gemfire CacheListener
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:attribute name="id" type="xsd:string" use="optional" />
<xsd:attribute name="channel" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type
type="org.springframework.integration.core.MessageChannel" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="error-channel" type="xsd:string"
use="optional">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type
type="org.springframework.integration.core.MessageChannel" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="region" type="xsd:string" use="required">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="com.gemstone.gemfire.cache.Region" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="expression" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
Expression to be evaluated to produce a value for the payload.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="cache-events" type="xsd:string"
use="optional" default="CREATED,UPDATED">
<xsd:annotation>
<xsd:documentation><![CDATA[
Enabled cache event types
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="value">
<tool:expected-type
type="org.springframework.integration.gemfire.inbound.EventType" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>
<xsd:element name="outbound-channel-adapter">
<xsd:complexType>
<xsd:annotation>
<xsd:documentation>
Configures an outbound Channel Adapter that writes Message payloads to a
File.
</xsd:documentation>
</xsd:annotation>
<xsd:sequence>
<xsd:element name="cache-entries" type="beans:mapType" minOccurs="0" maxOccurs="1">
<xsd:annotation>
<xsd:documentation>
A map of SpEL expressions used to create cache entries. If not provided, payload must be a Map
</xsd:documentation>
</xsd:annotation>
</xsd:element>
</xsd:sequence>
<xsd:attribute name="id" type="xsd:string" use="optional" />
<xsd:attribute name="channel" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type
type="org.springframework.integration.core.MessageChannel" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="region" type="xsd:string" use="required">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="com.gemstone.gemfire.cache.Region" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="order" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies the order for invocation when this endpoint is connected as a
subscriber to a SubscribableChannel.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>
</xsd:schema>

BIN
spring-integration-gemfire/src/main/resources/org/springframework/integration/gemfire/config/xml/spring-integration-gemfire.gif Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 539 B

1
spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/CacheListeningMessageProducerTests.java
View File

@@ -137,5 +137,4 @@ public class CacheListeningMessageProducerTests {
assertNotNull(message2);
assertEquals("foo was abc", message2.getPayload());
}
}

25
spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/GemfireInboundChannelAdapterTests-context.xml Normal file
View File

@@ -0,0 +1,25 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:gfe="http://www.springframework.org/schema/gemfire"
xmlns:int-gfe="http://www.springframework.org/schema/integration/gemfire"
xmlns:int="http://www.springframework.org/schema/integration"
xsi:schemaLocation="http://www.springframework.org/schema/integration/gemfire http://www.springframework.org/schema/integration/scripting/spring-integration-gemfire.xsd
http://www.springframework.org/schema/gemfire http://www.springframework.org/schema/gemfire/spring-gemfire.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<gfe:cache id="gemfire-cache-2"/>
<gfe:replicated-region id="region1" cache-ref="gemfire-cache-2"/>
<gfe:replicated-region id="region2" cache-ref="gemfire-cache-2"/>
<gfe:replicated-region id="region3" cache-ref="gemfire-cache-2"/>
<int-gfe:inbound-channel-adapter id="channel1" region="region1" cache-events="CREATED" expression="newValue"/>
<int-gfe:inbound-channel-adapter id="channel2" region="region2"/>
<int-gfe:inbound-channel-adapter id="channel3" region="region3" error-channel="errorChannel"/>
</beans>

118
spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/GemfireInboundChannelAdapterTests.java Normal file
View File

@@ -0,0 +1,118 @@
/*
* Copyright 2002-2011 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.gemfire.inbound;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.Message;
import org.springframework.integration.MessagingException;
import org.springframework.integration.core.MessageHandler;
import org.springframework.integration.core.SubscribableChannel;
import org.springframework.integration.message.ErrorMessage;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import com.gemstone.gemfire.cache.EntryEvent;
import com.gemstone.gemfire.internal.cache.DistributedRegion;
/**
* @author David Turanski
* @since 2.1
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
public class GemfireInboundChannelAdapterTests {
@Autowired
SubscribableChannel channel1;
@Autowired
SubscribableChannel channel2;
@Autowired
SubscribableChannel channel3;
@Autowired
SubscribableChannel errorChannel;
@Autowired
DistributedRegion region1;
@Autowired
DistributedRegion region2;
@Autowired
DistributedRegion region3;
@Test
public void testGemfireInboundChannelAdapterWithExpression() {
EventHandler eventHandler1 = new EventHandler();
channel1.subscribe(eventHandler1);
region1.put("payload", "payload");
assertEquals("payload", eventHandler1.event);
}
@Test
public void testGemfireInboundChannelAdapterDefault() {
EventHandler eventHandler2 = new EventHandler();
channel2.subscribe(eventHandler2);
region2.put("payload", "payload");
assertTrue(eventHandler2.event instanceof EntryEvent);
EntryEvent<?,?> event = (EntryEvent<?,?>)eventHandler2.event;
assertEquals("payload", event.getNewValue());
}
@Test
public void testErrorChannel() {
channel3.subscribe(new MessageHandler() {
public void handleMessage(Message<?> message) throws MessagingException {
throw new MessagingException("got an error");
}
});
ErrorHandler errorHandler = new ErrorHandler();
errorChannel.subscribe(errorHandler);
region3.put("payload", "payload");
assertEquals(1, errorHandler.count);
}
static class ErrorHandler implements MessageHandler {
public int count = 0;
public void handleMessage(Message<?> message) throws MessagingException {
assertTrue(message instanceof ErrorMessage);
count++;
}
}
static class EventHandler implements MessageHandler {
public Object event = null;
public void handleMessage(Message<?> message) throws MessagingException {
event = message.getPayload();
}
}
}

0
spring-integration-gemfire/src/test/resources/org/springframework/integration/gemfire/inbound/cq/CqClient-context.xml → spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/CqClient-context.xml
View File

0
spring-integration-gemfire/src/test/resources/org/springframework/integration/gemfire/inbound/cq/CqServer-context.xml → spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/CqServer-context.xml
View File

0
spring-integration-gemfire/src/test/resources/org/springframework/integration/gemfire/inbound/cq/common.properties → spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/common.properties
View File

0
spring-integration-gemfire/src/test/resources/org/springframework/integration/gemfire/inbound/cq/gfe-cache.properties → spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/gfe-cache.properties
View File

28
spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/outbound/CacheWritingMessageHandlerTests.java
View File

@@ -28,11 +28,13 @@ import org.springframework.data.gemfire.RegionFactoryBean;
import org.springframework.integration.Message;
import org.springframework.integration.support.MessageBuilder;
import com.gemstone.bp.edu.emory.mathcs.backport.java.util.Collections;
import com.gemstone.gemfire.cache.Cache;
import com.gemstone.gemfire.cache.Region;
/**
* @author Mark Fisher
* @author David Turanski
* @since 2.1
*/
public class CacheWritingMessageHandlerTests {
@@ -56,5 +58,31 @@ public class CacheWritingMessageHandlerTests {
assertEquals(1, region.size());
assertEquals("bar", region.get("foo"));
}
@Test
public void ExpressionsWriteToCache() throws Exception {
CacheFactoryBean cacheFactoryBean = new CacheFactoryBean();
cacheFactoryBean.afterPropertiesSet();
Cache cache = cacheFactoryBean.getObject();
RegionFactoryBean<String, String> regionFactoryBean = new RegionFactoryBean<String, String>();
regionFactoryBean.setName("test.expressionsWriteToCache");
regionFactoryBean.setCache(cache);
regionFactoryBean.afterPropertiesSet();
Region<String, String> region = regionFactoryBean.getObject();
assertEquals(0, region.size());
CacheWritingMessageHandler handler = new CacheWritingMessageHandler(region);
Map<String, String> expressions = new HashMap<String, String>();
expressions.put("'foo'", "'bar'");
expressions.put("payload.toUpperCase()", "headers['bar'].toUpperCase()");
handler.setCacheEntries(expressions);
@SuppressWarnings("unchecked")
Message<?> message = MessageBuilder.withPayload("foo").copyHeaders(Collections.singletonMap("bar", "bar")).build();
handler.handleMessage(message);
assertEquals(2, region.size());
assertEquals("BAR", region.get("FOO"));
assertEquals("bar", region.get("foo"));
}
}

26
spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/outbound/GemfireOutboundChannelAdapterTests-context.xml Normal file
View File

@@ -0,0 +1,26 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int-gfe="http://www.springframework.org/schema/integration/gemfire"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:gfe="http://www.springframework.org/schema/gemfire"
xsi:schemaLocation="http://www.springframework.org/schema/integration/gemfire http://www.springframework.org/schema/integration/scripting/spring-integration-gemfire.xsd
http://www.springframework.org/schema/gemfire http://www.springframework.org/schema/gemfire/spring-gemfire-1.1.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<gfe:cache />
<gfe:replicated-region id="region1"/>
<gfe:replicated-region id="region2"/>
<int-gfe:outbound-channel-adapter id="cacheChannel1" region="region1"/>
<int-gfe:outbound-channel-adapter id="cacheChannel2" region="region2" order="19">
<int-gfe:cache-entries>
<entry key="payload.toUpperCase()" value="payload.toLowerCase()"/>
<entry key="'foo'" value="'bar'"/>
</int-gfe:cache-entries>
</int-gfe:outbound-channel-adapter>
</beans>

78
spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/outbound/GemfireOutboundChannelAdapterTests.java Normal file
View File

@@ -0,0 +1,78 @@
/*
* Copyright 2002-2011 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.gemfire.outbound;
import static org.junit.Assert.assertEquals;
import java.util.HashMap;
import java.util.Map;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import com.gemstone.gemfire.internal.cache.DistributedRegion;
/**
* @author David Turanski
* @since 2.1
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
public class GemfireOutboundChannelAdapterTests {
@Autowired
MessageChannel cacheChannel1;
@Autowired
DistributedRegion region1;
@Autowired
MessageChannel cacheChannel2;
@Autowired
DistributedRegion region2;
@Before
public void setUp() {
region1.clear();
region2.clear();
}
@Test
public void testWriteMapPayload() {
Map<String,String> map = new HashMap<String,String>();
map.put("foo","bar");
Message<?> message = MessageBuilder.withPayload(map).build();
cacheChannel1.send(message);
assertEquals(1,region1.size());
assertEquals("bar",region1.get("foo"));
}
@Test
public void testWriteExpressions() {
Message<?> message = MessageBuilder.withPayload("Hello").build();
cacheChannel2.send(message);
assertEquals(2,region2.size());
assertEquals("hello",region2.get("HELLO"));
assertEquals("bar",region2.get("foo"));
}
}

0
spring-integration-gemfire/src/test/resources/org/springframework/integration/gemfire/store/messagegroupstore/GemfireMessageStore-context.xml → spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/GemfireMessageStore-context.xml
View File

28
spring-integration-gemfire/src/test/resources/log4j.xml Normal file
View File

@@ -0,0 +1,28 @@
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE log4j:configuration SYSTEM "log4j.dtd">
<log4j:configuration xmlns:log4j="http://jakarta.apache.org/log4j/">
<!-- Appenders -->
<appender name="console" class="org.apache.log4j.ConsoleAppender">
<param name="Target" value="System.out" />
<layout class="org.apache.log4j.PatternLayout">
<param name="ConversionPattern" value="[%t] %-5p: %c - %m%n" />
</layout>
</appender>
<!-- Loggers -->
<logger name="org.springframework">
<level value="warn" />
</logger>
<logger name="org.springframework.integration">
<level value="warn" />
</logger>
<!-- Root Logger -->
<root>
<priority value="info" />
<appender-ref ref="console" />
</root>
</log4j:configuration>

82
spring-integration-test/src/main/java/org/springframework/integration/test/support/AbstractRequestResponseScenarioTest.java Normal file
View File

@@ -0,0 +1,82 @@
package org.springframework.integration.test.support;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import java.util.List;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.core.PollableChannel;
import org.springframework.integration.core.SubscribableChannel;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* Convenience class for testing Spring Integration request-response message scenarios. Users
* create subclasses to execute on or more {@link RequestResponseScenario} tests. each scenario defines:
* <ul>
* <li>An inputChannelName</li>
* <li>An outputChannelName</li>
* <li>A payload or message to send as a request message on the inputChannel</li>
* <li>A handler to validate the response received on the outputChannel</li>
* </ul>
* @author David Turanski
*/
@RunWith(SpringJUnit4ClassRunner.class)
public abstract class AbstractRequestResponseScenarioTest {
private List<RequestResponseScenario> scenarios = null;
@Autowired
private ApplicationContext applicationContext;
@Before
public void setUp(){
scenarios = defineRequestResponseScenarios();
}
/**
* Execute each scenario. Instantiate the message channels, send the request message on the
* input channel and invoke the validator on the response received on the output channel.
* This can handle subscribable or pollable output channels.
*/
@Test
public void testRequestResponseScenarios(){
int i = 1;
for (RequestResponseScenario scenario: scenarios){
String name = scenario.getName() == null? "scenario-"+(i++) : scenario.getName();
scenario.init();
MessageChannel inputChannel = applicationContext.getBean(scenario.getInputChannelName(),MessageChannel.class);
MessageChannel outputChannel = applicationContext.getBean(scenario.getOutputChannelName(),MessageChannel.class);
if (outputChannel instanceof SubscribableChannel){
((SubscribableChannel) outputChannel).subscribe(scenario.getResponseValidator());
}
assertTrue(name + ": message not sent on " + scenario.getInputChannelName()
, inputChannel.send(scenario.getMessage()));
if (outputChannel instanceof PollableChannel){
Message<?> response = ((PollableChannel) outputChannel).receive(10000);
assertNotNull(name + ": receive timeout on " + scenario.getOutputChannelName(),response);
if (scenario.getResponseValidator() instanceof PayloadValidator){
scenario.getResponseValidator().validateResponse(response.getPayload());
} else {
scenario.getResponseValidator().validateResponse(response);
}
}
}
}
/**
* Implement this method to define RequestResponse scenarios
* @return - A List of {@link RequestResponseScenario}
*/
protected abstract List<RequestResponseScenario> defineRequestResponseScenarios();
}

27
spring-integration-test/src/main/java/org/springframework/integration/test/support/AbstractResponseValidator.java Normal file
View File

@@ -0,0 +1,27 @@
package org.springframework.integration.test.support;
import org.springframework.integration.Message;
import org.springframework.integration.MessagingException;
import org.springframework.integration.core.MessageHandler;
/**
* The base class for response validators used for {@link RequestResponseScenario}s
* @author David Turanski
*
*/
public abstract class AbstractResponseValidator implements MessageHandler {
public void handleMessage(Message<?> message) throws MessagingException {
validateResponse(extractPayload()? message.getPayload(): message );
}
/**
* Implement this method to validate the response (Message or Payload)
* @param response
*/
protected abstract void validateResponse(Object response);
/**
* If true will extract the payload as the parameter for validateResponse()
* @return true to extract the payload; false to process the message.
*/
protected abstract boolean extractPayload();
}

24
spring-integration-test/src/main/java/org/springframework/integration/test/support/MessageValidator.java Normal file
View File

@@ -0,0 +1,24 @@
package org.springframework.integration.test.support;
import org.springframework.integration.Message;
/**
* Validate a message. Create an anonymous instance or subclass to
* implement the validateMessage() method
* @author David Turanski
*
*/
public abstract class MessageValidator extends AbstractResponseValidator {
protected final boolean extractPayload(){
return false;
}
protected final void validateResponse(Object response){
validateMessage((Message<?>) response);
}
/**
* Implement this method to validate the message
* @param message
*/
protected abstract void validateMessage(Message<?> message);
}

12
spring-integration-test/src/main/java/org/springframework/integration/test/support/PayloadValidator.java Normal file
View File

@@ -0,0 +1,12 @@
package org.springframework.integration.test.support;
/**
* Validate a message payload. Create an anonymous instance or subclass this
* to validate a response payload.
* @author David Turanski
*
*/
public abstract class PayloadValidator extends AbstractResponseValidator {
protected final boolean extractPayload(){
return true;
}
}

124
spring-integration-test/src/main/java/org/springframework/integration/test/support/RequestResponseScenario.java Normal file
View File

@@ -0,0 +1,124 @@
package org.springframework.integration.test.support;
import org.springframework.integration.Message;
import org.springframework.integration.message.GenericMessage;
import org.springframework.util.Assert;
/**
* Defines a Spring Integration request response test scenario. All setter methods may
* be chained.
* @author David Turanski
*
*/
public class RequestResponseScenario {
private final String inputChannelName;
private final String outputChannelName;
private Object payload;
private Message<?> message;
private AbstractResponseValidator responseValidator;
private String name;
protected Message<? extends Object> getMessage(){
if (message == null){
return new GenericMessage<Object>(this.payload);
} else {
return message;
}
}
/**
* Create an instance
* @param inputChannelName the input channel name
* @param outputChannelName the output channel name
*/
public RequestResponseScenario(String inputChannelName, String outputChannelName){
this.inputChannelName = inputChannelName;
this.outputChannelName = outputChannelName;
}
/**
*
* @return the input channel name
*/
public String getInputChannelName() {
return inputChannelName;
}
/**
*
* @return the output channel name
*/
public String getOutputChannelName() {
return outputChannelName;
}
/**
*
* @return the request message payload
*/
public Object getPayload() {
return payload;
}
/**
* set the payload of the request message
* @param payload
* @return this
*/
public RequestResponseScenario setPayload(Object payload) {
this.payload = payload;
return this;
}
/**
*
* @return the scenario name
*/
public String getName() {
return name;
}
/**
* set the scenario name (optional)
* @param name the name
* @return this
*/
public RequestResponseScenario setName(String name) {
this.name = name;
return this;
}
/**
*
* @return the response validator
* @see AbstractResponseValidator
*/
public AbstractResponseValidator getResponseValidator(){
return responseValidator;
}
/**
* Set the response validator
* @see AbstractResponseValidator
* @param responseValidator
* @return
*/
public RequestResponseScenario setResponseValidator(AbstractResponseValidator responseValidator) {
this.responseValidator = responseValidator;
return this;
}
/**
* Set the request message (as an alternative to setPayload())
* @param message
* @return
*/
public RequestResponseScenario setMessage(Message<?> message) {
this.message = message;
return this;
}
protected void init(){
Assert.state(message == null || payload == null,"cannot set both message and payload");
}
}

17
spring-integration-test/src/test/java/org/springframework/integration/test/support/MessageScenariosTest-context.xml Normal file
View File

@@ -0,0 +1,17 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.0.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<int:transformer input-channel="inputChannel" output-channel="outputChannel" expression="payload.toUpperCase()"/>
<int:channel id="outputChannel"/>
<int:transformer input-channel="inputChannel2" output-channel="outputChannel2" expression="payload.toUpperCase()"/>
<int:channel id="outputChannel2">
<int:queue/>
</int:channel>
</beans>

61
spring-integration-test/src/test/java/org/springframework/integration/test/support/MessageScenariosTest.java Normal file
View File

@@ -0,0 +1,61 @@
package org.springframework.integration.test.support;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertThat;
import java.util.ArrayList;
import java.util.List;
import org.springframework.integration.Message;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.test.context.ContextConfiguration;
import static org.springframework.integration.test.matcher.PayloadMatcher.hasPayload;
import static org.springframework.integration.test.matcher.HeaderMatcher.hasHeader;
@ContextConfiguration
public class MessageScenariosTest extends AbstractRequestResponseScenarioTest {
@Override
protected List<RequestResponseScenario> defineRequestResponseScenarios() {
List<RequestResponseScenario> scenarios= new ArrayList<RequestResponseScenario>();
RequestResponseScenario scenario1 = new RequestResponseScenario(
"inputChannel","outputChannel")
.setPayload("hello")
.setResponseValidator(new PayloadValidator() {
@Override
protected void validateResponse(Object response) {
assertEquals("HELLO",response);
}
});
scenarios.add(scenario1);
RequestResponseScenario scenario2 = new RequestResponseScenario(
"inputChannel","outputChannel")
.setMessage(MessageBuilder.withPayload("hello").setHeader("foo", "bar").build())
.setResponseValidator(new MessageValidator() {
@Override
protected void validateMessage(Message<?> message) {
assertThat(message,hasPayload("HELLO"));
assertThat(message,hasHeader("foo","bar"));
}
});
scenarios.add(scenario2);
RequestResponseScenario scenario3 = new RequestResponseScenario(
"inputChannel2","outputChannel2")
.setMessage(MessageBuilder.withPayload("hello").setHeader("foo", "bar").build())
.setResponseValidator(new MessageValidator() {
@Override
protected void validateMessage(Message<?> message) {
assertThat(message,hasPayload("HELLO"));
assertThat(message,hasHeader("foo","bar"));
}
});
scenarios.add(scenario3);
return scenarios;
}
}