diff --git a/build.gradle b/build.gradle index 48bd58667d..2df0dfdc99 100644 --- a/build.gradle +++ b/build.gradle @@ -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" } } diff --git a/spring-integration-gemfire/.springBeans b/spring-integration-gemfire/.springBeans new file mode 100644 index 0000000000..8a3d5fe7e4 --- /dev/null +++ b/spring-integration-gemfire/.springBeans @@ -0,0 +1,15 @@ + + + 1 + + + + + + + src/test/java/org/springframework/integration/gemfire/inbound/GemfireInboundChannelAdapterTests-context.xml + src/test/java/org/springframework/integration/gemfire/outbound/GemfireOutboundChannelAdapterTests-context.xml + + + + diff --git a/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/config/xml/GemfireInboundChannelAdapterParser.java b/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/config/xml/GemfireInboundChannelAdapterParser.java new file mode 100644 index 0000000000..baa605a095 --- /dev/null +++ b/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/config/xml/GemfireInboundChannelAdapterParser.java @@ -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(); + } + +} diff --git a/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/config/xml/GemfireIntegrationNamespaceHandler.java b/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/config/xml/GemfireIntegrationNamespaceHandler.java new file mode 100644 index 0000000000..a686001814 --- /dev/null +++ b/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/config/xml/GemfireIntegrationNamespaceHandler.java @@ -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()); + + } + +} diff --git a/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/config/xml/GemfireOutboundChannelAdapterParser.java b/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/config/xml/GemfireOutboundChannelAdapterParser.java new file mode 100644 index 0000000000..7c86a5a8f8 --- /dev/null +++ b/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/config/xml/GemfireOutboundChannelAdapterParser.java @@ -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(); + } +} diff --git a/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/CacheListeningMessageProducer.java b/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/CacheListeningMessageProducer.java index 8890082244..8d7296e8c0 100644 --- a/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/CacheListeningMessageProducer.java +++ b/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/CacheListeningMessageProducer.java @@ -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()); } } + + } diff --git a/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/outbound/CacheWritingMessageHandler.java b/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/outbound/CacheWritingMessageHandler.java index cde54c777c..bda7c71fdd 100644 --- a/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/outbound/CacheWritingMessageHandler.java +++ b/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/outbound/CacheWritingMessageHandler.java @@ -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 cacheEntryExpressions = new LinkedHashMap(); 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() { @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 parseCacheEntries(Message message) { + if (cacheEntryExpressions.size() == 0) { + return null; + } + else { + Map cacheValues = new HashMap(); + for (Entry expressionEntry : cacheEntryExpressions.entrySet()) { + cacheValues.put(expressionEntry.getKey().getValue(message),expressionEntry.getValue().getValue(message)); + } + return cacheValues; + } + } + + public void setCacheEntries(Map cacheEntries) { + + if (cacheEntryExpressions.size() > 0) { + cacheEntryExpressions.clear(); + } + + for (Entry cacheEntry : cacheEntries.entrySet()) { + this.cacheEntryExpressions.put(new SpelExpressionParser().parseExpression(cacheEntry.getKey()), + new SpelExpressionParser().parseExpression(cacheEntry.getValue())); + } + } } diff --git a/spring-integration-gemfire/src/main/resources/META-INF/spring.handlers b/spring-integration-gemfire/src/main/resources/META-INF/spring.handlers new file mode 100644 index 0000000000..0e5c816b48 --- /dev/null +++ b/spring-integration-gemfire/src/main/resources/META-INF/spring.handlers @@ -0,0 +1 @@ +http\://www.springframework.org/schema/integration/gemfire=org.springframework.integration.gemfire.config.xml.GemfireIntegrationNamespaceHandler \ No newline at end of file diff --git a/spring-integration-gemfire/src/main/resources/META-INF/spring.schemas b/spring-integration-gemfire/src/main/resources/META-INF/spring.schemas new file mode 100644 index 0000000000..fc4b7ef1d9 --- /dev/null +++ b/spring-integration-gemfire/src/main/resources/META-INF/spring.schemas @@ -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 diff --git a/spring-integration-gemfire/src/main/resources/META-INF/spring.tooling b/spring-integration-gemfire/src/main/resources/META-INF/spring.tooling new file mode 100644 index 0000000000..54d874b910 --- /dev/null +++ b/spring-integration-gemfire/src/main/resources/META-INF/spring.tooling @@ -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 diff --git a/spring-integration-gemfire/src/main/resources/org/springframework/integration/gemfire/config/xml/spring-integration-gemfire-2.1.xsd b/spring-integration-gemfire/src/main/resources/org/springframework/integration/gemfire/config/xml/spring-integration-gemfire-2.1.xsd new file mode 100644 index 0000000000..a4c0f9ca2e --- /dev/null +++ b/spring-integration-gemfire/src/main/resources/org/springframework/integration/gemfire/config/xml/spring-integration-gemfire-2.1.xsd @@ -0,0 +1,140 @@ + + + + + + + + + + + + + + Configures an inbound Channel Adapter backed by a + Gemfire CacheListener + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + Configures an outbound Channel Adapter that writes Message payloads to a + File. + + + + + + + A map of SpEL expressions used to create cache entries. If not provided, payload must be a Map + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/spring-integration-gemfire/src/main/resources/org/springframework/integration/gemfire/config/xml/spring-integration-gemfire.gif b/spring-integration-gemfire/src/main/resources/org/springframework/integration/gemfire/config/xml/spring-integration-gemfire.gif new file mode 100644 index 0000000000..210e0764fa Binary files /dev/null and b/spring-integration-gemfire/src/main/resources/org/springframework/integration/gemfire/config/xml/spring-integration-gemfire.gif differ diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/CacheListeningMessageProducerTests.java b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/CacheListeningMessageProducerTests.java index a6f5039e5f..44127816d6 100644 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/CacheListeningMessageProducerTests.java +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/CacheListeningMessageProducerTests.java @@ -137,5 +137,4 @@ public class CacheListeningMessageProducerTests { assertNotNull(message2); assertEquals("foo was abc", message2.getPayload()); } - } diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/GemfireInboundChannelAdapterTests-context.xml b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/GemfireInboundChannelAdapterTests-context.xml new file mode 100644 index 0000000000..d31c297649 --- /dev/null +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/GemfireInboundChannelAdapterTests-context.xml @@ -0,0 +1,25 @@ + + + + + + + + + + + + + + + + + diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/GemfireInboundChannelAdapterTests.java b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/GemfireInboundChannelAdapterTests.java new file mode 100644 index 0000000000..5201d86214 --- /dev/null +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/GemfireInboundChannelAdapterTests.java @@ -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(); + } + } + +} diff --git a/spring-integration-gemfire/src/test/resources/org/springframework/integration/gemfire/inbound/cq/CqClient-context.xml b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/CqClient-context.xml similarity index 100% rename from spring-integration-gemfire/src/test/resources/org/springframework/integration/gemfire/inbound/cq/CqClient-context.xml rename to spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/CqClient-context.xml diff --git a/spring-integration-gemfire/src/test/resources/org/springframework/integration/gemfire/inbound/cq/CqServer-context.xml b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/CqServer-context.xml similarity index 100% rename from spring-integration-gemfire/src/test/resources/org/springframework/integration/gemfire/inbound/cq/CqServer-context.xml rename to spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/CqServer-context.xml diff --git a/spring-integration-gemfire/src/test/resources/org/springframework/integration/gemfire/inbound/cq/common.properties b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/common.properties similarity index 100% rename from spring-integration-gemfire/src/test/resources/org/springframework/integration/gemfire/inbound/cq/common.properties rename to spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/common.properties diff --git a/spring-integration-gemfire/src/test/resources/org/springframework/integration/gemfire/inbound/cq/gfe-cache.properties b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/gfe-cache.properties similarity index 100% rename from spring-integration-gemfire/src/test/resources/org/springframework/integration/gemfire/inbound/cq/gfe-cache.properties rename to spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/gfe-cache.properties diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/outbound/CacheWritingMessageHandlerTests.java b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/outbound/CacheWritingMessageHandlerTests.java index 074f0b5fdf..45ecb95114 100644 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/outbound/CacheWritingMessageHandlerTests.java +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/outbound/CacheWritingMessageHandlerTests.java @@ -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 regionFactoryBean = new RegionFactoryBean(); + regionFactoryBean.setName("test.expressionsWriteToCache"); + regionFactoryBean.setCache(cache); + regionFactoryBean.afterPropertiesSet(); + Region region = regionFactoryBean.getObject(); + assertEquals(0, region.size()); + CacheWritingMessageHandler handler = new CacheWritingMessageHandler(region); + + Map expressions = new HashMap(); + 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")); + } } diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/outbound/GemfireOutboundChannelAdapterTests-context.xml b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/outbound/GemfireOutboundChannelAdapterTests-context.xml new file mode 100644 index 0000000000..1d4f990a42 --- /dev/null +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/outbound/GemfireOutboundChannelAdapterTests-context.xml @@ -0,0 +1,26 @@ + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/outbound/GemfireOutboundChannelAdapterTests.java b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/outbound/GemfireOutboundChannelAdapterTests.java new file mode 100644 index 0000000000..5b26ad7b9b --- /dev/null +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/outbound/GemfireOutboundChannelAdapterTests.java @@ -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 map = new HashMap(); + 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")); + } +} diff --git a/spring-integration-gemfire/src/test/resources/org/springframework/integration/gemfire/store/messagegroupstore/GemfireMessageStore-context.xml b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/GemfireMessageStore-context.xml similarity index 100% rename from spring-integration-gemfire/src/test/resources/org/springframework/integration/gemfire/store/messagegroupstore/GemfireMessageStore-context.xml rename to spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/GemfireMessageStore-context.xml diff --git a/spring-integration-gemfire/src/test/resources/log4j.xml b/spring-integration-gemfire/src/test/resources/log4j.xml new file mode 100644 index 0000000000..d99732eeb1 --- /dev/null +++ b/spring-integration-gemfire/src/test/resources/log4j.xml @@ -0,0 +1,28 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/spring-integration-test/src/main/java/org/springframework/integration/test/support/AbstractRequestResponseScenarioTest.java b/spring-integration-test/src/main/java/org/springframework/integration/test/support/AbstractRequestResponseScenarioTest.java new file mode 100644 index 0000000000..496234c5a6 --- /dev/null +++ b/spring-integration-test/src/main/java/org/springframework/integration/test/support/AbstractRequestResponseScenarioTest.java @@ -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: + *
    + *
  • An inputChannelName
    • + *
  • An outputChannelName
    • + *
  • A payload or message to send as a request message on the inputChannel
    • + *
  • A handler to validate the response received on the outputChannel
    • + *
