diff --git a/build.gradle b/build.gradle index 060323abdc..f494af88ee 100644 --- a/build.gradle +++ b/build.gradle @@ -126,7 +126,7 @@ configure(javaprojects) { springAmqpVersion = '1.0.0.RELEASE' springDataMongoVersion = '1.0.0.M4' springDataRedisVersion = '1.0.0.M4' - springGemfireVersion = '1.1.0.M2' + springGemfireVersion = '1.1.0.M3' springSecurityVersion = '3.0.6.RELEASE' springWsVersion = '2.0.2.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 98774a464e..c3527de24b 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 @@ -20,14 +20,17 @@ import java.util.ArrayList; import java.util.Collection; import java.util.List; +import org.springframework.beans.factory.InitializingBean; import org.springframework.data.gemfire.RegionFactoryBean; import org.springframework.integration.store.AbstractKeyValueMessageStore; import org.springframework.integration.store.MessageGroupStore; +import org.springframework.integration.store.MessageStore; import org.springframework.util.Assert; import org.springframework.util.PatternMatchUtils; import com.gemstone.gemfire.cache.Cache; import com.gemstone.gemfire.cache.Region; +import com.gemstone.gemfire.internal.cache.xmlcache.RegionAttributesCreation; /** * Gemfire implementation of the key/value style {@link MessageStore} and {@link MessageGroupStore} @@ -36,21 +39,21 @@ import com.gemstone.gemfire.cache.Region; * @author Oleg Zhurakousky * @since 2.1 */ -public class GemfireMessageStore extends AbstractKeyValueMessageStore{ +public class GemfireMessageStore extends AbstractKeyValueMessageStore implements InitializingBean{ - private final Region messageStoreRegion; + private volatile Region messageStoreRegion; + private final Cache cache; + + private volatile boolean ignoreJta = true; + public GemfireMessageStore(Cache cache) { Assert.notNull(cache, "'cache' must not be null"); - try { - RegionFactoryBean messageRegionFactoryBean = new RegionFactoryBean(); - messageRegionFactoryBean.setBeanName("messageStoreRegion"); - messageRegionFactoryBean.setCache(cache); - messageRegionFactoryBean.afterPropertiesSet(); - this.messageStoreRegion = messageRegionFactoryBean.getObject(); - } catch (Exception e) { - throw new IllegalArgumentException("Failed to initialize Gemfire Region"); - } + this.cache = cache; + } + + public void setIgnoreJta(boolean ignoreJta) { + this.ignoreJta = ignoreJta; } @Override @@ -85,4 +88,21 @@ public class GemfireMessageStore extends AbstractKeyValueMessageStore{ } return keyList; } + + @SuppressWarnings("unchecked") + public void afterPropertiesSet() throws Exception { + try { + RegionAttributesCreation attributes = new RegionAttributesCreation(); + attributes.setIgnoreJTA(ignoreJta); + RegionFactoryBean messageRegionFactoryBean = new RegionFactoryBean(); + messageRegionFactoryBean.setBeanName("messageStoreRegion"); + messageRegionFactoryBean.setAttributes(attributes); + messageRegionFactoryBean.setCache(cache); + messageRegionFactoryBean.afterPropertiesSet(); + + this.messageStoreRegion = messageRegionFactoryBean.getObject(); + } catch (Exception e) { + throw new IllegalArgumentException("Failed to initialize Gemfire Region", e); + } + } } diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireGroupStoreTests.java b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireGroupStoreTests.java index 4c59ebc1d9..1997edab18 100644 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireGroupStoreTests.java +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireGroupStoreTests.java @@ -57,7 +57,7 @@ public class GemfireGroupStoreTests { @Test public void testNonExistingEmptyMessageGroup() throws Exception{ GemfireMessageStore store = new GemfireMessageStore(this.cache); - + store.afterPropertiesSet(); MessageGroup messageGroup = store.getMessageGroup(1); assertNotNull(messageGroup); assertTrue(messageGroup instanceof SimpleMessageGroup); @@ -67,7 +67,7 @@ public class GemfireGroupStoreTests { @Test public void testMessageGroupWithAddedMessage() throws Exception{ GemfireMessageStore store = new GemfireMessageStore(this.cache); - + store.afterPropertiesSet(); MessageGroup messageGroup = store.getMessageGroup(1); Message message = new GenericMessage("Hello"); messageGroup = store.addMessageToGroup(1, message); @@ -75,7 +75,8 @@ public class GemfireGroupStoreTests { // make sure the store is properly rebuild from Gemfire store = new GemfireMessageStore(this.cache); - + store.afterPropertiesSet(); + messageGroup = store.getMessageGroup(1); assertEquals(1, messageGroup.size()); } @@ -83,7 +84,7 @@ public class GemfireGroupStoreTests { @Test public void testRemoveMessageGroup() throws Exception{ GemfireMessageStore store = new GemfireMessageStore(this.cache); - + store.afterPropertiesSet(); MessageGroup messageGroup = store.getMessageGroup(1); Message message = new GenericMessage("Hello"); messageGroup = store.addMessageToGroup(messageGroup.getGroupId(), message); @@ -98,7 +99,8 @@ public class GemfireGroupStoreTests { // make sure the store is properly rebuild from Gemfire store = new GemfireMessageStore(this.cache); - + store.afterPropertiesSet(); + messageGroup = store.getMessageGroup(1); assertEquals(0, messageGroup.getMarked().size()); @@ -109,7 +111,7 @@ public class GemfireGroupStoreTests { @Test public void testRemoveMessageFromTheGroup() throws Exception{ GemfireMessageStore store = new GemfireMessageStore(this.cache); - + store.afterPropertiesSet(); MessageGroup messageGroup = store.getMessageGroup(1); Message message = new GenericMessage("2"); store.addMessageToGroup(messageGroup.getGroupId(), new GenericMessage("1")); @@ -122,7 +124,8 @@ public class GemfireGroupStoreTests { // make sure the store is properly rebuild from Gemfire store = new GemfireMessageStore(this.cache); - + store.afterPropertiesSet(); + messageGroup = store.getMessageGroup(1); assertEquals(2, messageGroup.size()); @@ -131,7 +134,7 @@ public class GemfireGroupStoreTests { @Test public void testMarkAllMessagesInMessageGroup() throws Exception{ GemfireMessageStore store = new GemfireMessageStore(this.cache); - + store.afterPropertiesSet(); MessageGroup messageGroup = store.getMessageGroup(1); store.addMessageToGroup(messageGroup.getGroupId(), new GenericMessage("1")); store.addMessageToGroup(messageGroup.getGroupId(), new GenericMessage("2")); @@ -146,7 +149,8 @@ public class GemfireGroupStoreTests { // make sure the store is properly rebuild from Gemfire store = new GemfireMessageStore(this.cache); - + store.afterPropertiesSet(); + messageGroup = store.getMessageGroup(1); assertEquals(0, messageGroup.getUnmarked().size()); assertEquals(3, messageGroup.getMarked().size()); @@ -155,7 +159,7 @@ public class GemfireGroupStoreTests { @Test public void testRemoveNonExistingMessageFromTheGroup() throws Exception{ GemfireMessageStore store = new GemfireMessageStore(this.cache); - + store.afterPropertiesSet(); MessageGroup messageGroup = store.getMessageGroup(1); store.addMessageToGroup(messageGroup.getGroupId(), new GenericMessage("1")); store.removeMessageFromGroup(1, new GenericMessage("2")); @@ -164,12 +168,14 @@ public class GemfireGroupStoreTests { @Test public void testRemoveNonExistingMessageFromNonExistingTheGroup() throws Exception{ GemfireMessageStore store = new GemfireMessageStore(this.cache); + store.afterPropertiesSet(); store.removeMessageFromGroup(1, new GenericMessage("2")); } @Test public void testMarkMessageInMessageGroup() throws Exception{ GemfireMessageStore store = new GemfireMessageStore(this.cache); + store.afterPropertiesSet(); MessageGroup messageGroup = store.getMessageGroup(1); Message messageToMark = new GenericMessage("1"); store.addMessageToGroup(messageGroup.getGroupId(), messageToMark); @@ -184,7 +190,8 @@ public class GemfireGroupStoreTests { // make sure the store is properly rebuild from Gemfire store = new GemfireMessageStore(this.cache); - + store.afterPropertiesSet(); + messageGroup = store.getMessageGroup(1); assertEquals(2, messageGroup.getUnmarked().size()); assertEquals(1, messageGroup.getMarked().size()); @@ -193,6 +200,7 @@ public class GemfireGroupStoreTests { @Test public void testCompleteMessageGroup() throws Exception{ GemfireMessageStore store = new GemfireMessageStore(this.cache); + store.afterPropertiesSet(); MessageGroup messageGroup = store.getMessageGroup(1); Message messageToMark = new GenericMessage("1"); store.addMessageToGroup(messageGroup.getGroupId(), messageToMark); @@ -204,6 +212,7 @@ public class GemfireGroupStoreTests { @Test public void testLastReleasedSequenceNumber() throws Exception{ GemfireMessageStore store = new GemfireMessageStore(this.cache); + store.afterPropertiesSet(); MessageGroup messageGroup = store.getMessageGroup(1); Message messageToMark = new GenericMessage("1"); store.addMessageToGroup(messageGroup.getGroupId(), messageToMark); @@ -215,8 +224,10 @@ public class GemfireGroupStoreTests { @Test public void testMultipleInstancesOfGroupStore() throws Exception{ GemfireMessageStore store1 = new GemfireMessageStore(this.cache); + store1.afterPropertiesSet(); GemfireMessageStore store2 = new GemfireMessageStore(this.cache); + store2.afterPropertiesSet(); Message message = new GenericMessage("1"); store1.addMessageToGroup(1, message); @@ -226,6 +237,7 @@ public class GemfireGroupStoreTests { assertEquals(0, messageGroup.getMarked().size()); GemfireMessageStore store3 = new GemfireMessageStore(this.cache); + store3.afterPropertiesSet(); messageGroup = store3.markMessageFromGroup(1, message); @@ -236,7 +248,9 @@ public class GemfireGroupStoreTests { @Test public void testIteratorOfMessageGroups() throws Exception{ GemfireMessageStore store1 = new GemfireMessageStore(this.cache); + store1.afterPropertiesSet(); GemfireMessageStore store2 = new GemfireMessageStore(this.cache); + store2.afterPropertiesSet(); store1.addMessageToGroup(1, new GenericMessage("1")); store2.addMessageToGroup(2, new GenericMessage("2")); @@ -267,7 +281,9 @@ public class GemfireGroupStoreTests { public void testConcurrentModifications() throws Exception{ final GemfireMessageStore store1 = new GemfireMessageStore(this.cache); + store1.afterPropertiesSet(); final GemfireMessageStore store2 = new GemfireMessageStore(this.cache); + store2.afterPropertiesSet(); final Message message = new GenericMessage("1"); 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 28b65a40d2..2245c8f2af 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 @@ -19,7 +19,6 @@ package org.springframework.integration.gemfire.store; import org.junit.Test; import org.springframework.data.gemfire.CacheFactoryBean; import org.springframework.integration.Message; -import org.springframework.integration.store.MessageStore; import org.springframework.integration.support.MessageBuilder; import com.gemstone.gemfire.cache.Cache; @@ -37,7 +36,8 @@ public class GemfireMessageStoreTests { CacheFactoryBean cacheFactoryBean = new CacheFactoryBean(); cacheFactoryBean.afterPropertiesSet(); Cache cache = (Cache)cacheFactoryBean.getObject(); - MessageStore store = new GemfireMessageStore(cache); + GemfireMessageStore store = new GemfireMessageStore(cache); + store.afterPropertiesSet(); Message message = MessageBuilder.withPayload("test").build(); store.addMessage(message);