From 894a60b8bb4da626023f30c2815419d3afc02097 Mon Sep 17 00:00:00 2001 From: Vinicius Carvalho Date: Thu, 6 Jul 2017 12:23:19 -0400 Subject: [PATCH] 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 | 21 ++- .../hazelcast/leader/package-info.java | 4 + .../store/HazelcastMessageStore.java | 73 +++++++++ .../hazelcast/store/package-info.java | 4 + .../store/HazelcastMessageStoreTests.java | 138 ++++++++++++++++++ 5 files changed, 239 insertions(+), 1 deletion(-) create mode 100644 spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/leader/package-info.java create mode 100644 spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/store/HazelcastMessageStore.java create mode 100644 spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/store/package-info.java create mode 100644 spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/store/HazelcastMessageStoreTests.java diff --git a/spring-integration-hazelcast/README.md b/spring-integration-hazelcast/README.md index dc9795b..de60453 100644 --- a/spring-integration-hazelcast/README.md +++ b/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`. \ No newline at end of file diff --git a/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/leader/package-info.java b/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/leader/package-info.java new file mode 100644 index 0000000..edd7be0 --- /dev/null +++ b/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; diff --git a/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/store/HazelcastMessageStore.java b/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/store/HazelcastMessageStore.java new file mode 100644 index 0000000..e2612e2 --- /dev/null +++ b/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 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 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); + } + +} diff --git a/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/store/package-info.java b/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/store/package-info.java new file mode 100644 index 0000000..4657766 --- /dev/null +++ b/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; diff --git a/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/store/HazelcastMessageStoreTests.java b/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/store/HazelcastMessageStoreTests.java new file mode 100644 index 0000000..8edc787 --- /dev/null +++ b/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 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> messages = new ArrayList<>(); + for (int i = 0; i < 25; i++) { + Message 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); + } + +}