Add Hazelcast MessageStore implementation

- Hazelcast implementation of the `MessageStore`
- Test case uses the same tests as gemfire plus a couple of more scenarios
* Some polishing and Docs

spring-integration-hazelcast/README.md

@@ -477,5 +477,24 @@ public LeaderInitiator initiator() {
 ```
 
 Then when a node is elected leader it will send `OnGrantedEvent` to all application listeners. See
-the [Spring Integration User Guide](http://docs.spring.io/spring-integration/reference/htmlsingle/#endpoint-roles)
+the [Spring Integration User Guide](http://docs.spring.io/spring-integration/reference/html/#endpoint-roles)
 for more information on how to use those events to control messaging endpoints.
+
+## HAZELCAST MESSAGE STORE
+
+For distributed messaging state management, for example for persistent `QueueChannel` or tracking `Aggregator` message groups, the `HazelcastMessageStore` implementation is provided:
+```java
+@Bean
+public HazelcastInstance hazelcastInstance() {
+    return Hazelcast.newHazelcastInstance();
+}
+
+@Bean
+public MessageGroupStore messageStore() {
+    return new HazelcastMessageStore(hazelcastInstance());
+}
+```
+
+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`.

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

@@ -0,0 +1,4 @@
+/**
+ * Provides the Leader Initiator support classes.
+ */
+package org.springframework.integration.hazelcast.leader;

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/store/HazelcastMessageStore.java

@@ -0,0 +1,73 @@
+/*
+ * 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.store;
+
+import java.util.Collection;
+
+import org.springframework.integration.store.AbstractKeyValueMessageStore;
+import org.springframework.util.Assert;
+
+import com.hazelcast.core.HazelcastInstance;
+import com.hazelcast.core.IMap;
+import com.hazelcast.query.SqlPredicate;
+
+/**
+ * The Hazelcast {@link IMap}-based {@link AbstractKeyValueMessageStore} implementation.
+ *
+ * @author Vinicius Carvalho
+ * @author Artem Bilan
+ */
+public class HazelcastMessageStore extends AbstractKeyValueMessageStore {
+
+	private static final String MESSAGE_STORE_MAP_NAME = "SPRING_INTEGRATION_MESSAGE_STORE";
+
+	private final IMap<Object, Object> map;
+
+	public HazelcastMessageStore(HazelcastInstance hazelcastInstance) {
+		Assert.notNull(hazelcastInstance, "Hazelcast instance can't be null");
+		this.map = hazelcastInstance.getMap(MESSAGE_STORE_MAP_NAME);
+	}
+
+	public HazelcastMessageStore(IMap<Object, Object> map) {
+		Assert.notNull(map, "IMap reference can not be null");
+		this.map = map;
+	}
+
+	@Override
+	protected Object doRetrieve(Object id) {
+		return this.map.get(id);
+	}
+
+	@Override
+	protected void doStore(Object id, Object objectToStore) {
+		this.map.put(id, objectToStore);
+	}
+
+	@Override
+	protected Object doRemove(Object id) {
+		return this.map.remove(id);
+	}
+
+	@Override
+	protected Collection<?> doListKeys(String keyPattern) {
+		Assert.hasText(keyPattern, "'keyPattern' must not be empty");
+		keyPattern = keyPattern.replaceAll("\\*", "%");
+		SqlPredicate sqlPredicate = new SqlPredicate("__key like " + keyPattern);
+		return this.map.values(sqlPredicate);
+	}
+
+}

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

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

spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/store/HazelcastMessageStoreTests.java

@@ -0,0 +1,138 @@
+/*
+ * 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.store;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNotSame;
+import static org.junit.Assert.assertSame;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.history.MessageHistory;
+import org.springframework.integration.store.MessageGroup;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.integration.test.util.TestUtils;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.GenericMessage;
+
+import com.hazelcast.core.Hazelcast;
+import com.hazelcast.core.HazelcastInstance;
+import com.hazelcast.core.IMap;
+
+/**
+ * @author Vinicius Carvalho
+ * @author Artem Bilan
+ */
+public class HazelcastMessageStoreTests {
+
+	private static HazelcastMessageStore store;
+
+	private static HazelcastInstance instance;
+
+	private static IMap<Object, Object> map;
+
+	@BeforeClass
+	public static void init() throws Exception {
+		instance = Hazelcast.newHazelcastInstance();
+		map = instance.getMap("customTestsMessageStore");
+		store = new HazelcastMessageStore(map);
+	}
+
+	@AfterClass
+	public static void destroy() throws Exception {
+		instance.shutdown();
+	}
+
+	@Before
+	public void clean() throws Exception {
+		map.clear();
+	}
+
+	@Test
+	public void testWithMessageHistory() throws Exception {
+
+		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);
+		message = store.getMessage(message.getHeaders().getId());
+		MessageHistory messageHistory = MessageHistory.read(message);
+		assertNotNull(messageHistory);
+		assertEquals(2, messageHistory.size());
+		Properties fooChannelHistory = messageHistory.get(0);
+		assertEquals("fooChannel", fooChannelHistory.get("name"));
+		assertEquals("channel", fooChannelHistory.get("type"));
+
+	}
+
+	@Test
+	public void testAddAndRemoveMessagesFromMessageGroup() throws Exception {
+		String groupId = "X";
+		List<Message<?>> messages = new ArrayList<>();
+		for (int i = 0; i < 25; i++) {
+			Message<String> message = MessageBuilder.withPayload("foo").setCorrelationId(groupId).build();
+			store.addMessagesToGroup(groupId, message);
+			messages.add(message);
+		}
+		MessageGroup group = store.getMessageGroup(groupId);
+		assertEquals(25, group.size());
+		store.removeMessagesFromGroup(groupId, messages);
+		group = store.getMessageGroup(groupId);
+		assertEquals(0, group.size());
+	}
+
+	@Test
+	public void addAndGetMessage() throws Exception {
+
+		Message<?> message = MessageBuilder.withPayload("test").build();
+		store.addMessage(message);
+		Message<?> retrieved = store.getMessage(message.getHeaders().getId());
+		assertEquals(message, retrieved);
+	}
+
+	@Test
+	public void customMap() throws Exception {
+		assertSame(map, TestUtils.getPropertyValue(store, "map"));
+		HazelcastMessageStore store2 = new HazelcastMessageStore(instance);
+		assertNotSame(map, TestUtils.getPropertyValue(store2, "map"));
+	}
+
+	@Test
+	public void messageStoreSize() throws Exception {
+		Message<?> message1 = MessageBuilder.withPayload("test").build();
+		Message<?> message2 = MessageBuilder.withPayload("test").build();
+		store.addMessage(message1);
+		store.addMessage(message2);
+		long size = store.getMessageCount();
+		assertEquals(2, size);
+	}
+
+}