Initial implementation of HazelcastMetadataStore

HazelcastMetadataStore improvements
 - Adding ListenableMetadataStore capabilities
 - Extra tests for listeners

Requested changes from review

* Make `HazelcastMetadataStore.MapListener` as static class

spring-integration-hazelcast/README.md

@@ -497,4 +497,25 @@ public MessageGroupStore messageStore() {
 
 By default the `SPRING_INTEGRATION_MESSAGE_STORE` `IMap` is used to store messages and groups key/value manner.
 Any custom `IMap` can be provided to the `HazelcastMessageStore`.
-See [Spring Integration User Guide](http://docs.spring.io/spring-integration/reference/html/system-management-chapter.html#message-store) for more information about `MessageStore`.
+See [Spring Integration User Guide](http://docs.spring.io/spring-integration/reference/html/system-management-chapter.html#message-store) for more information about `MessageStore`.
+
+## HAZELCAST METADATA STORE
+An implementation of a [MetadataStore](http://docs.spring.io/spring-integration/reference/html/system-management-chapter.html#metadata-store) is available using a backing Hazelcast `IMap`
+
+You can provide your own implementation of an `IMap` or rely on the default map created with name `SPRING_INTEGRATION_METADATA_STORE`.
+
+```java
+@Bean
+public HazelcastInstance hazelcastInstance() {
+    return Hazelcast.newHazelcastInstance();
+}
+
+@Bean
+public MetadataStore metadataStore() {
+    return new HazelcastMetadataStore(hazelcastInstance());
+}
+```
+
+The `HazelcastMetadataStore` implements `ListenableMetadataStore` which allows you to register your own listeners of type `MetadataStoreListener` to listen for events via `addListener(MetadataStoreListener callback)`
+
+See [Spring Integration User Guide](http://docs.spring.io/spring-integration/reference/html/system-management-chapter.html#metadatastore-listener) for more information about the `MetadataStoreListener` interface. 

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/metadata/HazelcastMetadataStore.java

@@ -0,0 +1,140 @@
+/*
+ * Copyright 2017 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.hazelcast.metadata;
+
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.integration.metadata.ListenableMetadataStore;
+import org.springframework.integration.metadata.MetadataStoreListener;
+import org.springframework.util.Assert;
+
+import com.hazelcast.core.EntryEvent;
+import com.hazelcast.core.HazelcastInstance;
+import com.hazelcast.core.IMap;
+import com.hazelcast.map.listener.EntryAddedListener;
+import com.hazelcast.map.listener.EntryRemovedListener;
+import com.hazelcast.map.listener.EntryUpdatedListener;
+
+/**
+ * The Hazelcast {@link IMap}-based {@link ListenableMetadataStore} implementation.
+ *
+ * @author Vinicius Carvalho
+ * @author Artem Bilan
+ */
+public class HazelcastMetadataStore implements ListenableMetadataStore, InitializingBean {
+
+	private static final String METADATA_STORE_MAP_NAME = "SPRING_INTEGRATION_METADATA_STORE";
+
+	private final IMap<String, String> map;
+
+	private final List<MetadataStoreListener> listeners = new CopyOnWriteArrayList<MetadataStoreListener>();
+
+	public HazelcastMetadataStore(HazelcastInstance hazelcastInstance) {
+		Assert.notNull(hazelcastInstance, "Hazelcast instance can't be null");
+		this.map = hazelcastInstance.getMap(METADATA_STORE_MAP_NAME);
+	}
+
+	public HazelcastMetadataStore(IMap<String, String> map) {
+		Assert.notNull(map, "IMap reference can not be null");
+		this.map = map;
+	}
+
+	@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.map.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.map.replace(key, oldValue, newValue);
+	}
+
+	@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.map.put(key, value);
+	}
+
+	@Override
+	public String get(String key) {
+		Assert.notNull(key, "'key' must not be null.");
+		return this.map.get(key);
+	}
+
+	@Override
+	public String remove(String key) {
+		Assert.notNull(key, "'key' must not be null.");
+		return this.map.remove(key);
+	}
+
+	@Override
+	public void addListener(MetadataStoreListener callback) {
+		Assert.notNull(callback, "callback object can not be null");
+		this.listeners.add(callback);
+	}
+
+	@Override
+	public void removeListener(MetadataStoreListener callback) {
+		this.listeners.remove(callback);
+	}
+
+	@Override
+	public void afterPropertiesSet() throws Exception {
+		this.map.addEntryListener(new MapListener(this.listeners), true);
+	}
+
+	private static class MapListener implements EntryAddedListener<String, String>,
+			EntryRemovedListener<String, String>, EntryUpdatedListener<String, String> {
+
+		private final List<MetadataStoreListener> listeners;
+
+		MapListener(List<MetadataStoreListener> listeners) {
+			this.listeners = listeners;
+		}
+
+		@Override
+		public void entryAdded(EntryEvent<String, String> event) {
+			for (MetadataStoreListener listener : this.listeners) {
+				listener.onAdd(event.getKey(), event.getValue());
+			}
+		}
+
+		@Override
+		public void entryRemoved(EntryEvent<String, String> event) {
+			for (MetadataStoreListener listener : this.listeners) {
+				listener.onRemove(event.getKey(), event.getOldValue());
+			}
+		}
+
+		@Override
+		public void entryUpdated(EntryEvent<String, String> event) {
+			for (MetadataStoreListener listener : this.listeners) {
+				listener.onUpdate(event.getKey(), event.getValue());
+			}
+		}
+
+	}
+
+}

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/metadata/package-info.java

@@ -0,0 +1,4 @@
+/**
+ * Provides the Metadata Store support classes.
+ */
+package org.springframework.integration.hazelcast.metadata;

spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/metadata/HazelcastMetadataStoreTests.java

@@ -0,0 +1,182 @@
+/*
+ * Copyright 2017 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.hazelcast.metadata;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.springframework.integration.metadata.MetadataStoreListener;
+
+import com.hazelcast.core.Hazelcast;
+import com.hazelcast.core.HazelcastInstance;
+import com.hazelcast.core.IMap;
+
+/**
+ * @author Vinicius Carvalho
+ */
+public class HazelcastMetadataStoreTests {
+
+	private static HazelcastInstance instance;
+
+	private static IMap<String, String> map;
+
+	HazelcastMetadataStore metadataStore;
+
+	@BeforeClass
+	public static void init() {
+		instance = Hazelcast.newHazelcastInstance();
+		map = instance.getMap("customTestsMetadataStore");
+	}
+
+	@AfterClass
+	public static void destroy() throws Exception {
+		instance.shutdown();
+	}
+
+	@Before
+	public void setup() throws Exception {
+		this.metadataStore = new HazelcastMetadataStore(map);
+		this.metadataStore.afterPropertiesSet();
+	}
+
+	@After
+	public void clean() {
+		map.clear();
+	}
+
+	@Test
+	public void testGetNonExistingKeyValue() {
+		String retrievedValue = this.metadataStore.get("does-not-exist");
+		assertNull(retrievedValue);
+	}
+
+	@Test
+	public void testPersistKeyValue() {
+		this.metadataStore.put("HazelcastMetadataStoreTests-Spring", "Integration");
+		assertEquals("Integration", map.get("HazelcastMetadataStoreTests-Spring"));
+	}
+
+	@Test
+	public void testGetValueFromMetadataStore() {
+		this.metadataStore.put("HazelcastMetadataStoreTests-GetValue", "Hello Hazelcast");
+		String retrievedValue = this.metadataStore
+				.get("HazelcastMetadataStoreTests-GetValue");
+		assertEquals("Hello Hazelcast", retrievedValue);
+	}
+
+	@Test
+	public void testPersistEmptyStringToMetadataStore() {
+		this.metadataStore.put("HazelcastMetadataStoreTests-PersistEmpty", "");
+
+		String retrievedValue = this.metadataStore
+				.get("HazelcastMetadataStoreTests-PersistEmpty");
+		assertEquals("", retrievedValue);
+	}
+
+	@Test
+	public void testPersistNullStringToMetadataStore() {
+		try {
+			this.metadataStore.put("HazelcastMetadataStoreTests-PersistEmpty", null);
+			fail("Expected an IllegalArgumentException to be thrown.");
+		}
+		catch (IllegalArgumentException e) {
+			assertEquals("'value' must not be null.", e.getMessage());
+		}
+	}
+
+	@Test
+	public void testPersistWithEmptyKeyToMetadataStore() {
+		this.metadataStore.put("", "PersistWithEmptyKey");
+
+		String retrievedValue = this.metadataStore.get("");
+		assertEquals("PersistWithEmptyKey", retrievedValue);
+	}
+
+	@Test
+	public void testPersistWithNullKeyToMetadataStore() {
+		try {
+			this.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 {
+			this.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 = "HazelcastMetadataStoreTests-Remove";
+		String testValue = "Integration";
+
+		this.metadataStore.put(testKey, testValue);
+
+		assertEquals(testValue, this.metadataStore.remove(testKey));
+		assertNull(this.metadataStore.remove(testKey));
+	}
+
+	@Test
+	public void testPersistKeyValueIfAbsent() {
+		this.metadataStore.putIfAbsent("HazelcastMetadataStoreTests-Spring",
+				"Integration");
+		assertEquals("Integration", map.get("HazelcastMetadataStoreTests-Spring"));
+	}
+
+	@Test
+	public void testReplaceValue() {
+		this.metadataStore.put("key", "old");
+		assertEquals("old", map.get("key"));
+		this.metadataStore.replace("key", "old", "new");
+		assertEquals("new", map.get("key"));
+	}
+
+	@Test
+	public void testListener() {
+		MetadataStoreListener listener = mock(MetadataStoreListener.class);
+		this.metadataStore.addListener(listener);
+
+		this.metadataStore.put("foo", "bar");
+		this.metadataStore.replace("foo", "bar", "baz");
+		this.metadataStore.remove("foo");
+		verify(listener).onAdd("foo", "bar");
+		verify(listener).onUpdate("foo", "baz");
+		verify(listener).onRemove("foo", "baz");
+	}
+
+}