INT-3367: Add GemfireMetadataStore

JIRA: https://jira.spring.io/browse/INT-3367

INT-3367 Polishing

Rename test class; move to metadata package; doc polishing.

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"
 	}
 }

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<String, ?> template = this.createTemplate();
-		template.delete("persistentAcceptOnceFileListFilterRedisTests");
-	}
-
-	private RedisTemplate<String, ?> createTemplate() {
+	@Test
+	@RedisAvailable
+	public void testFileSystemWithRedisMetadataStore() throws Exception {
 		RedisTemplate<String, ?> template = new RedisTemplate<String, Object>();
 		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<Object>() {
 
 			@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());

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<String, String> region;
+
+	public GemfireMetadataStore(Cache cache) {
+		Assert.notNull(cache, "'cache' must not be null");
+		this.region = cache.<String, String>createRegionFactory().setScope(Scope.LOCAL).create(KEY);
+	}
+
+	public GemfireMetadataStore(Region<String, String> 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);
+	}
+
+}

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;

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<Object, Object> getRegion() {
+		return cache.getRegion(GemfireMetadataStore.KEY);
+	}
+
+}

src/reference/docbook/gemfire.xml

@@ -200,4 +200,39 @@ Note the <emphasis>pool</emphasis> element is configured with the address of a c
 			</note>
 		</para>
 	</section>
+	<section id="gemfire-metadata-store">
+		<title>Gemfire Metadata Store</title>
+		<para>
+			As of <emphasis>Spring Integration 4.0</emphasis>, a new Gemfire-based
+			<interfacename>MetadataStore</interfacename>
+			(<xref linkend="metadata-store"/>) implementation is available. The <classname>GemfireMetadataStore</classname>
+			can be used to maintain metadata state
+			across application restarts. This new <interfacename>MetadataStore</interfacename>
+			implementation can be used with adapters such as:
+		</para>
+		<itemizedlist>
+			<listitem><xref linkend="twitter-inbound"/></listitem>
+			<listitem><xref linkend="feed-inbound-channel-adapter"/></listitem>
+			<listitem><xref linkend="file-reading"/></listitem>
+			<listitem><xref linkend="ftp-inbound"/></listitem>
+			<listitem><xref linkend="sftp-inbound"/></listitem>
+		</itemizedlist>
+		<para>
+			In order to instruct these adapters to use the new <classname>GemfireMetadataStore</classname>,
+			simply declare a Spring bean using the bean name <emphasis role="bold">metadataStore</emphasis>.
+			The <emphasis>Twitter Inbound Channel Adapter</emphasis> and the
+			<emphasis>Feed Inbound Channel Adapter</emphasis> will both automatically
+			pick up and use the declared <classname>GemfireMetadataStore</classname>.
+		</para>
+		<note>
+			<para>
+				The <classname>GemfireMetadataStore</classname> also implements
+				<interfacename>ConcurrentMetadataStore</interfacename>, 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.
+			</para>
+		</note>
+	</section>
 </chapter>

src/reference/docbook/meta-data-store.xml

@@ -28,6 +28,7 @@
 	<itemizedlist>
 		<listitem>PropertiesPersistingMetadataStore</listitem>
 		<listitem><xref linkend="redis-metadata-store"/></listitem>
+		<listitem><xref linkend="gemfire-metadata-store"/></listitem>
 	</itemizedlist>
 	<para>
 		The <classname>PropertiesPersistingMetadataStore</classname> is backed by a properties file and a
@@ -49,7 +50,7 @@
 	<section id="idempotent-receiver">
 		<title>Idempotent Receiver</title>
 		<para>
-			The <emphasis>Metadata Store</emphasis> is useful for implementating the
+			The <emphasis>Metadata Store</emphasis> is useful for implementing the
 			EIP <ulink url="http://eaipatterns.com/IdempotentReceiver.html">Idempotent Receiver</ulink> pattern, when
 			there is need to <emphasis>filter</emphasis> 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:

src/reference/docbook/whats-new.xml

@@ -182,6 +182,17 @@
 				For more information, see <xref linkend="twitter-sog"/>.
 			</para>
 		</section>
+		<section id="4.0-gemfire-metadata">
+			<title>Gemfire Metadata Store</title>
+			<para>
+				The <classname>GemfireMetadataStore</classname> is provided, allowing it to be used, for example,
+				in a <classname>AbstractPersistentAcceptOnceFileListFilter</classname>
+				implementation in a multiple application instance/server environment.
+				For more information, see <xref linkend="metadata-store"/>,
+				<xref linkend="file-reading"/>, <xref linkend="ftp-inbound"/> and
+				<xref linkend="sftp-inbound"/>.
+			</para>
+		</section>
 	</section>
 
 	<section id="4.0-general">