From 6dd48eff609115a62c6a7bc33f19127b8c92699e Mon Sep 17 00:00:00 2001 From: David Turanski Date: Mon, 14 May 2012 13:39:34 -0400 Subject: [PATCH] INT-2550,INT-2552:added support for properties regionName and region to GemfireMessageStore INT-2555:Upgraded to spring-data-gemfire 1.1.1.RELEASE document update INT-2550,INT-2552: updated GemfireMessageStore, test, and docs per Oleg's comments removed setRegion --- build.gradle | 2 +- .../gemfire/store/GemfireMessageStore.java | 39 ++++++-- .../CqInboundChannelAdapterTests-context.xml | 2 +- .../store/GemfireMessageStoreTests.java | 25 ++++- src/reference/docbook/gemfire.xml | 92 ++++++++++++------- 5 files changed, 114 insertions(+), 46 deletions(-) diff --git a/build.gradle b/build.gradle index 13442b5748..e1091261fe 100644 --- a/build.gradle +++ b/build.gradle @@ -50,7 +50,7 @@ subprojects { subproject -> springAmqpVersion = '1.1.0.RELEASE' springDataMongoVersion = '1.1.0.M1' springDataRedisVersion = '1.0.0.RELEASE' - springGemfireVersion = '1.1.0.RELEASE' + springGemfireVersion = '1.1.1.RELEASE' springSecurityVersion = '3.1.0.RELEASE' springSocialTwitterVersion = '1.0.1.RELEASE' springWsVersion = '2.0.3.RELEASE' diff --git a/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/store/GemfireMessageStore.java b/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/store/GemfireMessageStore.java index 9bdc9a7e2e..3bd1921329 100644 --- a/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/store/GemfireMessageStore.java +++ b/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/store/GemfireMessageStore.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 the original author or authors. + * Copyright 2002-2012 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. @@ -33,39 +33,64 @@ import com.gemstone.gemfire.cache.Cache; import com.gemstone.gemfire.cache.Region; /** - * Gemfire implementation of the key/value style {@link MessageStore} and {@link MessageGroupStore} + * Gemfire implementation of the key/value style {@link MessageStore} and + * {@link MessageGroupStore} * * @author Mark Fisher * @author Oleg Zhurakousky + * @author David Turanski * @since 2.1 */ public class GemfireMessageStore extends AbstractKeyValueMessageStore implements InitializingBean { + private static final String MESSAGE_STORE_REGION_NAME = "messageStoreRegion"; + private volatile Region messageStoreRegion; private final Cache cache; private volatile boolean ignoreJta = true; - + /** + * Provides the region to be used for the message store. This is useful when + * using a configured region. This is also required if using a client region + * on a remote cache server. + * @param messageStoreRegion the region + */ + public GemfireMessageStore(Region messageStoreRegion) { + cache = null; + this.messageStoreRegion = messageStoreRegion; + } + /** + * Provides a cache reference used to create a message store region named + * 'messageStoreRegion' + * @param cache + */ public GemfireMessageStore(Cache cache) { Assert.notNull(cache, "'cache' must not be null"); this.cache = cache; } - public void setIgnoreJta(boolean ignoreJta) { this.ignoreJta = ignoreJta; } - + @SuppressWarnings("unchecked") public void afterPropertiesSet() { + if (this.messageStoreRegion != null) { + return; + } + try { + if (logger.isDebugEnabled()){ + logger.debug("creating message store region as '" + MESSAGE_STORE_REGION_NAME + "'"); + } + RegionAttributesFactoryBean attributesFactoryBean = new RegionAttributesFactoryBean(); attributesFactoryBean.setIgnoreJTA(this.ignoreJta); attributesFactoryBean.afterPropertiesSet(); RegionFactoryBean messageRegionFactoryBean = new RegionFactoryBean(); - messageRegionFactoryBean.setBeanName("messageStoreRegion"); + messageRegionFactoryBean.setBeanName(MESSAGE_STORE_REGION_NAME); messageRegionFactoryBean.setAttributes(attributesFactoryBean.getObject()); messageRegionFactoryBean.setCache(cache); messageRegionFactoryBean.afterPropertiesSet(); @@ -102,7 +127,7 @@ public class GemfireMessageStore extends AbstractKeyValueMessageStore implements List keyList = new ArrayList(); for (Object key : keys) { String keyValue = key.toString(); - if (PatternMatchUtils.simpleMatch(keyPattern, keyValue)){ + if (PatternMatchUtils.simpleMatch(keyPattern, keyValue)) { keyList.add(keyValue); } } diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/CqInboundChannelAdapterTests-context.xml b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/CqInboundChannelAdapterTests-context.xml index b95ed9d2fb..75ed42757a 100644 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/CqInboundChannelAdapterTests-context.xml +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/CqInboundChannelAdapterTests-context.xml @@ -11,7 +11,7 @@ http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util.xsd"> - + diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireMessageStoreTests.java b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireMessageStoreTests.java index 2884467f26..e277ee2a27 100644 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireMessageStoreTests.java +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireMessageStoreTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2012 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. @@ -16,27 +16,30 @@ package org.springframework.integration.gemfire.store; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertSame; + import java.util.Properties; import org.junit.After; import org.junit.Before; import org.junit.Test; - import org.springframework.data.gemfire.CacheFactoryBean; +import org.springframework.data.gemfire.RegionFactoryBean; import org.springframework.integration.Message; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.history.MessageHistory; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.support.MessageBuilder; +import org.springframework.integration.test.util.TestUtils; import org.springframework.util.Assert; import com.gemstone.gemfire.cache.Cache; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNotNull; - /** * @author Mark Fisher + * @author David Turanski * @since 2.1 */ public class GemfireMessageStoreTests { @@ -54,6 +57,18 @@ public class GemfireMessageStoreTests { assertEquals(message, retrieved); } + @Test + public void testRegionConstructor() throws Exception { + RegionFactoryBean region = new RegionFactoryBean(); + region.setName("someRegion"); + region.setCache(this.cache); + region.afterPropertiesSet(); + + GemfireMessageStore store = new GemfireMessageStore(region.getObject()); + store.afterPropertiesSet(); + assertSame(region.getObject(),TestUtils.getPropertyValue(store, "messageStoreRegion")); + } + @Test public void testWithMessageHistory() throws Exception{ GemfireMessageStore store = new GemfireMessageStore(this.cache); diff --git a/src/reference/docbook/gemfire.xml b/src/reference/docbook/gemfire.xml index b29cae3386..99c41fc9f3 100644 --- a/src/reference/docbook/gemfire.xml +++ b/src/reference/docbook/gemfire.xml @@ -11,8 +11,8 @@ VMWare vFabric GemFire (GemFire) is a distributed data management platform providing a key-value data grid along with advanced distributed system features such as event processing, continuous querying, and remote function execution. This guide assumes - some familiarity with GemFire - and its API. + some familiarity with GemFire + and its API. Spring integration provides support for GemFire by providing inbound adapters for entry and continuous query events, @@ -65,8 +65,7 @@ xsi:schemaLocation="http://www.springframework.org/schema/integration/gemfire - - + @@ -84,7 +83,7 @@ xsi:schemaLocation="http://www.springframework.org/schema/integration/gemfire In the above configuration, we are creating a GemFire client cache - (recall a cache server is required for this implementation and its address is configured as a sub-element of the pool), a client region and a ContinuousQueryListenerContainer + (recall a remote cache server is required for this implementation and its address is configured as a sub-element of the pool), a client region and a ContinuousQueryListenerContainer using Spring GemFire. The continuous query inbound channel adapter requires a cq-listener-container attribute which contains a reference to the ContinuousQueryListenerContainer. Optionally, it accepts an expression attribute which uses SpEL to transform the CqEvent or extract an individual property as needed. The cq-inbound-channel-adapter provides a query-events attribute, containing a comma separated list of event types for which a message will be produced on the input channel. Available event types are CREATED, UPDATED, DESTROYED, @@ -101,23 +100,22 @@ xsi:schemaLocation="http://www.springframework.org/schema/integration/gemfire The outbound-channel-adapter writes cache entries mapped from the message payload. In its simplest form, it expects a payload of type java.util.Map and puts the map entries into its configured region. - - ]]> - - +]]> + Given the above configuration, an exception will be thrown if the payload is not a Map. Additionally, the outbound channel adapter can be configured to create a map of cache entries using SpEL of course. - - - - - - + + + + + - ]]> - +]]> + In the above configuration, the inner element cache-entries is semantically equivalent to Spring 'map' element. The adapter interprets the key and value attributes as SpEL expressions with the message as the evaluation context. Note that this contain arbitrary cache entries (not only those derived from the message) and that literal values must be enclosed in single quotes. In the above example, if the message sent to @@ -142,27 +140,57 @@ xsi:schemaLocation="http://www.springframework.org/schema/integration/gemfire the MessageStore strategy (mainly used by the QueueChannel and ClaimCheck patterns) and the MessageGroupStore strategy (mainly used by the Aggregator and Resequencer patterns). - - - - - - + + + - + ]]> - + message-store="gemfireMessageStore"/> +]]> + - - Above is a sample GemfireMessageStore configuration that shows its usage by a QueueChannel - and an Aggregator. As you can see it is a simple bean configuration, and it expects a - GemFireCache (created by CacheFactoryBean) as a constructor argument. - - + +Above is a sample GemfireMessageStore configuration that shows its usage by a QueueChannel and an Aggregator. As you can see it is a normal Spring bean configuration. The simplest configuration requires a reference to a GemFireCache (created by CacheFactoryBean) as a constructor argument. If the cache is standalone, i.e., embedded in the same JVM, the MessageStore will create a message store region named "messageStoreRegion". If your application requires customization of the messageStore region, for example, multiple Gemfire message stores each with its own region, you can configure a region for each message store instance and use the Region as the constructor argument: + + + + + + + + +]]> + + + +In the above examle, the cache and region are configured using the spring-gemfire namespace (not to be confused with the spring-integration-gemfire namespace). Often it is desirable for the message store to be maintained in one or more remote cache servers in a client-server configuration (See the +GemFire product documentation for more details). In this case, you configure a client cache, client region, and client pool and inject the region into the MessageStore. Here is an example: + + + + + + + + + + + + +]]> + + +Note the pool element is configured with the address of a cache server (a locator may be substituted here). The region is configured as a 'PROXY' so that no data will be stored locally. The region's id corresponds to a region with the same name configured in the cache server. + +