diff --git a/build.gradle b/build.gradle index 212e8defc0..03d5433caf 100644 --- a/build.gradle +++ b/build.gradle @@ -264,6 +264,7 @@ project('spring-integration-file') { compile "commons-io:commons-io:$commonsIoVersion" testCompile project(":spring-integration-redis") testCompile project(":spring-integration-redis").sourceSets.test.output + testCompile project(":spring-integration-gemfire") testCompile "com.lambdaworks:lettuce:$lettuceVersion" } } diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/filters/PersistentAcceptOnceFileListFilterRedisTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/filters/PersistentAcceptOnceFileListFilterExternalStoreTests.java similarity index 72% rename from spring-integration-file/src/test/java/org/springframework/integration/file/filters/PersistentAcceptOnceFileListFilterRedisTests.java rename to spring-integration-file/src/test/java/org/springframework/integration/file/filters/PersistentAcceptOnceFileListFilterExternalStoreTests.java index ce600890c3..7668b7372e 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/filters/PersistentAcceptOnceFileListFilterRedisTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/filters/PersistentAcceptOnceFileListFilterExternalStoreTests.java @@ -26,49 +26,64 @@ import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; -import org.junit.After; -import org.junit.Before; import org.junit.Test; +import org.mockito.Mockito; +import org.mockito.invocation.InvocationOnMock; +import org.mockito.stubbing.Answer; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.data.redis.serializer.StringRedisSerializer; +import org.springframework.integration.gemfire.metadata.GemfireMetadataStore; +import org.springframework.integration.metadata.ConcurrentMetadataStore; import org.springframework.integration.redis.metadata.RedisMetadataStore; import org.springframework.integration.redis.rules.RedisAvailable; import org.springframework.integration.redis.rules.RedisAvailableTests; +import com.gemstone.gemfire.cache.CacheFactory; + /** * @author Gary Russell + * @author Artem Bilan * @since 4.0 * */ -public class PersistentAcceptOnceFileListFilterRedisTests extends RedisAvailableTests { +public class PersistentAcceptOnceFileListFilterExternalStoreTests extends RedisAvailableTests { - @Before - @After - public void setupShutDown() { - RedisTemplate template = this.createTemplate(); - template.delete("persistentAcceptOnceFileListFilterRedisTests"); - } - - private RedisTemplate createTemplate() { + @Test + @RedisAvailable + public void testFileSystemWithRedisMetadataStore() throws Exception { RedisTemplate template = new RedisTemplate(); template.setConnectionFactory(this.getConnectionFactoryForTest()); template.setKeySerializer(new StringRedisSerializer()); template.afterPropertiesSet(); - return template; + template.delete("persistentAcceptOnceFileListFilterRedisTests"); + + try { + this.testFileSystem(new RedisMetadataStore(this.getConnectionFactoryForTest(), + "persistentAcceptOnceFileListFilterRedisTests")); + } + finally { + template.delete("persistentAcceptOnceFileListFilterRedisTests"); + } } @Test - @RedisAvailable - public void testFileSystem() throws Exception { + public void testFileSystemWithGemfireMetadataStore() throws Exception { + this.testFileSystem(new GemfireMetadataStore(new CacheFactory().create())); + } + + private void testFileSystem(ConcurrentMetadataStore store) throws Exception { final AtomicBoolean suspend = new AtomicBoolean(); + final CountDownLatch latch1 = new CountDownLatch(1); final CountDownLatch latch2 = new CountDownLatch(1); - RedisMetadataStore store = new RedisMetadataStore(this.getConnectionFactoryForTest(), - "persistentAcceptOnceFileListFilterRedisTests") { + + store = Mockito.spy(store); + + Mockito.doAnswer(new Answer() { @Override - public boolean replace(String key, String oldValue, String newValue) { + public Object answer(InvocationOnMock invocation) throws Throwable { if (suspend.get()) { latch2.countDown(); try { @@ -78,12 +93,12 @@ public class PersistentAcceptOnceFileListFilterRedisTests extends RedisAvailable Thread.currentThread().interrupt(); } } - return super.replace(key, oldValue, newValue); + return invocation.callRealMethod(); } + }).when(store).replace(Mockito.anyString(), Mockito.anyString(), Mockito.anyString()); - }; final FileSystemPersistentAcceptOnceFileListFilter filter = - new FileSystemPersistentAcceptOnceFileListFilter(store,"foo:"); + new FileSystemPersistentAcceptOnceFileListFilter(store, "foo:"); final File file = File.createTempFile("foo", ".txt"); assertEquals(1, filter.filterFiles(new File[] {file}).size()); String ts = store.get("foo:" + file.getAbsolutePath()); diff --git a/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/metadata/GemfireMetadataStore.java b/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/metadata/GemfireMetadataStore.java new file mode 100644 index 0000000000..2894d9ee1b --- /dev/null +++ b/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/metadata/GemfireMetadataStore.java @@ -0,0 +1,85 @@ +/* + * Copyright 2014 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.metadata; + +import org.springframework.integration.metadata.ConcurrentMetadataStore; +import org.springframework.util.Assert; + +import com.gemstone.gemfire.cache.Cache; +import com.gemstone.gemfire.cache.Region; +import com.gemstone.gemfire.cache.Scope; + +/** + * Gemfire implementation of {@link ConcurrentMetadataStore}. + * Use this {@link org.springframework.integration.metadata.MetadataStore} + * to achieve meta-data persistence shared across application instances and + * restarts. + * + * @author Artem Bilan + * @since 4.0 + */ +public class GemfireMetadataStore implements ConcurrentMetadataStore { + + public static final String KEY = "MetaData"; + + private final Region region; + + public GemfireMetadataStore(Cache cache) { + Assert.notNull(cache, "'cache' must not be null"); + this.region = cache.createRegionFactory().setScope(Scope.LOCAL).create(KEY); + } + + public GemfireMetadataStore(Region region) { + Assert.notNull(region, "'region' must not be null"); + this.region = region; + } + + @Override + public void put(String key, String value) { + Assert.notNull(key, "'key' must not be null."); + Assert.notNull(value, "'value' must not be null."); + this.region.put(key, value); + } + + @Override + public String putIfAbsent(String key, String value) { + Assert.notNull(key, "'key' must not be null."); + Assert.notNull(value, "'value' must not be null."); + return this.region.putIfAbsent(key, value); + } + + @Override + public boolean replace(String key, String oldValue, String newValue) { + Assert.notNull(key, "'key' must not be null."); + Assert.notNull(oldValue, "'oldValue' must not be null."); + Assert.notNull(newValue, "'newValue' must not be null."); + return this.region.replace(key, oldValue, newValue); + } + + @Override + public String get(String key) { + Assert.notNull(key, "'key' must not be null."); + return this.region.get(key); + } + + @Override + public String remove(String key) { + Assert.notNull(key, "'key' must not be null."); + return this.region.remove(key); + } + +} diff --git a/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/metadata/package-info.java b/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/metadata/package-info.java new file mode 100644 index 0000000000..5454827492 --- /dev/null +++ b/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/metadata/package-info.java @@ -0,0 +1,4 @@ +/** + * Provides classes for the Gemfire MetadataStore. + */ +package org.springframework.integration.gemfire.metadata; diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/metadata/GemfireMetadataStoreTests.java b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/metadata/GemfireMetadataStoreTests.java new file mode 100644 index 0000000000..9dd452dfb7 --- /dev/null +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/metadata/GemfireMetadataStoreTests.java @@ -0,0 +1,165 @@ +/* + * Copyright 2014 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.metadata; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.fail; + +import org.junit.After; +import org.junit.AfterClass; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; + +import org.springframework.data.gemfire.GemfireTemplate; +import org.springframework.integration.gemfire.metadata.GemfireMetadataStore; +import org.springframework.integration.metadata.ConcurrentMetadataStore; +import org.springframework.util.Assert; + +import com.gemstone.gemfire.cache.Cache; +import com.gemstone.gemfire.cache.CacheFactory; +import com.gemstone.gemfire.cache.Region; + +/** + * @author Artem Bilan + * @since 4.0 + * + */ +public class GemfireMetadataStoreTests { + + public static Cache cache; + + public static ConcurrentMetadataStore metadataStore; + + @BeforeClass + public static void startUp() throws Exception { + cache = new CacheFactory().create(); + metadataStore = new GemfireMetadataStore(cache); + } + + @AfterClass + public static void cleanUp() { + cache.close(); + Assert.isTrue(cache.isClosed(), "Cache did not close after close() call"); + } + + @Before + @After + public void setup() { + getRegion().clear(); + } + + @Test + public void testGetNonExistingKeyValue() { + String retrievedValue = metadataStore.get("does-not-exist"); + assertNull(retrievedValue); + } + + @Test + public void testPersistKeyValue() { + metadataStore.put("GemfireMetadataStoreTests-Spring", "Integration"); + + GemfireTemplate gemfireTemplate = new GemfireTemplate(getRegion()); + + assertEquals("Integration", gemfireTemplate.get("GemfireMetadataStoreTests-Spring")); + } + + @Test + public void testGetValueFromMetadataStore() { + metadataStore.put("GemfireMetadataStoreTests-GetValue", "Hello Gemfire"); + + String retrievedValue = metadataStore.get("GemfireMetadataStoreTests-GetValue"); + assertEquals("Hello Gemfire", retrievedValue); + } + + @Test + public void testPersistEmptyStringToMetadataStore() { + metadataStore.put("GemfireMetadataStoreTests-PersistEmpty", ""); + + String retrievedValue = metadataStore.get("GemfireMetadataStoreTests-PersistEmpty"); + assertEquals("", retrievedValue); + } + + @Test + public void testPersistNullStringToMetadataStore() { + try { + metadataStore.put("GemfireMetadataStoreTests-PersistEmpty", null); + fail("Expected an IllegalArgumentException to be thrown."); + } + catch (IllegalArgumentException e) { + assertEquals("'value' must not be null.", e.getMessage()); + } + } + + @Test + public void testPersistWithEmptyKeyToMetadataStore() { + metadataStore.put("", "PersistWithEmptyKey"); + + String retrievedValue = metadataStore.get(""); + assertEquals("PersistWithEmptyKey", retrievedValue); + } + + @Test + public void testPersistWithNullKeyToMetadataStore() { + try { + metadataStore.put(null, "something"); + fail("Expected an IllegalArgumentException to be thrown."); + + } + catch (IllegalArgumentException e) { + assertEquals("'key' must not be null.", e.getMessage()); + } + } + + @Test + public void testGetValueWithNullKeyFromMetadataStore() { + try { + metadataStore.get(null); + } + catch (IllegalArgumentException e) { + assertEquals("'key' must not be null.", e.getMessage()); + return; + } + + fail("Expected an IllegalArgumentException to be thrown."); + } + + @Test + public void testRemoveFromMetadataStore() { + String testKey = "GemfireMetadataStoreTests-Remove"; + String testValue = "Integration"; + + metadataStore.put(testKey, testValue); + + assertEquals(testValue, metadataStore.remove(testKey)); + assertNull(metadataStore.remove(testKey)); + } + + @Test + public void testPersistKeyValueIfAbsent() { + metadataStore.putIfAbsent("GemfireMetadataStoreTests-Spring", "Integration"); + + GemfireTemplate gemfireTemplate = new GemfireTemplate(getRegion()); + + assertEquals("Integration", gemfireTemplate.get("GemfireMetadataStoreTests-Spring")); + } + + private static Region getRegion() { + return cache.getRegion(GemfireMetadataStore.KEY); + } + +} diff --git a/src/reference/docbook/gemfire.xml b/src/reference/docbook/gemfire.xml index f5abd70875..5594c29c48 100644 --- a/src/reference/docbook/gemfire.xml +++ b/src/reference/docbook/gemfire.xml @@ -200,4 +200,39 @@ Note the pool element is configured with the address of a c +
+ Gemfire Metadata Store + + As of Spring Integration 4.0, a new Gemfire-based + MetadataStore + () implementation is available. The GemfireMetadataStore + can be used to maintain metadata state + across application restarts. This new MetadataStore + implementation can be used with adapters such as: + + + + + + + + + + In order to instruct these adapters to use the new GemfireMetadataStore, + simply declare a Spring bean using the bean name metadataStore. + The Twitter Inbound Channel Adapter and the + Feed Inbound Channel Adapter will both automatically + pick up and use the declared GemfireMetadataStore. + + + + The GemfireMetadataStore also implements + ConcurrentMetadataStore, allowing it to be reliably shared across + multiple application instances where only one instance will be allowed to store or modify a key's value. + These methods give various levels of concurrency guarantees based on the scope and data policy of the region. + They are implemented in the peer cache and client/server cache but are disallowed in peer Regions having + NORMAL or EMPTY data policies. + + +
diff --git a/src/reference/docbook/meta-data-store.xml b/src/reference/docbook/meta-data-store.xml index 54eb63fab5..66cd1a2afa 100644 --- a/src/reference/docbook/meta-data-store.xml +++ b/src/reference/docbook/meta-data-store.xml @@ -28,6 +28,7 @@ PropertiesPersistingMetadataStore + The PropertiesPersistingMetadataStore is backed by a properties file and a @@ -49,7 +50,7 @@
Idempotent Receiver - The Metadata Store is useful for implementating the + The Metadata Store is useful for implementing the EIP Idempotent Receiver pattern, when there is need to filter an incoming Message if it has already been processed, and just discard it or perform some other logic on discarding. The following configuration is an example of how to do this: diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml index 7bae732505..c2959304ec 100644 --- a/src/reference/docbook/whats-new.xml +++ b/src/reference/docbook/whats-new.xml @@ -182,6 +182,17 @@ For more information, see .
+
+ Gemfire Metadata Store + + The GemfireMetadataStore is provided, allowing it to be used, for example, + in a AbstractPersistentAcceptOnceFileListFilter + implementation in a multiple application instance/server environment. + For more information, see , + , and + . + +