diff --git a/spring-integration-hazelcast/README.md b/spring-integration-hazelcast/README.md index 0f443ab..16a34f4 100644 --- a/spring-integration-hazelcast/README.md +++ b/spring-integration-hazelcast/README.md @@ -37,7 +37,7 @@ Basically, Hazelcast Event-Driven Inbound Channel Adapter requires following att 3. Supported cache event types for IList, ISet and IQueue : ADDED, REMOVED. 4. There is no need to cache event type definition for ITopic. -* **cache-listening-policy :** Specifies cache listening policy as SINGLE or ALL. It is optional attribute and its default value is SINGLE. Each Hazelcast CQ inbound channel adapter listening same cache object with same cache-events attribute, can receive a single event message or all event messages. If it is ALL, all Hazelcast CQ inbound channel adapters listening same cache object with same cache-events attribute, will receive same event messages. If it is SINGLE, they will receive unique event messages. +* **cache-listening-policy :** Specifies cache listening policy as SINGLE or ALL. It is optional attribute and its default value is SINGLE. Each Hazelcast inbound channel adapter listening same cache object with same cache-events attribute, can receive a single event message or all event messages. If it is ALL, all Hazelcast inbound channel adapters listening same cache object with same cache-events attribute, will receive same event messages. If it is SINGLE, they will receive unique event messages. Sample namespace and schemaLocation definitions are as follows : ``` @@ -247,59 +247,28 @@ Sample definition is as follows : ## HAZELCAST OUTBOUND CHANNEL ADAPTER -Hazelcast Outbound Channel Adapter listens its defined channel and writes incoming messages to related distributed cache. It expects one of java.util.Map, List, Set and Queue data structures in incoming message's payload. Its definition is as follows : +Hazelcast Outbound Channel Adapter listens its defined channel and writes incoming messages to related distributed cache. It expects one of cache, cache-expression or HazelcastHeaders.CACHE_NAME for distributed object definition. Supported Distributed Objects : IMap, MultiMap, ReplicatedMap, IList, ISet, IQueue and ITopic. Its sample definition is as follows : ``` - + ``` -Basically, it requires two attributes as follows : +Basically, it requires the following attributes : -* **channel :** Specifies channel which message is sent. It is mandatory attribute. -* **cache :** Specifies distributed Map reference which is queried. It is mandatory attribute. +**channel :** Specifies channel which message is sent. +* **cache :** Specifies distributed object reference. It is optional attribute. +* **cache-expression :** Specifies distributed object via Spring Expression Language(SpEL). It is optional attribute. +* **key-expression :** Specifies key of K,V pair via Spring Expression Language(SpEL). It is optional attribute and required for just IMap, MultiMap and ReplicatedMap distributed data structures. +* **extract-payload :** Specifies whole message or just payload to send. It is optional attribute with **true** default value. If it is true, just payload will be written to distributed object. Otherwise, whole message will be written by covering both message header and payload. -**Case 1-** If incoming messages need to be written to distributed map(com.hazelcast.core.IMap), Message should have java.util.Map payload. Sample definition should be as follows : +**Sample Definitions :** ``` - + +``` +**OR** +``` + +``` +By setting distributed object name in the header, messages can be written to different distributed objects via same channel. - +**OR** - - - - - - - - - -``` -**Case 2-** If incoming messages need to be written to distributed list(com.hazelcast.core.IList), Message should have java.util.List payload. Sample definition should be as follows : -``` - - - - - - - -``` -**Case 3-** If incoming messages need to be written to distributed set(com.hazelcast.core.ISet), Message should have java.util.Set payload. Sample definition should be as follows : -``` - - - - - - - -``` -**Case 4-** If incoming messages need to be written to distributed queue(com.hazelcast.core.IQueue), Message should have java.util.Queue payload. Sample definition should be as follows : -``` - - - - - - - -``` +If **cache** or **cache-expression** attributes are not defined, HazelcastHeaders.CACHE_NAME has to be set in Message. diff --git a/spring-integration-hazelcast/build.gradle b/spring-integration-hazelcast/build.gradle index 6342e9a..e1e86ec 100644 --- a/spring-integration-hazelcast/build.gradle +++ b/spring-integration-hazelcast/build.gradle @@ -16,8 +16,8 @@ sourceCompatibility = targetCompatibility = 1.7 ext { hazelcastVersion = '3.4.2' jacocoVersion = '0.7.2.201409121644' - slf4jVersion = '1.7.10' - springIntegrationVersion = '4.1.2.RELEASE' + slf4jVersion = '1.7.11' + springIntegrationVersion = '4.1.3.RELEASE' idPrefix = 'hazelcast' diff --git a/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/HazelcastIntegrationDefinitionValidator.java b/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/HazelcastIntegrationDefinitionValidator.java index e9e68fa..485af74 100644 --- a/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/HazelcastIntegrationDefinitionValidator.java +++ b/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/HazelcastIntegrationDefinitionValidator.java @@ -61,17 +61,6 @@ public class HazelcastIntegrationDefinitionValidator { } } - public static void validateCacheTypeForCacheWritingMessageHandler(final DistributedObject distributedObject) { - if (!(distributedObject instanceof IMap - || distributedObject instanceof IList - || distributedObject instanceof ISet - || distributedObject instanceof IQueue)) { - throw new IllegalArgumentException( - "Invalid 'cache' type is set. IMap, IList, ISet and IQueue cache object types are acceptable " - + "for Hazelcast Outbound Channel Adapter."); - } - } - public static void validateCacheEventsByDistributedObject( final DistributedObject distributedObject, final Set cacheEventTypeSet) { List supportedCacheEventTypes = getSupportedCacheEventTypes(distributedObject); diff --git a/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/config/xml/HazelcastOutboundChannelAdapterParser.java b/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/config/xml/HazelcastOutboundChannelAdapterParser.java index 33623ac..261a217 100644 --- a/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/config/xml/HazelcastOutboundChannelAdapterParser.java +++ b/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/config/xml/HazelcastOutboundChannelAdapterParser.java @@ -18,16 +18,17 @@ package org.springframework.integration.hazelcast.config.xml; import org.w3c.dom.Element; +import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.support.AbstractBeanDefinition; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.xml.ParserContext; import org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser; +import org.springframework.integration.config.xml.IntegrationNamespaceUtils; import org.springframework.integration.hazelcast.outbound.HazelcastCacheWritingMessageHandler; -import reactor.util.StringUtils; - /** - * Hazelcast Outbound Channel Adapter Parser for {@code }. + * Hazelcast Outbound Channel Adapter Parser for + * {@code }. * * @author Eren Avsarogullari * @since 1.0.0 @@ -36,16 +37,33 @@ public class HazelcastOutboundChannelAdapterParser extends AbstractOutboundChann private static final String CACHE_ATTRIBUTE = "cache"; + private static final String CACHE_EXPRESSION_ATTRIBUTE = "cache-expression"; + + private static final String KEY_EXPRESSION_ATTRIBUTE = "key-expression"; + + private static final String EXTRACT_PAYLOAD_ATTRIBUTE = "extract-payload"; + + private static final String DISTRIBUTED_OBJECT = "distributedObject"; + @Override protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) { BeanDefinitionBuilder builder = BeanDefinitionBuilder .genericBeanDefinition(HazelcastCacheWritingMessageHandler.class); - if (!StringUtils.hasText(element.getAttribute(CACHE_ATTRIBUTE))) { - parserContext.getReaderContext().error("'" + CACHE_ATTRIBUTE + "' attribute is required.", element); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, CACHE_ATTRIBUTE, DISTRIBUTED_OBJECT); + BeanDefinition cacheExpressionDef = + IntegrationNamespaceUtils.createExpressionDefIfAttributeDefined(CACHE_EXPRESSION_ATTRIBUTE, element); + if (cacheExpressionDef != null) { + builder.addPropertyValue("cacheExpression", cacheExpressionDef); } - builder.addConstructorArgReference(element.getAttribute(CACHE_ATTRIBUTE)); + BeanDefinition keyExpressionDef = + IntegrationNamespaceUtils.createExpressionDefIfAttributeDefined(KEY_EXPRESSION_ATTRIBUTE, element); + if (keyExpressionDef != null) { + builder.addPropertyValue("keyExpression", keyExpressionDef); + } + + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, EXTRACT_PAYLOAD_ATTRIBUTE); return builder.getBeanDefinition(); } diff --git a/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/outbound/HazelcastCacheWritingMessageHandler.java b/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/outbound/HazelcastCacheWritingMessageHandler.java index f58b709..736c3f9 100644 --- a/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/outbound/HazelcastCacheWritingMessageHandler.java +++ b/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/outbound/HazelcastCacheWritingMessageHandler.java @@ -16,64 +16,150 @@ package org.springframework.integration.hazelcast.outbound; -import java.util.List; +import java.util.Collection; import java.util.Map; -import java.util.Queue; -import java.util.Set; +import org.springframework.expression.EvaluationContext; +import org.springframework.expression.Expression; +import org.springframework.integration.expression.IntegrationEvaluationContextAware; import org.springframework.integration.handler.AbstractMessageHandler; -import org.springframework.integration.hazelcast.HazelcastIntegrationDefinitionValidator; +import org.springframework.integration.hazelcast.HazelcastHeaders; import org.springframework.messaging.Message; import org.springframework.util.Assert; import com.hazelcast.core.DistributedObject; -import com.hazelcast.core.IList; -import com.hazelcast.core.IMap; -import com.hazelcast.core.IQueue; -import com.hazelcast.core.ISet; +import com.hazelcast.core.ITopic; +import com.hazelcast.core.MultiMap; /** - * MessageHandler implementation that writes {@link Message} payload to defined Hazelcast - * distributed cache object. Currently, it supports {@link java.util.Map}, - * {@link java.util.List}, {@link java.util.Set} and {@link java.util.Queue} data - * structures. + * MessageHandler implementation that writes {@link Message} or payload to defined + * Hazelcast distributed cache object. * * @author Eren Avsarogullari + * @author Artem Bilan * @since 1.0.0 */ -public class HazelcastCacheWritingMessageHandler extends AbstractMessageHandler { +public class HazelcastCacheWritingMessageHandler extends AbstractMessageHandler + implements IntegrationEvaluationContextAware { - private final DistributedObject distributedObject; + private DistributedObject distributedObject; - public HazelcastCacheWritingMessageHandler(DistributedObject distributedObject) { + private Expression cacheExpression; + + private Expression keyExpression; + + private boolean extractPayload = true; + + private EvaluationContext evaluationContext; + + public void setDistributedObject(DistributedObject distributedObject) { Assert.notNull(distributedObject, "'distributedObject' must not be null"); this.distributedObject = distributedObject; } - @Override - protected void onInit() throws Exception { - super.onInit(); - HazelcastIntegrationDefinitionValidator.validateCacheTypeForCacheWritingMessageHandler(this.distributedObject); + public void setCacheExpression(Expression cacheExpression) { + Assert.notNull(cacheExpression, "'cacheExpression' must not be null"); + this.cacheExpression = cacheExpression; + } + + public void setKeyExpression(Expression keyExpression) { + Assert.notNull(keyExpression, "'keyExpression' must not be null"); + this.keyExpression = keyExpression; + } + + public void setExtractPayload(boolean extractPayload) { + this.extractPayload = extractPayload; } @Override - protected void handleMessageInternal(Message message) throws Exception { - writeToCache(message); + public void setIntegrationEvaluationContext(EvaluationContext evaluationContext) { + this.evaluationContext = evaluationContext; } - @SuppressWarnings({ "unchecked", "rawtypes" }) - private void writeToCache(Message message) { - if (this.distributedObject instanceof IMap) { - ((IMap) this.distributedObject).putAll((Map) message.getPayload()); + @Override + @SuppressWarnings({"unchecked", "rawtypes"}) + protected void handleMessageInternal(final Message message) throws Exception { + Object objectToStore = message; + if (this.extractPayload) { + objectToStore = message.getPayload(); } - else if (this.distributedObject instanceof IList) { - ((IList) this.distributedObject).addAll((List) message.getPayload()); + + DistributedObject distributedObject = getDistributedObject(message); + + if (distributedObject instanceof Map) { + Map map = (Map) distributedObject; + if (objectToStore instanceof Map) { + map.putAll((Map) objectToStore); + } + else if (objectToStore instanceof Map.Entry) { + Map.Entry entry = (Map.Entry) objectToStore; + map.put(entry.getKey(), entry.getValue()); + } + else { + map.put(getKey(message), objectToStore); + } } - else if (this.distributedObject instanceof ISet) { - ((ISet) this.distributedObject).addAll((Set) message.getPayload()); + else if (distributedObject instanceof MultiMap) { + MultiMap map = (MultiMap) distributedObject; + if (objectToStore instanceof Map) { + Map mapToStore = (Map) objectToStore; + for (Map.Entry entry : mapToStore.entrySet()) { + map.put(entry.getKey(), entry.getValue()); + } + } + else if (objectToStore instanceof Map.Entry) { + Map.Entry entry = (Map.Entry) objectToStore; + map.put(entry.getKey(), entry.getValue()); + } + else { + map.put(getKey(message), objectToStore); + } } - else if (this.distributedObject instanceof IQueue) { - ((IQueue) this.distributedObject).addAll((Queue) message.getPayload()); + else if (distributedObject instanceof ITopic) { + ((ITopic) distributedObject).publish(objectToStore); + } + else if (distributedObject instanceof Collection) { + if (objectToStore instanceof Collection) { + ((Collection) distributedObject).addAll((Collection) objectToStore); + } + else { + ((Collection) distributedObject).add(objectToStore); + } + } + else { + throw new IllegalStateException("The 'distributedObject' for 'HazelcastCacheWritingMessageHandler' " + + "must be of 'IMap', 'MultiMap', 'ITopic', 'ISet' or 'IList' type, " + + "but gotten: [" + distributedObject + "]."); + } + } + + + private DistributedObject getDistributedObject(final Message message) { + if (this.distributedObject != null) { + return this.distributedObject; + } + else if (this.cacheExpression != null) { + return this.cacheExpression.getValue(this.evaluationContext, message, DistributedObject.class); + } + else if (message.getHeaders().containsKey(HazelcastHeaders.CACHE_NAME)) { + return getBeanFactory() + .getBean(message.getHeaders().get(HazelcastHeaders.CACHE_NAME, String.class), + DistributedObject.class); + } + else { + throw new IllegalStateException("One of 'cache', 'cache-expression' and " + + HazelcastHeaders.CACHE_NAME + + " must be set for cache object definition."); + } + } + + private Object getKey(Message message) { + if (this.keyExpression != null) { + return this.keyExpression.getValue(this.evaluationContext, message); + } + else { + throw new IllegalStateException( + "'key-expression' must be set to place the raw 'payload' to the IMap, MultiMap and ReplicatedMap"); } } diff --git a/spring-integration-hazelcast/src/main/resources/org/springframework/integration/hazelcast/config/xml/spring-integration-hazelcast-1.0.xsd b/spring-integration-hazelcast/src/main/resources/org/springframework/integration/hazelcast/config/xml/spring-integration-hazelcast-1.0.xsd index 1864d20..95c772a 100644 --- a/spring-integration-hazelcast/src/main/resources/org/springframework/integration/hazelcast/config/xml/spring-integration-hazelcast-1.0.xsd +++ b/spring-integration-hazelcast/src/main/resources/org/springframework/integration/hazelcast/config/xml/spring-integration-hazelcast-1.0.xsd @@ -84,7 +84,7 @@ - + @@ -96,6 +96,30 @@ + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/HazelcastIntegrationTestUser.java b/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/HazelcastIntegrationTestUser.java index eebcde8..b619096 100644 --- a/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/HazelcastIntegrationTestUser.java +++ b/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/HazelcastIntegrationTestUser.java @@ -24,7 +24,7 @@ import java.io.Serializable; * @author Eren Avsarogullari * @since 1.0.0 */ -public class HazelcastIntegrationTestUser implements Serializable { +public class HazelcastIntegrationTestUser implements Comparable, Serializable { private static final long serialVersionUID = -5357485957528362705L; @@ -126,4 +126,9 @@ public class HazelcastIntegrationTestUser implements Serializable { return true; } + @Override + public int compareTo(HazelcastIntegrationTestUser user) { + return (this.id < user.getId()) ? -1: (this.id > user.getId()) ? 1 : 0; + } + } diff --git a/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/HazelcastTestRequestHandlerAdvice.java b/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/HazelcastTestRequestHandlerAdvice.java new file mode 100644 index 0000000..820a8dc --- /dev/null +++ b/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/HazelcastTestRequestHandlerAdvice.java @@ -0,0 +1,48 @@ +/* + * Copyright 2015 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; + +import java.util.concurrent.CountDownLatch; + +import org.springframework.integration.handler.advice.AbstractRequestHandlerAdvice; +import org.springframework.messaging.Message; + +/** + * {@link MessageHandler} advice class for Hazelcast Integration Unit Tests. + * + * @author Eren Avsarogullari + * @since 1.0.0 + */ +public class HazelcastTestRequestHandlerAdvice extends AbstractRequestHandlerAdvice { + + public CountDownLatch executeLatch = null; + + public HazelcastTestRequestHandlerAdvice(int count) { + this.executeLatch = new CountDownLatch(count); + } + + @Override + protected Object doInvoke(ExecutionCallback callback, Object target, Message message) throws Exception { + try { + return callback.execute(); + } + finally { + this.executeLatch.countDown(); + } + } + +} diff --git a/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/outbound/HazelcastOutboundChannelAdapterTests-context.xml b/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/outbound/HazelcastOutboundChannelAdapterTests-context.xml index bf0702b..cc3283f 100644 --- a/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/outbound/HazelcastOutboundChannelAdapterTests-context.xml +++ b/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/outbound/HazelcastOutboundChannelAdapterTests-context.xml @@ -10,52 +10,269 @@ http://www.springframework.org/schema/integration/hazelcast http://www.springframework.org/schema/integration/hazelcast/spring-integration-hazelcast.xsd"> - + + + + + + + + + + + + + + + + + + + + + + + - + - + - + + + - + + + + + - + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + - + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/outbound/HazelcastOutboundChannelAdapterTests.java b/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/outbound/HazelcastOutboundChannelAdapterTests.java index 1b44984..8e6d737 100644 --- a/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/outbound/HazelcastOutboundChannelAdapterTests.java +++ b/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/outbound/HazelcastOutboundChannelAdapterTests.java @@ -16,9 +16,8 @@ package org.springframework.integration.hazelcast.outbound; -import static org.hamcrest.Matchers.instanceOf; +import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import java.util.ArrayList; @@ -28,31 +27,41 @@ import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Map.Entry; import java.util.Queue; import java.util.Set; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.LinkedBlockingQueue; +import java.util.SortedSet; +import java.util.TreeMap; +import java.util.TreeSet; +import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.TimeUnit; import javax.annotation.Resource; -import org.junit.Assert; import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.integration.handler.advice.AbstractRequestHandlerAdvice; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.integration.hazelcast.HazelcastHeaders; +import org.springframework.integration.hazelcast.HazelcastIntegrationTestUser; +import org.springframework.integration.hazelcast.HazelcastTestRequestHandlerAdvice; +import org.springframework.integration.support.DefaultMessageBuilderFactory; +import org.springframework.integration.support.MessageBuilderFactory; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandlingException; -import org.springframework.messaging.PollableChannel; -import org.springframework.messaging.support.ErrorMessage; import org.springframework.messaging.support.GenericMessage; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import com.hazelcast.core.ITopic; +import com.hazelcast.core.MessageListener; +import com.hazelcast.core.MultiMap; +import com.hazelcast.core.ReplicatedMap; + /** * Hazelcast Outbound Channel Adapter Test Class * @@ -63,175 +72,477 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @RunWith(SpringJUnit4ClassRunner.class) @ContextConfiguration @DirtiesContext +@SuppressWarnings({ "rawtypes", "unchecked" }) public class HazelcastOutboundChannelAdapterTests { private static final int DATA_COUNT = 100; - @Autowired - private MessageChannel mapChannel; + private static final int DEFAULT_AGE = 5; + + private static final String TEST_NAME = "Test_Name"; + + private static final String TEST_SURNAME = "Test_Surname"; + + private static final String DISTRIBUTED_MAP = "distributedMap"; + + private static final String CACHE_HEADER = "CACHE_HEADER"; + + private final MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory(); @Autowired + @Qualifier("firstMapChannel") + private MessageChannel firstMapChannel; + + @Autowired + @Qualifier("secondMapChannel") + private MessageChannel secondMapChannel; + + @Autowired + @Qualifier("thirdMapChannel") + private MessageChannel thirdMapChannel; + + @Autowired + @Qualifier("fourthMapChannel") + private MessageChannel fourthMapChannel; + + @Autowired + @Qualifier("fifthMapChannel") + private MessageChannel fifthMapChannel; + + @Autowired + @Qualifier("sixthMapChannel") + private MessageChannel sixthMapChannel; + + @Autowired + @Qualifier("bulkMapChannel") + private MessageChannel bulkMapChannel; + + @Autowired + @Qualifier("multiMapChannel") + private MessageChannel multiMapChannel; + + @Autowired + @Qualifier("replicatedMapChannel") + private MessageChannel replicatedMapChannel; + + @Autowired + @Qualifier("bulkReplicatedMapChannel") + private MessageChannel bulkReplicatedMapChannel; + + @Autowired + @Qualifier("listChannel") private MessageChannel listChannel; @Autowired + @Qualifier("bulkListChannel") + private MessageChannel bulkListChannel; + + @Autowired + @Qualifier("setChannel") private MessageChannel setChannel; @Autowired + @Qualifier("bulkSetChannel") + private MessageChannel bulkSetChannel; + + @Autowired + @Qualifier("queueChannel") private MessageChannel queueChannel; @Autowired - private PollableChannel errorChannel; + @Qualifier("bulkQueueChannel") + private MessageChannel bulkQueueChannel; + + @Autowired + @Qualifier("topicChannel") + private MessageChannel topicChannel; + + @Autowired + @Qualifier("lockChannel") + private MessageChannel lockChannel; @Resource private Map distributedMap; @Resource - private List distributedList; + private Map distributedBulkMap; @Resource - private Set distributedSet; + private MultiMap multiMap; @Resource - private Queue distributedQueue; + private ReplicatedMap replicatedMap; + + @Resource + private ReplicatedMap bulkReplicatedMap; + + @Resource + private List distributedList; + + @Resource + private List distributedBulkList; + + @Resource + private Set distributedSet; + + @Resource + private Set distributedBulkSet; + + @Resource + private Queue distributedQueue; + + @Resource + private Queue distributedBulkQueue; + + @Resource + private ITopic topic; @Autowired - private TestRequestHandlerAdvice testRequestHandlerAdvice; + @Qualifier("testFirstMapRequestHandlerAdvice") + private HazelcastTestRequestHandlerAdvice testFirstMapRequestHandlerAdvice; + + @Autowired + @Qualifier("testSecondMapRequestHandlerAdvice") + private HazelcastTestRequestHandlerAdvice testSecondMapRequestHandlerAdvice; + + @Autowired + @Qualifier("testThirdMapRequestHandlerAdvice") + private HazelcastTestRequestHandlerAdvice testThirdMapRequestHandlerAdvice; + + @Autowired + @Qualifier("testFourthMapRequestHandlerAdvice") + private HazelcastTestRequestHandlerAdvice testFourthMapRequestHandlerAdvice; + + @Autowired + @Qualifier("testBulkMapRequestHandlerAdvice") + private HazelcastTestRequestHandlerAdvice testBulkMapRequestHandlerAdvice; + + @Autowired + @Qualifier("testMultiMapRequestHandlerAdvice") + private HazelcastTestRequestHandlerAdvice testMultiMapRequestHandlerAdvice; + + @Autowired + @Qualifier("testReplicatedMapRequestHandlerAdvice") + private HazelcastTestRequestHandlerAdvice testReplicatedMapRequestHandlerAdvice; + + @Autowired + @Qualifier("testBulkReplicatedMapRequestHandlerAdvice") + private HazelcastTestRequestHandlerAdvice testBulkReplicatedMapRequestHandlerAdvice; + + @Autowired + @Qualifier("testListRequestHandlerAdvice") + private HazelcastTestRequestHandlerAdvice testListRequestHandlerAdvice; + + @Autowired + @Qualifier("testBulkListRequestHandlerAdvice") + private HazelcastTestRequestHandlerAdvice testBulkListRequestHandlerAdvice; + + @Autowired + @Qualifier("testSetRequestHandlerAdvice") + private HazelcastTestRequestHandlerAdvice testSetRequestHandlerAdvice; + + @Autowired + @Qualifier("testBulkSetRequestHandlerAdvice") + private HazelcastTestRequestHandlerAdvice testBulkSetRequestHandlerAdvice; + + @Autowired + @Qualifier("testQueueRequestHandlerAdvice") + private HazelcastTestRequestHandlerAdvice testQueueRequestHandlerAdvice; + + @Autowired + @Qualifier("testBulkQueueRequestHandlerAdvice") + private HazelcastTestRequestHandlerAdvice testBulkQueueRequestHandlerAdvice; + + @Autowired + @Qualifier("testTopicRequestHandlerAdvice") + private HazelcastTestRequestHandlerAdvice testTopicRequestHandlerAdvice; @Before public void setUp() { - distributedMap.clear(); - distributedList.clear(); - distributedSet.clear(); - distributedQueue.clear(); + this.distributedMap.clear(); + this.distributedBulkMap.clear(); + this.distributedList.clear(); + this.distributedBulkList.clear(); + this.distributedSet.clear(); + this.distributedBulkSet.clear(); + this.distributedQueue.clear(); + this.distributedBulkQueue.clear(); + this.multiMap.clear(); + this.replicatedMap.clear(); + this.bulkReplicatedMap.clear(); } @Test - public void testWriteDistributedMap() { - Map map = createMapByEntryCount(); - mapChannel.send(new GenericMessage<>(map)); - verifyDistributedMap(); + public void testWriteToDistributedMap() throws InterruptedException { + sendMessageToChannel(this.firstMapChannel); + assertTrue(this.testFirstMapRequestHandlerAdvice.executeLatch.await(10, + TimeUnit.SECONDS)); + verifyMapForPayload(new TreeMap(this.distributedMap)); } @Test - public void testWriteDistributedList() { - List list = (List) fillCollectionByEntryCount(new ArrayList()); - listChannel.send(new GenericMessage<>(list)); - verifyDistributedList(); + public void testBulkWriteToDistributedMap() throws InterruptedException { + Map userMap = new HashMap<>(DATA_COUNT); + for (int index = 1; index <= DATA_COUNT; index++) { + userMap.put(index, getTestUser(index)); + } + + this.bulkMapChannel.send(new GenericMessage<>(userMap)); + + assertTrue(this.testBulkMapRequestHandlerAdvice.executeLatch.await(10, + TimeUnit.SECONDS)); + verifyMapForPayload(new TreeMap(this.distributedBulkMap)); } @Test - public void testWriteDistributedSet() { - Set set = (Set) fillCollectionByEntryCount(new HashSet()); - setChannel.send(new GenericMessage<>(set)); - verifyDistributedSet(); + public void testWriteToDistributedMapWhenCacheExpressionIsSet() + throws InterruptedException { + sendMessageWithCacheHeaderToChannel(this.secondMapChannel, CACHE_HEADER, + DISTRIBUTED_MAP); + assertTrue(this.testSecondMapRequestHandlerAdvice.executeLatch.await(10, + TimeUnit.SECONDS)); + verifyMapForPayload(new TreeMap(this.distributedMap)); } @Test - public void testWriteDistributedQueue() throws InterruptedException { - Collection queue = fillCollectionByEntryCount(new LinkedBlockingQueue(DATA_COUNT)); - this.queueChannel.send(new GenericMessage<>(queue)); - - assertTrue(this.testRequestHandlerAdvice.executeLatch.await(10, TimeUnit.SECONDS)); - - Assert.assertEquals(true, this.distributedQueue.size() == DATA_COUNT); - int index = 0; - for (Object o : this.distributedQueue) { - Assert.assertEquals(index++, o); - } + public void testWriteToDistributedMapWhenHazelcastHeaderIsSet() + throws InterruptedException { + sendMessageWithCacheHeaderToChannel(this.thirdMapChannel, + HazelcastHeaders.CACHE_NAME, DISTRIBUTED_MAP); + assertTrue(this.testThirdMapRequestHandlerAdvice.executeLatch.await(10, + TimeUnit.SECONDS)); + verifyMapForPayload(new TreeMap(this.distributedMap)); } - @Test(expected = MessageHandlingException.class) - public void testMapChannelWithIncorrectDataType() { - Set set = new HashSet<>(); - set.add(1); - mapChannel.send(new GenericMessage<>(set)); + @Test + public void testWriteToDistributedMapWhenExtractPayloadIsFalse() + throws InterruptedException { + sendMessageWithCacheHeaderToChannel(this.fourthMapChannel, + HazelcastHeaders.CACHE_NAME, DISTRIBUTED_MAP); + assertTrue(this.testFourthMapRequestHandlerAdvice.executeLatch.await(10, + TimeUnit.SECONDS)); + verifyMapForMessage(new TreeMap(this.distributedMap)); } - @Test(expected = MessageHandlingException.class) - public void testListChannelWithIncorrectDataType() { - Set set = new HashSet<>(); - set.add(1); - listChannel.send(new GenericMessage<>(set)); + @Test + public void testWriteToMultiMap() throws InterruptedException { + sendMessageToChannel(this.multiMapChannel); + assertTrue(this.testMultiMapRequestHandlerAdvice.executeLatch.await(10, + TimeUnit.SECONDS)); + verifyMultiMapForPayload(this.multiMap); } - @Test(expected = MessageHandlingException.class) - public void testSetChannelWithIncorrectDataType() { - List list = new ArrayList<>(); - list.add(1); - setChannel.send(new GenericMessage<>(list)); + @Test + public void testWriteToReplicatedMap() throws InterruptedException { + sendMessageToChannel(this.replicatedMapChannel); + assertTrue(this.testReplicatedMapRequestHandlerAdvice.executeLatch.await(10, + TimeUnit.SECONDS)); + verifyMapForPayload(new TreeMap(this.replicatedMap)); } - public void testQueueChannelWithIncorrectDataType() { - Set set = new HashSet<>(); - set.add(1); - this.queueChannel.send(new GenericMessage<>(set)); - Message receive = this.errorChannel.receive(10000); - assertNotNull(receive); - assertThat(receive, instanceOf(ErrorMessage.class)); - assertThat(receive.getPayload(), instanceOf(MessageHandlingException.class)); - - } - - private Map createMapByEntryCount() { - Map map = new HashMap<>(); - StringBuilder strBuilder = new StringBuilder(); - for (int index = 0; index < DATA_COUNT; index++) { - String value = strBuilder.append("Value_").append(index).toString(); - map.put(index, value); - strBuilder.delete(0, strBuilder.length()); + @Test + public void testBulkWriteToReplicatedMap() throws InterruptedException { + Map userMap = new HashMap<>(DATA_COUNT); + for (int index = 1; index <= DATA_COUNT; index++) { + userMap.put(index, getTestUser(index)); } - return map; + this.bulkReplicatedMapChannel.send(new GenericMessage<>(userMap)); + + assertTrue(this.testBulkReplicatedMapRequestHandlerAdvice.executeLatch.await(10, + TimeUnit.SECONDS)); + verifyMapForPayload(new TreeMap(this.bulkReplicatedMap)); } - private void verifyDistributedMap() { - Assert.assertEquals(true, distributedMap.size() == DATA_COUNT); - StringBuilder strBuilder = new StringBuilder(); - for (int index = 0; index < DATA_COUNT; index++) { - String value = strBuilder.append("Value_").append(index).toString(); - Assert.assertEquals(value, distributedMap.get(index)); - strBuilder.delete(0, strBuilder.length()); - } + @Test + public void testWriteToDistributedList() throws InterruptedException { + sendMessageToChannel(this.listChannel); + assertTrue(this.testListRequestHandlerAdvice.executeLatch.await(10, + TimeUnit.SECONDS)); + verifyCollection(this.distributedList, DATA_COUNT); } - private Collection fillCollectionByEntryCount(Collection coll) { - for (int index = 0; index < DATA_COUNT; index++) { - coll.add(index); + @Test + public void testBulkWriteToDistributedList() throws InterruptedException { + List userList = new ArrayList<>(DATA_COUNT); + for (int index = 1; index <= DATA_COUNT; index++) { + userList.add(getTestUser(index)); } - return coll; + this.bulkListChannel.send(new GenericMessage<>(userList)); + + assertTrue(this.testBulkListRequestHandlerAdvice.executeLatch.await(10, + TimeUnit.SECONDS)); + verifyCollection(this.distributedBulkList, DATA_COUNT); } - private void verifyDistributedList() { - Assert.assertEquals(true, distributedList.size() == DATA_COUNT); - for (int index = 0; index < DATA_COUNT; index++) { - Assert.assertEquals(index, distributedList.get(index)); - } - } - - @SuppressWarnings({"unchecked", "rawtypes"}) - private void verifyDistributedSet() { - Assert.assertEquals(true, distributedSet.size() == DATA_COUNT); - List list = new ArrayList(distributedSet); + @Test + public void testWriteToDistributedSet() throws InterruptedException { + sendMessageToChannel(this.setChannel); + assertTrue(this.testSetRequestHandlerAdvice.executeLatch.await(10, + TimeUnit.SECONDS)); + final List list = new ArrayList(this.distributedSet); Collections.sort(list); - for (int index = 0; index < DATA_COUNT; index++) { - Assert.assertEquals(index, list.get(index)); + verifyCollection(list, DATA_COUNT); + } + + @Test + public void testBulkWriteToDistributedSet() throws InterruptedException { + Set userSet = new HashSet<>(DATA_COUNT); + for (int index = 1; index <= DATA_COUNT; index++) { + userSet.add(getTestUser(index)); + } + + this.bulkSetChannel.send(new GenericMessage<>(userSet)); + + assertTrue(this.testBulkSetRequestHandlerAdvice.executeLatch.await(10, + TimeUnit.SECONDS)); + final List list = new ArrayList(this.distributedBulkSet); + Collections.sort(list); + verifyCollection(list, DATA_COUNT); + } + + @Test + public void testWriteToDistributedQueue() throws InterruptedException { + sendMessageToChannel(this.queueChannel); + assertTrue(this.testQueueRequestHandlerAdvice.executeLatch.await(10, + TimeUnit.SECONDS)); + verifyCollection(this.distributedQueue, DATA_COUNT); + } + + @Test + public void testBulkWriteToDistributedQueue() throws InterruptedException { + Queue userQueue = new ArrayBlockingQueue(DATA_COUNT); + for (int index = 1; index <= DATA_COUNT; index++) { + userQueue.add(getTestUser(index)); + } + + this.bulkQueueChannel.send(new GenericMessage<>(userQueue)); + + assertTrue(this.testBulkQueueRequestHandlerAdvice.executeLatch.await(10, + TimeUnit.SECONDS)); + verifyCollection(this.distributedBulkQueue, DATA_COUNT); + } + + @Test + public void testWriteToTopic() throws InterruptedException { + this.topic.addMessageListener(new TestTopicMessageListener()); + sendMessageToChannel(this.topicChannel); + assertTrue(this.testTopicRequestHandlerAdvice.executeLatch.await(10, + TimeUnit.SECONDS)); + } + + @Test(expected = MessageHandlingException.class) + public void testWriteToDistributedMapWhenCacheIsNotSet() { + this.fifthMapChannel.send(new GenericMessage<>(getTestUser(1))); + } + + @Test(expected = MessageHandlingException.class) + public void testWriteToDistributedMapWhenKeyExpressionIsNotSet() { + Message message = this.messageBuilderFactory + .withPayload(getTestUser(1)) + .setHeader(HazelcastHeaders.CACHE_NAME, DISTRIBUTED_MAP).build(); + this.sixthMapChannel.send(message); + } + + @Test(expected = MessageHandlingException.class) + public void testWriteToLock() { + this.lockChannel.send(new GenericMessage<>("foo")); + } + + private void sendMessageToChannel(final MessageChannel channel) { + for (int index = 1; index <= DATA_COUNT; index++) { + channel.send(new GenericMessage<>(getTestUser(index))); } } - public static class TestRequestHandlerAdvice extends AbstractRequestHandlerAdvice { + private void sendMessageWithCacheHeaderToChannel(final MessageChannel channel, + final String headerName, final String distributedObjectName) { + for (int index = 1; index <= DATA_COUNT; index++) { + Message message = this.messageBuilderFactory + .withPayload(getTestUser(index)) + .setHeader(headerName, distributedObjectName).build(); + channel.send(message); + } + } - public final CountDownLatch executeLatch = new CountDownLatch(1); + private void verifyMapForPayload(final Map map) { + int index = 1; + assertNotNull(map); + assertEquals(true, map.size() == DATA_COUNT); + for (Entry entry : map.entrySet()) { + assertNotNull(entry); + assertEquals(index, entry.getKey().intValue()); + verifyHazelcastIntegrationTestUser(entry.getValue(), index); + index++; + } + } + + private void verifyMultiMapForPayload( + final MultiMap multiMap) { + int index = 1; + assertNotNull(multiMap); + assertEquals(true, multiMap.size() == DATA_COUNT); + SortedSet keys = new TreeSet<>(multiMap.keySet()); + for (Integer key : keys) { + assertNotNull(key); + assertEquals(index, key.intValue()); + HazelcastIntegrationTestUser user = multiMap.get(key).iterator().next(); + verifyHazelcastIntegrationTestUser(user, index); + index++; + } + } + + private void verifyMapForMessage( + final Map> map) { + int index = 1; + assertNotNull(map); + assertEquals(true, map.size() == DATA_COUNT); + for (Entry> entry : map.entrySet()) { + assertNotNull(entry); + assertEquals(index, entry.getKey().intValue()); + assertTrue(entry.getValue().getHeaders().size() > 0); + verifyHazelcastIntegrationTestUser(entry.getValue().getPayload(), index); + index++; + } + } + + private void verifyCollection(final Collection coll, + final int dataCount) { + int index = 1; + assertNotNull(coll); + assertEquals(true, coll.size() == dataCount); + for (HazelcastIntegrationTestUser user : coll) { + verifyHazelcastIntegrationTestUser(user, index); + index++; + } + } + + private void verifyHazelcastIntegrationTestUser(HazelcastIntegrationTestUser user, + int index) { + assertNotNull(user); + assertEquals(index, user.getId()); + assertEquals(TEST_NAME, user.getName()); + assertEquals(TEST_SURNAME, user.getSurname()); + assertEquals(index + DEFAULT_AGE, user.getAge()); + } + + private HazelcastIntegrationTestUser getTestUser(int index) { + return new HazelcastIntegrationTestUser(index, TEST_NAME, TEST_SURNAME, index + + DEFAULT_AGE); + } + + private class TestTopicMessageListener implements MessageListener { + + private int index = 1; @Override - protected Object doInvoke(ExecutionCallback callback, Object target, Message message) throws Exception { - try { - return callback.execute(); - } - finally { - this.executeLatch.countDown(); - } + public void onMessage(com.hazelcast.core.Message message) { + HazelcastIntegrationTestUser user = (HazelcastIntegrationTestUser) message + .getMessageObject(); + verifyHazelcastIntegrationTestUser(user, index); + index++; } }