diff --git a/build.gradle b/build.gradle index e0fbb1bd7d..27b775abb7 100644 --- a/build.gradle +++ b/build.gradle @@ -52,7 +52,7 @@ subprojects { subproject -> springAmqpVersion = '1.2.0.M1' springDataMongoVersion = '1.1.1.RELEASE' springDataRedisVersion = '1.0.2.RELEASE' - springGemfireVersion = '1.2.2.RELEASE' + springGemfireVersion = '1.3.1.RELEASE' springSecurityVersion = '3.1.3.RELEASE' springSocialTwitterVersion = '1.0.1.RELEASE' springWsVersion = '2.1.1.RELEASE' diff --git a/spring-integration-gemfire/src/main/resources/org/springframework/integration/gemfire/config/xml/spring-integration-gemfire-3.0.xsd b/spring-integration-gemfire/src/main/resources/org/springframework/integration/gemfire/config/xml/spring-integration-gemfire-3.0.xsd index cee663565f..5e27f28590 100644 --- a/spring-integration-gemfire/src/main/resources/org/springframework/integration/gemfire/config/xml/spring-integration-gemfire-3.0.xsd +++ b/spring-integration-gemfire/src/main/resources/org/springframework/integration/gemfire/config/xml/spring-integration-gemfire-3.0.xsd @@ -7,10 +7,10 @@ elementFormDefault="qualified" attributeFormDefault="unqualified"> + schemaLocation="http://www.springframework.org/schema/beans/spring-beans.xsd" /> + schemaLocation="http://www.springframework.org/schema/integration/spring-integration.xsd" /> regionFactoryBean = new RegionFactoryBean(); regionFactoryBean.setName("test.receiveNewValuePayloadForCreateEvent"); regionFactoryBean.setCache(cache); this.setRegionAttributes(regionFactoryBean); - + regionFactoryBean.afterPropertiesSet(); Region region = regionFactoryBean.getObject(); QueueChannel channel = new QueueChannel(); @@ -65,14 +64,13 @@ public class CacheListeningMessageProducerTests { @Test public void receiveNewValuePayloadForUpdateEvent() throws Exception { CacheFactoryBean cacheFactoryBean = new CacheFactoryBean(); - cacheFactoryBean.afterPropertiesSet(); - Cache cache = (Cache)cacheFactoryBean.getObject(); - + Cache cache = cacheFactoryBean.getObject(); + RegionFactoryBean regionFactoryBean = new RegionFactoryBean(); regionFactoryBean.setName("test.receiveNewValuePayloadForUpdateEvent"); regionFactoryBean.setCache(cache); this.setRegionAttributes(regionFactoryBean); - + regionFactoryBean.afterPropertiesSet(); Region region = regionFactoryBean.getObject(); QueueChannel channel = new QueueChannel(); @@ -89,20 +87,19 @@ public class CacheListeningMessageProducerTests { region.put("x", "xyz"); Message message2 = channel.receive(0); assertNotNull(message2); - assertEquals("xyz", message2.getPayload()); + assertEquals("xyz", message2.getPayload()); } @Test public void receiveOldValuePayloadForDestroyEvent() throws Exception { CacheFactoryBean cacheFactoryBean = new CacheFactoryBean(); - cacheFactoryBean.afterPropertiesSet(); - Cache cache = (Cache)cacheFactoryBean.getObject(); - + Cache cache = cacheFactoryBean.getObject(); + RegionFactoryBean regionFactoryBean = new RegionFactoryBean(); regionFactoryBean.setName("test.receiveOldValuePayloadForDestroyEvent"); regionFactoryBean.setCache(cache); this.setRegionAttributes(regionFactoryBean); - + regionFactoryBean.afterPropertiesSet(); Region region = regionFactoryBean.getObject(); QueueChannel channel = new QueueChannel(); @@ -118,20 +115,19 @@ public class CacheListeningMessageProducerTests { region.destroy("foo"); Message message2 = channel.receive(0); assertNotNull(message2); - assertEquals("abc", message2.getPayload()); + assertEquals("abc", message2.getPayload()); } @Test public void receiveOldValuePayloadForInvalidateEvent() throws Exception { CacheFactoryBean cacheFactoryBean = new CacheFactoryBean(); - cacheFactoryBean.afterPropertiesSet(); - Cache cache = (Cache)cacheFactoryBean.getObject(); - + Cache cache = cacheFactoryBean.getObject(); + RegionFactoryBean regionFactoryBean = new RegionFactoryBean(); regionFactoryBean.setName("test.receiveOldValuePayloadForDestroyEvent"); regionFactoryBean.setCache(cache); this.setRegionAttributes(regionFactoryBean); - + regionFactoryBean.afterPropertiesSet(); Region region = regionFactoryBean.getObject(); QueueChannel channel = new QueueChannel(); @@ -147,13 +143,12 @@ public class CacheListeningMessageProducerTests { region.invalidate("foo"); Message message2 = channel.receive(0); assertNotNull(message2); - assertEquals("foo was abc", message2.getPayload()); + assertEquals("foo was abc", message2.getPayload()); } - + @SuppressWarnings("unchecked") - private void setRegionAttributes(RegionFactoryBean regionFactoryBean) throws Exception{ + private void setRegionAttributes(RegionFactoryBean regionFactoryBean) throws Exception { RegionAttributesFactoryBean attributesFactoryBean = new RegionAttributesFactoryBean(); - attributesFactoryBean.setIgnoreJTA(true); attributesFactoryBean.afterPropertiesSet(); regionFactoryBean.setAttributes(attributesFactoryBean.getObject()); } 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 75ed42757a..8c369c1d9b 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 @@ -1,41 +1,41 @@ - - - - + + + + - - - - - - - - - + + + + + + + - + - - - - + + + - + - + 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 e2a2d8da5e..1a54aac995 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 @@ -43,8 +43,7 @@ public class CacheWritingMessageHandlerTests { @Test public void mapPayloadWritesToCache() throws Exception { CacheFactoryBean cacheFactoryBean = new CacheFactoryBean(); - cacheFactoryBean.afterPropertiesSet(); - Cache cache = (Cache)cacheFactoryBean.getObject(); + Cache cache = cacheFactoryBean.getObject(); RegionFactoryBean regionFactoryBean = new RegionFactoryBean(); regionFactoryBean.setName("test.mapPayloadWritesToCache"); regionFactoryBean.setCache(cache); @@ -63,8 +62,7 @@ public class CacheWritingMessageHandlerTests { @Test public void ExpressionsWriteToCache() throws Exception { CacheFactoryBean cacheFactoryBean = new CacheFactoryBean(); - cacheFactoryBean.afterPropertiesSet(); - Cache cache = (Cache)cacheFactoryBean.getObject(); + Cache cache = cacheFactoryBean.getObject(); RegionFactoryBean regionFactoryBean = new RegionFactoryBean(); regionFactoryBean.setName("test.expressionsWriteToCache"); regionFactoryBean.setCache(cache); 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 93dbfea660..0e9a15b0cd 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 @@ -53,14 +53,15 @@ import static org.junit.Assert.assertTrue; /** * @author Oleg Zhurakousky + * @author David Turanski * */ public class GemfireGroupStoreTests { - + private Cache cache; @Test - public void testNonExistingEmptyMessageGroup() throws Exception{ + public void testNonExistingEmptyMessageGroup() throws Exception { GemfireMessageStore store = new GemfireMessageStore(this.cache); store.afterPropertiesSet(); MessageGroup messageGroup = store.getMessageGroup(1); @@ -68,103 +69,102 @@ public class GemfireGroupStoreTests { assertTrue(messageGroup instanceof SimpleMessageGroup); assertEquals(0, messageGroup.size()); } - + @Test - public void testMessageGroupWithAddedMessage() throws Exception{ + 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); assertEquals(1, messageGroup.size()); - + // make sure the store is properly rebuild from Gemfire store = new GemfireMessageStore(this.cache); store.afterPropertiesSet(); - + messageGroup = store.getMessageGroup(1); assertEquals(1, messageGroup.size()); } - + @Test - public void testRemoveMessageFromTheGroup() throws Exception{ + public void testRemoveMessageFromTheGroup() throws Exception { GemfireMessageStore store = new GemfireMessageStore(this.cache); store.afterPropertiesSet(); MessageGroup messageGroup = store.getMessageGroup(1); Message message = new GenericMessage("2"); - + messageGroup = store.addMessageToGroup(messageGroup.getGroupId(), new GenericMessage("1")); messageGroup = store.getMessageGroup(1); assertEquals(1, messageGroup.size()); Thread.sleep(1); //since it adds to a local region some times CREATED_DATE ends up to be the same // Unrealistic in a real scenario - + messageGroup = store.addMessageToGroup(messageGroup.getGroupId(), message); messageGroup = store.getMessageGroup(1); assertEquals(2, messageGroup.size()); Thread.sleep(1); - + messageGroup = store.addMessageToGroup(messageGroup.getGroupId(), new GenericMessage("3")); messageGroup = store.getMessageGroup(1); assertEquals(3, messageGroup.size()); - + messageGroup = store.removeMessageFromGroup(messageGroup.getGroupId(), message); messageGroup = store.getMessageGroup(1); assertEquals(2, messageGroup.size()); - + // make sure the store is properly rebuild from Gemfire store = new GemfireMessageStore(this.cache); store.afterPropertiesSet(); - + messageGroup = store.getMessageGroup(1); assertEquals(2, messageGroup.size()); } - + @Test - public void testRemoveMessageGroup() throws Exception{ + 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); assertEquals(1, messageGroup.size()); - + store.removeMessageGroup(1); MessageGroup messageGroupA = store.getMessageGroup(1); assertNotSame(messageGroup, messageGroupA); assertEquals(0, messageGroupA.getMessages().size()); assertEquals(0, messageGroupA.size()); - + // make sure the store is properly rebuild from Gemfire store = new GemfireMessageStore(this.cache); store.afterPropertiesSet(); - + messageGroup = store.getMessageGroup(1); - + assertEquals(0, messageGroup.getMessages().size()); assertEquals(0, messageGroup.size()); } - + @Test - public void testRemoveNonExistingMessageFromTheGroup() throws Exception{ + 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")); } - + @Test - public void testRemoveNonExistingMessageFromNonExistingTheGroup() throws Exception{ + public void testRemoveNonExistingMessageFromNonExistingTheGroup() throws Exception { GemfireMessageStore store = new GemfireMessageStore(this.cache); store.afterPropertiesSet(); store.removeMessageFromGroup(1, new GenericMessage("2")); } - - + @Test - public void testCompleteMessageGroup() throws Exception{ + public void testCompleteMessageGroup() throws Exception { GemfireMessageStore store = new GemfireMessageStore(this.cache); store.afterPropertiesSet(); MessageGroup messageGroup = store.getMessageGroup(1); @@ -174,9 +174,9 @@ public class GemfireGroupStoreTests { messageGroup = store.getMessageGroup(1); assertTrue(messageGroup.isComplete()); } - + @Test - public void testLastReleasedSequenceNumber() throws Exception{ + public void testLastReleasedSequenceNumber() throws Exception { GemfireMessageStore store = new GemfireMessageStore(this.cache); store.afterPropertiesSet(); MessageGroup messageGroup = store.getMessageGroup(1); @@ -186,48 +186,48 @@ public class GemfireGroupStoreTests { messageGroup = store.getMessageGroup(1); assertEquals(5, messageGroup.getLastReleasedMessageSequenceNumber()); } - + @Test - public void testMultipleInstancesOfGroupStore() throws Exception{ + 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); MessageGroup messageGroup = store2.addMessageToGroup(1, new GenericMessage("2")); - + assertEquals(2, messageGroup.getMessages().size()); - + GemfireMessageStore store3 = new GemfireMessageStore(this.cache); store3.afterPropertiesSet(); - + messageGroup = store3.removeMessageFromGroup(1, message); - + assertEquals(1, messageGroup.getMessages().size()); } - + @Test - public void testWithMessageHistory() throws Exception{ + public void testWithMessageHistory() throws Exception { GemfireMessageStore store = new GemfireMessageStore(this.cache); store.afterPropertiesSet(); - + store.getMessageGroup(1); - + Message message = new GenericMessage("Hello"); DirectChannel fooChannel = new DirectChannel(); fooChannel.setBeanName("fooChannel"); DirectChannel barChannel = new DirectChannel(); barChannel.setBeanName("barChannel"); - + message = MessageHistory.write(message, fooChannel); message = MessageHistory.write(message, barChannel); store.addMessageToGroup(1, message); - + message = store.getMessageGroup(1).getMessages().iterator().next(); - + MessageHistory messageHistory = MessageHistory.read(message); assertNotNull(messageHistory); assertEquals(2, messageHistory.size()); @@ -235,19 +235,19 @@ public class GemfireGroupStoreTests { assertEquals("fooChannel", fooChannelHistory.get("name")); assertEquals("channel", fooChannelHistory.get("type")); } - + @Test - public void testIteratorOfMessageGroups() throws Exception{ + 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")); store1.addMessageToGroup(3, new GenericMessage("3")); store2.addMessageToGroup(3, new GenericMessage("3A")); - + Iterator messageGroups = store1.iterator(); int counter = 0; while (messageGroups.hasNext()) { @@ -255,9 +255,9 @@ public class GemfireGroupStoreTests { counter++; } assertEquals(3, counter); - + store2.removeMessageGroup(3); - + messageGroups = store1.iterator(); counter = 0; while (messageGroups.hasNext()) { @@ -266,80 +266,86 @@ public class GemfireGroupStoreTests { } assertEquals(2, counter); } - + @Test @Ignore - public void testConcurrentModifications() throws Exception{ - + 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"); + + final Message message = new GenericMessage("1"); ExecutorService executor = null; - + final List failures = new ArrayList(); - + for (int i = 0; i < 100; i++) { executor = Executors.newCachedThreadPool(); - - executor.execute(new Runnable() { - public void run() { + + executor.execute(new Runnable() { + public void run() { MessageGroup group = store1.addMessageToGroup(1, message); - if (group.getMessages().size() != 1){ + if (group.getMessages().size() != 1) { failures.add("ADD"); throw new AssertionFailedError("Failed on ADD"); - } + } } }); - executor.execute(new Runnable() { + executor.execute(new Runnable() { public void run() { MessageGroup group = store2.removeMessageFromGroup(1, message); - if (group.getMessages().size() != 0){ + if (group.getMessages().size() != 0) { failures.add("REMOVE"); throw new AssertionFailedError("Failed on Remove"); - } + } } }); - + executor.shutdown(); executor.awaitTermination(10, TimeUnit.SECONDS); store2.removeMessageFromGroup(1, message); // ensures that if ADD thread executed after REMOVE, the store is empty for the next cycle } assertTrue(failures.size() == 0); } - + @Test - public void testWithAggregatorWithShutdown(){ - - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("gemfire-aggregator-config.xml", this.getClass()); + public void testWithAggregatorWithShutdown() { + + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("gemfire-aggregator-config.xml", + this.getClass()); MessageChannel input = context.getBean("inputChannel", MessageChannel.class); QueueChannel output = context.getBean("outputChannel", QueueChannel.class); - - Message m1 = MessageBuilder.withPayload("1").setSequenceNumber(1).setSequenceSize(3).setCorrelationId(1).build(); - Message m2 = MessageBuilder.withPayload("2").setSequenceNumber(2).setSequenceSize(3).setCorrelationId(1).build(); + + Message m1 = MessageBuilder.withPayload("1").setSequenceNumber(1).setSequenceSize(3).setCorrelationId(1) + .build(); + Message m2 = MessageBuilder.withPayload("2").setSequenceNumber(2).setSequenceSize(3).setCorrelationId(1) + .build(); input.send(m1); assertNull(output.receive(1000)); input.send(m2); assertNull(output.receive(1000)); - + context = new ClassPathXmlApplicationContext("gemfire-aggregator-config-a.xml", this.getClass()); MessageChannel inputA = context.getBean("inputChannel", MessageChannel.class); QueueChannel outputA = context.getBean("outputChannel", QueueChannel.class); - - Message m3 = MessageBuilder.withPayload("3").setSequenceNumber(3).setSequenceSize(3).setCorrelationId(1).build(); + + Message m3 = MessageBuilder.withPayload("3").setSequenceNumber(3).setSequenceSize(3).setCorrelationId(1) + .build(); inputA.send(m3); assertNotNull(outputA.receive(1000)); } + @Test - public void testQueue() throws Exception{ - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("gemfire-queue-config.xml", this.getClass()); - + public void testQueue() throws Exception { + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("gemfire-queue-config.xml", + this.getClass()); + QueueChannel gemfireQueue = context.getBean("gemfireQueue", QueueChannel.class); QueueChannel outputQueue = context.getBean("outputQueue", QueueChannel.class); - + for (int i = 0; i < 20; i++) { gemfireQueue.send(new GenericMessage("Hello")); Thread.sleep(1); @@ -349,18 +355,17 @@ public class GemfireGroupStoreTests { } assertNull(outputQueue.receive(1)); } - + @Before - public void init() throws Exception{ + public void init() throws Exception { CacheFactoryBean cacheFactoryBean = new CacheFactoryBean(); - cacheFactoryBean.afterPropertiesSet(); - this.cache = (Cache)cacheFactoryBean.getObject(); + this.cache = cacheFactoryBean.getObject(); } - + @After - public void cleanup(){ + public void cleanup() { this.cache.close(); Assert.isTrue(this.cache.isClosed(), "Cache did not close after close() call"); } - + } 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 e277ee2a27..6f06c7842f 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 @@ -45,18 +45,18 @@ import com.gemstone.gemfire.cache.Cache; public class GemfireMessageStoreTests { private Cache cache; - + @Test public void addAndGetMessage() throws Exception { GemfireMessageStore store = new GemfireMessageStore(this.cache); store.afterPropertiesSet(); - + Message message = MessageBuilder.withPayload("test").build(); store.addMessage(message); Message retrieved = store.getMessage(message.getHeaders().getId()); assertEquals(message, retrieved); } - + @Test public void testRegionConstructor() throws Exception { RegionFactoryBean region = new RegionFactoryBean(); @@ -66,20 +66,20 @@ public class GemfireMessageStoreTests { GemfireMessageStore store = new GemfireMessageStore(region.getObject()); store.afterPropertiesSet(); - assertSame(region.getObject(),TestUtils.getPropertyValue(store, "messageStoreRegion")); + assertSame(region.getObject(), TestUtils.getPropertyValue(store, "messageStoreRegion")); } - + @Test - public void testWithMessageHistory() throws Exception{ + public void testWithMessageHistory() throws Exception { GemfireMessageStore store = new GemfireMessageStore(this.cache); store.afterPropertiesSet(); - + Message message = new GenericMessage("Hello"); DirectChannel fooChannel = new DirectChannel(); fooChannel.setBeanName("fooChannel"); DirectChannel barChannel = new DirectChannel(); barChannel.setBeanName("barChannel"); - + message = MessageHistory.write(message, fooChannel); message = MessageHistory.write(message, barChannel); store.addMessage(message); @@ -93,14 +93,13 @@ public class GemfireMessageStoreTests { } @Before - public void init() throws Exception{ + public void init() throws Exception { CacheFactoryBean cacheFactoryBean = new CacheFactoryBean(); - cacheFactoryBean.afterPropertiesSet(); - this.cache = (Cache)cacheFactoryBean.getObject(); + this.cache = cacheFactoryBean.getObject(); } - + @After - public void cleanup(){ + public void cleanup() { this.cache.close(); Assert.isTrue(this.cache.isClosed(), "Cache did not close after close() call"); }