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

Merge pull request #437 from dturanski/INT-2550

Browse Source 
* INT-2550:
  INT-2550,INT-2552:added support for properties regionName and region to GemfireMessageStore
This commit is contained in:
Oleg Zhurakousky
2012-05-15 13:12:02 -04:00
parent 2061c37608 6dd48eff60
commit c0e03a1666
5 changed files with 114 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
build.gradle
View File

@@ -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'

39
spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/store/GemfireMessageStore.java
View File

@@ -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<Object, Object> 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<Object,Object> 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<Object, Object> messageRegionFactoryBean = new RegionFactoryBean<Object, Object>();
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<Object> keyList = new ArrayList<Object>();
for (Object key : keys) {
String keyValue = key.toString();
if (PatternMatchUtils.simpleMatch(keyPattern, keyValue)){
if (PatternMatchUtils.simpleMatch(keyPattern, keyValue)) {
keyList.add(keyValue);
}
}

2
spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/CqInboundChannelAdapterTests-context.xml
View File

@@ -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">
<gfe:cache use-bean-factory-locator="false" id="client-cache"/>
<gfe:client-cache use-bean-factory-locator="false" id="client-cache" pool-name="client-pool"/>
<gfe:pool id="client-pool" subscription-enabled="true" >
<gfe:server host="localhost" port="40404"/>

25
spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireMessageStoreTests.java
View File

@@ -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<Object, Object> region = new RegionFactoryBean<Object, Object>();
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);

92
src/reference/docbook/gemfire.xml
View File

@@ -11,8 +11,8 @@
<para>
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 <ulink url="http://www.gemstone.com/docs/6.6.RC/product/docs/html/user_guide/UserGuide_GemFire.html#Getting%20Started%20with%20Gemfire">GemFire</ulink>
and its <ulink url="http://www.gemstone.com/docs/6.6.RC/product/docs/japi/index.html">API</ulink>.
some familiarity with <ulink url="http://www.vmware.com/support/pubs/vfabric-gemfire.html">GemFire</ulink>
and its <ulink url="http://www.vmware.com/support/developer/vfabric-gemfire/662-api/index.html">API</ulink>.
</para>
<para>
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
</note>
<programlisting language="xml">
<![CDATA[<gfe:cache id="client-cache"/>
<programlisting language="xml"><![CDATA[<gfe:client-cache id="client-cache" pool-name="client-pool"/>
<gfe:pool id="client-pool" subscription-enabled="true" >
<!--configure server or locator here required to address the cache server -->
@@ -84,7 +83,7 @@ xsi:schemaLocation="http://www.springframework.org/schema/integration/gemfire
</programlisting>
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 <classname>ContinuousQueryListenerContainer</classname>
(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 <classname>ContinuousQueryListenerContainer</classname>
using Spring GemFire. The continuous query inbound channel adapter requires a <code>cq-listener-container</code> attribute which contains a reference to the <classname>ContinuousQueryListenerContainer</classname>. Optionally,
it accepts an <code>expression</code> attribute which uses SpEL to transform the <code>CqEvent</code> or extract an individual property as needed. The cq-inbound-channel-adapter provides a
<code>query-events</code> 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 <emphasis>outbound-channel-adapter</emphasis> writes cache entries mapped from the message payload. In its simplest form, it expects a
payload of type <classname>java.util.Map</classname> and puts the map entries into its configured region.
<programlisting language="xml">
<![CDATA[<int-gfe:outbound-channel-adapter id="cacheChannel" region="region"/>]]>
</programlisting>
<programlisting language="xml"><![CDATA[
<int-gfe:outbound-channel-adapter id="cacheChannel" region="region"/>]]>
</programlisting>
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.
<programlisting language="xml">
<![CDATA[<int-gfe:outbound-channel-adapter id="cacheChannel" region="region">
<int-gfe:cache-entries>
<entry key="payload.toUpperCase()" value="payload.toLowerCase()"/>
<entry key="'foo'" value="'bar'"/>
</int-gfe:cache-entries>
<programlisting language="xml"><![CDATA[
<int-gfe:outbound-channel-adapter id="cacheChannel" region="region">
<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>
]]>
</programlisting>
]]>
</programlisting>
In the above configuration, the inner element <code>cache-entries</code> is semantically equivalent to Spring 'map' element. The adapter interprets the <code>key</code> and
<code>value</code> 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 <classname>MessageStore</classname> strategy (mainly used by the <emphasis>QueueChannel</emphasis> and <emphasis>ClaimCheck</emphasis>
patterns) and the <classname>MessageGroupStore</classname> strategy (mainly used by the <emphasis>Aggregator</emphasis> and
<emphasis>Resequencer</emphasis> patterns).
</para>
<para>
<programlisting lang="xml"><![CDATA[<bean id="gemfireMessageStore" class="org.springframework.integration.gemfire.store.GemfireMessageStore">
<constructor-arg ref="myCache"/>
</bean>
<programlisting language="xml"><![CDATA[
<bean id="gemfireMessageStore" class="org.springframework.integration.gemfire.store.GemfireMessageStore">
<constructor-arg ref="myCache"/>
</bean>
<bean id="myCache" class="org.springframework.data.gemfire.CacheFactoryBean"/>
<bean id="myCache" class="org.springframework.data.gemfire.CacheFactoryBean"/>
<int:channel id="somePersistentQueueChannel">
<int:queue message-store="gemfireMessageStore"/>
<int:channel>
<int:aggregator input-channel="inputChannel" output-channel="outputChannel"
message-store="gemfireMessageStore"/>]]></programlisting>
</para>
message-store="gemfireMessageStore"/>
]]></programlisting>
</para>
<para>
Above is a sample <classname>GemfireMessageStore</classname> configuration that shows its usage by a <emphasis>QueueChannel</emphasis>
and an <emphasis>Aggregator</emphasis>. As you can see it is a simple bean configuration, and it expects a
<classname>GemFireCache</classname> (created by <classname>CacheFactoryBean</classname>) as a constructor argument.
</para>
</section>
<para>
Above is a sample <classname>GemfireMessageStore</classname> configuration that shows its usage by a <emphasis>QueueChannel</emphasis> and an <emphasis>Aggregator</emphasis>. As you can see it is a normal Spring bean configuration. The simplest configuration requires a reference to a <classname>GemFireCache</classname> (created by <classname>CacheFactoryBean</classname>) 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 <classname>Region</classname> as the constructor argument:
<programlisting language="xml"><![CDATA[
<bean id="gemfireMessageStore" class="org.springframework.integration.gemfire.store.GemfireMessageStore">
<constructor-arg ref="myRegion"/>
</bean>
<gfe:cache/>
<gfe:replicated-region id="myRegion"/>
]]></programlisting>
</para>
<para>
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
<ulink url="http://www.vmware.com/support/pubs/vfabric-gemfire.html">GemFire product documentation</ulink> 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:
<programlisting language="xml"><![CDATA[
<bean id="gemfireMessageStore"
class="org.springframework.integration.gemfire.store.GemfireMessageStore">
<constructor-arg ref="myRegion"/>
</bean>
<gfe:client-cache/>
<gfe:client-region id="myRegion" shortcut="PROXY" pool-name="messageStorePool"/>
<gfe:pool id="messageStorePool">
<gfe:server host="localhost" port="40404" />
</gfe:pool>
]]></programlisting>
</para>
<para>
Note the <emphasis>pool</emphasis> 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.
</para>
</section>
</chapter>