+ * @author David Turanski + */ +@RunWith(SpringJUnit4ClassRunner.class) +public abstract class AbstractRequestResponseScenarioTest { + private List 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 defineRequestResponseScenarios(); + + +} diff --git a/spring-integration-test/src/main/java/org/springframework/integration/test/support/AbstractResponseValidator.java b/spring-integration-test/src/main/java/org/springframework/integration/test/support/AbstractResponseValidator.java new file mode 100644 index 0000000000..15fb868baf --- /dev/null +++ b/spring-integration-test/src/main/java/org/springframework/integration/test/support/AbstractResponseValidator.java @@ -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(); + +} diff --git a/spring-integration-test/src/main/java/org/springframework/integration/test/support/MessageValidator.java b/spring-integration-test/src/main/java/org/springframework/integration/test/support/MessageValidator.java new file mode 100644 index 0000000000..47c077adb3 --- /dev/null +++ b/spring-integration-test/src/main/java/org/springframework/integration/test/support/MessageValidator.java @@ -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); + +} diff --git a/spring-integration-test/src/main/java/org/springframework/integration/test/support/PayloadValidator.java b/spring-integration-test/src/main/java/org/springframework/integration/test/support/PayloadValidator.java new file mode 100644 index 0000000000..8fda95ba58 --- /dev/null +++ b/spring-integration-test/src/main/java/org/springframework/integration/test/support/PayloadValidator.java @@ -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; + } +} diff --git a/spring-integration-test/src/main/java/org/springframework/integration/test/support/RequestResponseScenario.java b/spring-integration-test/src/main/java/org/springframework/integration/test/support/RequestResponseScenario.java new file mode 100644 index 0000000000..32e15d1dd7 --- /dev/null +++ b/spring-integration-test/src/main/java/org/springframework/integration/test/support/RequestResponseScenario.java @@ -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 getMessage(){ + if (message == null){ + return new GenericMessage(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"); + } +} diff --git a/spring-integration-test/src/test/java/org/springframework/integration/test/support/MessageScenariosTest-context.xml b/spring-integration-test/src/test/java/org/springframework/integration/test/support/MessageScenariosTest-context.xml new file mode 100644 index 0000000000..3b4184b2b4 --- /dev/null +++ b/spring-integration-test/src/test/java/org/springframework/integration/test/support/MessageScenariosTest-context.xml @@ -0,0 +1,17 @@ + + + + + + + + + + + + + diff --git a/spring-integration-test/src/test/java/org/springframework/integration/test/support/MessageScenariosTest.java b/spring-integration-test/src/test/java/org/springframework/integration/test/support/MessageScenariosTest.java new file mode 100644 index 0000000000..15c3f92676 --- /dev/null +++ b/spring-integration-test/src/test/java/org/springframework/integration/test/support/MessageScenariosTest.java @@ -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 defineRequestResponseScenarios() { + List scenarios= new ArrayList(); + 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; + } +}