INTEXT-149: Refactor HCCacheWritingMessageHandler

JIRA: https://jira.spring.io/browse/INTEXT-149 Some refactorings are done. Polishing * Add `IllegalStateException` for wrong `distributedObject` * Add an invalid test to attempt to write to `ILock`
2015-05-12 15:20:18 +03:00
parent 37d70e0e84
commit c6975af37b
10 changed files with 887 additions and 220 deletions
--- 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 :
 ```
-<int-hazelcast:outbound-channel-adapter  channel="mapChannel" cache="distributedMap" /> 
+<int-hazelcast:outbound-channel-adapter channel="mapChannel" cache="distributedMap" key-expression="payload.id" extract-payload="false"/>
 ```
-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 :**
 ```
-<int:channel id="mapChannel"/> 
+<int-hazelcast:outbound-channel-adapter channel="mapChannel" cache="distributedMap" key-expression="payload.id" extract-payload="false"/>
+```
+**OR**
+```
+<int-hazelcast:outbound-channel-adapter channel="mapChannel" cache-expression="headers['CACHE_HEADER']" key-expression="payload.key" extract-payload="true"/> 
+```
+By setting distributed object name in the header, messages can be written to different distributed objects via same channel.

-<int-hazelcast:outbound-channel-adapter  channel="mapChannel" cache="distributedMap" /> 
+**OR**

-<bean id="distributedMap" factory-bean="instance" factory-method="getMap"> 
-	<constructor-arg value="distributedMap"/> 
-</bean> 
-
-<bean id="instance" class="com.hazelcast.core.Hazelcast" 
-			factory-method="newHazelcastInstance"> 
-	<constructor-arg> 
-		<bean class="com.hazelcast.config.Config" /> 
-	</constructor-arg> 
-</bean> 
-```
-**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 : 
-```
-<int:channel id="listChannel"/> 
-	 
-<int-hazelcast:outbound-channel-adapter channel="listChannel" cache="distributedList" /> 
-
-<bean id="distributedList" factory-bean="instance" factory-method="getList"> 
-	<constructor-arg value="distributedList"/> 
-</bean> 
-```
-**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 : 
-```
-<int:channel id="setChannel"/> 
-
-<int-hazelcast:outbound-channel-adapter channel="setChannel" cache="distributedSet" /> 
-	 
-<bean id="distributedSet" factory-bean="instance" factory-method="getSet"> 
-	<constructor-arg value="distributedSet"/> 
-</bean> 
-```
-**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 : 
-```
-<int:channel id="queueChannel"/> 
-		 
-<int-hazelcast:outbound-channel-adapter  channel="queueChannel" cache="distributedQueue" /> 
-		 
-<bean id="distributedQueue" factory-bean="instance" factory-method="getQueue"> 
-	<constructor-arg value="distributedQueue"/> 
-</bean> 
-```	
+If **cache** or **cache-expression** attributes are not defined, HazelcastHeaders.CACHE_NAME has to be set in Message.
--- 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'

--- 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<String> cacheEventTypeSet) {
 		List<String> supportedCacheEventTypes = getSupportedCacheEventTypes(distributedObject);
--- 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 <int-hazelcast:inbound-channel-adapter />}.
+ * Hazelcast Outbound Channel Adapter Parser for
+ * {@code <int-hazelcast:inbound-channel-adapter />}.
 *
 * @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();
 	}
--- 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");
 		}
 	}

--- 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 @@

 			<xsd:attributeGroup ref="integration:channelAdapterAttributes"/>

-			<xsd:attribute name="cache" use="required" type="xsd:string">
+			<xsd:attribute name="cache" type="xsd:string" use="optional">
 				<xsd:annotation>
 					<xsd:appinfo>
 						<tool:annotation kind="ref">
@@ -96,6 +96,30 @@
 					</xsd:documentation>
 				</xsd:annotation>
 			</xsd:attribute>
+			
+			<xsd:attribute name="cache-expression" type="xsd:string" use="optional">
+				<xsd:annotation>
+					<xsd:documentation>
+						<![CDATA[ Specifies cache name to listen ]]>
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+			
+			<xsd:attribute name="key-expression" type="xsd:string" use="optional">
+				<xsd:annotation>
+					<xsd:documentation>
+						<![CDATA[ Specifies entry key ]]>
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+			
+			<xsd:attribute name="extract-payload" type="xsd:boolean" default="true">
+				<xsd:annotation>
+					<xsd:documentation>
+						<![CDATA[ Specifies whole message or just payload to send ]]>
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>

 			<xsd:attribute name="order" type="xsd:string" use="optional">
                <xsd:annotation>
--- 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<HazelcastIntegrationTestUser>, 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;
+	}
+
 }
--- 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
@@ -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();
+		}
+	}
+
+}
--- 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">

-	<int:channel id="mapChannel"/>
+	<int:channel id="firstMapChannel"/>
+
+	<int:channel id="secondMapChannel"/>
+
+	<int:channel id="thirdMapChannel"/>
+
+	<int:channel id="fourthMapChannel"/>
+
+	<int:channel id="fifthMapChannel"/>
+
+	<int:channel id="sixthMapChannel"/>
+
+	<int:channel id="bulkMapChannel"/>
+
+	<int:channel id="multiMapChannel"/>
+
+	<int:channel id="replicatedMapChannel"/>
+
+	<int:channel id="bulkReplicatedMapChannel"/>

 	<int:channel id="listChannel"/>

+	<int:channel id="bulkListChannel"/>
+
 	<int:channel id="setChannel"/>

+	<int:channel id="bulkSetChannel"/>
+
 	<int:channel id="queueChannel">
 		<int:queue/>
 	</int:channel>

-	<int:channel id="errorChannel">
+	<int:channel id="bulkQueueChannel">
 		<int:queue/>
 	</int:channel>

-	<int-hazelcast:outbound-channel-adapter channel="mapChannel" cache="distributedMap"/>
+	<int:channel id="topicChannel"/>

-	<int-hazelcast:outbound-channel-adapter channel="listChannel" cache="distributedList"/>
+	<bean id="testFirstMapRequestHandlerAdvice"
+		  class="org.springframework.integration.hazelcast.HazelcastTestRequestHandlerAdvice">
+		<constructor-arg type="int" value="100"/>
+	</bean>

-	<int-hazelcast:outbound-channel-adapter channel="setChannel" cache="distributedSet"/>
+	<int-hazelcast:outbound-channel-adapter channel="firstMapChannel" cache="distributedMap" key-expression="payload.id">
+		<int-hazelcast:request-handler-advice-chain>
+			<ref bean="testFirstMapRequestHandlerAdvice"/>
+		</int-hazelcast:request-handler-advice-chain>
+	</int-hazelcast:outbound-channel-adapter>

-	<bean id="testRequestHandlerAdvice"
-		  class="org.springframework.integration.hazelcast.outbound.HazelcastOutboundChannelAdapterTests$TestRequestHandlerAdvice"/>
+	<bean id="testSecondMapRequestHandlerAdvice"
+		  class="org.springframework.integration.hazelcast.HazelcastTestRequestHandlerAdvice">
+		<constructor-arg type="int" value="100"/>
+	</bean>
+
+	<int-hazelcast:outbound-channel-adapter channel="secondMapChannel" cache-expression="@distributedMap" key-expression="payload.id">
+		<int-hazelcast:request-handler-advice-chain>
+			<ref bean="testSecondMapRequestHandlerAdvice"/>
+		</int-hazelcast:request-handler-advice-chain>
+	</int-hazelcast:outbound-channel-adapter>
+
+	<bean id="testThirdMapRequestHandlerAdvice"
+		  class="org.springframework.integration.hazelcast.HazelcastTestRequestHandlerAdvice">
+		<constructor-arg type="int" value="100"/>
+	</bean>
+
+	<int-hazelcast:outbound-channel-adapter channel="thirdMapChannel" key-expression="payload.id">
+		<int-hazelcast:request-handler-advice-chain>
+			<ref bean="testThirdMapRequestHandlerAdvice"/>
+		</int-hazelcast:request-handler-advice-chain>
+	</int-hazelcast:outbound-channel-adapter>
+
+	<bean id="testFourthMapRequestHandlerAdvice"
+		  class="org.springframework.integration.hazelcast.HazelcastTestRequestHandlerAdvice">
+		<constructor-arg type="int" value="100"/>
+	</bean>
+
+	<int-hazelcast:outbound-channel-adapter channel="fourthMapChannel" key-expression="payload.id" extract-payload="false">
+		<int-hazelcast:request-handler-advice-chain>
+			<ref bean="testFourthMapRequestHandlerAdvice"/>
+		</int-hazelcast:request-handler-advice-chain>
+	</int-hazelcast:outbound-channel-adapter>
+
+	<int-hazelcast:outbound-channel-adapter channel="fifthMapChannel" key-expression="payload.id" />
+
+	<int-hazelcast:outbound-channel-adapter channel="sixthMapChannel" />
+
+	<bean id="testBulkMapRequestHandlerAdvice"
+		  class="org.springframework.integration.hazelcast.HazelcastTestRequestHandlerAdvice">
+		<constructor-arg type="int" value="1"/>
+	</bean>
+
+	<int-hazelcast:outbound-channel-adapter channel="bulkMapChannel" cache="distributedBulkMap" key-expression="payload.id">
+		<int-hazelcast:request-handler-advice-chain>
+			<ref bean="testBulkMapRequestHandlerAdvice"/>
+		</int-hazelcast:request-handler-advice-chain>
+	</int-hazelcast:outbound-channel-adapter>
+
+	<bean id="testMultiMapRequestHandlerAdvice"
+		  class="org.springframework.integration.hazelcast.HazelcastTestRequestHandlerAdvice">
+		<constructor-arg type="int" value="100"/>
+	</bean>
+
+	<int-hazelcast:outbound-channel-adapter channel="multiMapChannel" cache="multiMap" key-expression="payload.id">
+		<int-hazelcast:request-handler-advice-chain>
+			<ref bean="testMultiMapRequestHandlerAdvice"/>
+		</int-hazelcast:request-handler-advice-chain>
+	</int-hazelcast:outbound-channel-adapter>
+
+	<bean id="testReplicatedMapRequestHandlerAdvice"
+		  class="org.springframework.integration.hazelcast.HazelcastTestRequestHandlerAdvice">
+		<constructor-arg type="int" value="100"/>
+	</bean>
+
+	<int-hazelcast:outbound-channel-adapter channel="replicatedMapChannel" cache="replicatedMap" key-expression="payload.id">
+		<int-hazelcast:request-handler-advice-chain>
+			<ref bean="testReplicatedMapRequestHandlerAdvice"/>
+		</int-hazelcast:request-handler-advice-chain>
+	</int-hazelcast:outbound-channel-adapter>
+
+	<bean id="testBulkReplicatedMapRequestHandlerAdvice"
+		  class="org.springframework.integration.hazelcast.HazelcastTestRequestHandlerAdvice">
+		<constructor-arg type="int" value="1"/>
+	</bean>
+
+	<int-hazelcast:outbound-channel-adapter channel="bulkReplicatedMapChannel" cache="bulkReplicatedMap" key-expression="payload.id">
+		<int-hazelcast:request-handler-advice-chain>
+			<ref bean="testBulkReplicatedMapRequestHandlerAdvice"/>
+		</int-hazelcast:request-handler-advice-chain>
+	</int-hazelcast:outbound-channel-adapter>
+
+	<bean id="testListRequestHandlerAdvice"
+		  class="org.springframework.integration.hazelcast.HazelcastTestRequestHandlerAdvice">
+		<constructor-arg type="int" value="100"/>
+	</bean>
+
+	<int-hazelcast:outbound-channel-adapter channel="listChannel" cache="distributedList">
+		<int-hazelcast:request-handler-advice-chain>
+			<ref bean="testListRequestHandlerAdvice"/>
+		</int-hazelcast:request-handler-advice-chain>
+	</int-hazelcast:outbound-channel-adapter>
+
+	<bean id="testBulkListRequestHandlerAdvice"
+		  class="org.springframework.integration.hazelcast.HazelcastTestRequestHandlerAdvice">
+		  <constructor-arg type="int" value="1"/>
+	</bean>
+
+	<int-hazelcast:outbound-channel-adapter channel="bulkListChannel" cache="distributedBulkList">
+		<int-hazelcast:request-handler-advice-chain>
+			<ref bean="testBulkListRequestHandlerAdvice"/>
+		</int-hazelcast:request-handler-advice-chain>
+	</int-hazelcast:outbound-channel-adapter>
+
+	<bean id="testSetRequestHandlerAdvice"
+		  class="org.springframework.integration.hazelcast.HazelcastTestRequestHandlerAdvice">
+		<constructor-arg type="int" value="100"/>
+	</bean>
+
+	<int-hazelcast:outbound-channel-adapter channel="setChannel" cache="distributedSet" >
+		<int-hazelcast:request-handler-advice-chain>
+			<ref bean="testSetRequestHandlerAdvice"/>
+		</int-hazelcast:request-handler-advice-chain>
+	</int-hazelcast:outbound-channel-adapter>
+
+	<bean id="testBulkSetRequestHandlerAdvice"
+		  class="org.springframework.integration.hazelcast.HazelcastTestRequestHandlerAdvice">
+		<constructor-arg type="int" value="1"/>
+	</bean>
+
+	<int-hazelcast:outbound-channel-adapter channel="bulkSetChannel" cache="distributedBulkSet" >
+		<int-hazelcast:request-handler-advice-chain>
+			<ref bean="testBulkSetRequestHandlerAdvice"/>
+		</int-hazelcast:request-handler-advice-chain>
+	</int-hazelcast:outbound-channel-adapter>
+
+	<bean id="testQueueRequestHandlerAdvice"
+		  class="org.springframework.integration.hazelcast.HazelcastTestRequestHandlerAdvice">
+		<constructor-arg type="int" value="100"/>
+	</bean>

 	<int-hazelcast:outbound-channel-adapter channel="queueChannel" cache="distributedQueue">
 		<int:poller fixed-delay="100"/>
 		<int-hazelcast:request-handler-advice-chain>
-			<ref bean="testRequestHandlerAdvice"/>
+			<ref bean="testQueueRequestHandlerAdvice"/>
 		</int-hazelcast:request-handler-advice-chain>
 	</int-hazelcast:outbound-channel-adapter>

+	<bean id="testBulkQueueRequestHandlerAdvice"
+		  class="org.springframework.integration.hazelcast.HazelcastTestRequestHandlerAdvice">
+		<constructor-arg type="int" value="1"/>
+	</bean>
+
+	<int-hazelcast:outbound-channel-adapter channel="bulkQueueChannel" cache="distributedBulkQueue">
+		<int:poller fixed-delay="100"/>
+		<int-hazelcast:request-handler-advice-chain>
+			<ref bean="testBulkQueueRequestHandlerAdvice"/>
+		</int-hazelcast:request-handler-advice-chain>
+	</int-hazelcast:outbound-channel-adapter>
+
+	<bean id="testTopicRequestHandlerAdvice"
+		  class="org.springframework.integration.hazelcast.HazelcastTestRequestHandlerAdvice">
+		<constructor-arg type="int" value="100" />
+	</bean>
+
+	<int-hazelcast:outbound-channel-adapter channel="topicChannel" cache="topic" >
+		<int-hazelcast:request-handler-advice-chain>
+			<ref bean="testTopicRequestHandlerAdvice"/>
+		</int-hazelcast:request-handler-advice-chain>
+	</int-hazelcast:outbound-channel-adapter>
+
+	<int-hazelcast:outbound-channel-adapter id="lockChannel" cache="myLock" />
+
 	<bean id="distributedMap" factory-bean="instance" factory-method="getMap">
 		<constructor-arg value="distributedMap"/>
 	</bean>

+	<bean id="distributedBulkMap" factory-bean="instance" factory-method="getMap">
+		<constructor-arg value="distributedBulkMap"/>
+	</bean>
+
 	<bean id="distributedList" factory-bean="instance" factory-method="getList">
 		<constructor-arg value="distributedList"/>
 	</bean>

+	<bean id="distributedBulkList" factory-bean="instance" factory-method="getList">
+		<constructor-arg value="distributedBulkList"/>
+	</bean>
+
 	<bean id="distributedSet" factory-bean="instance" factory-method="getSet">
 		<constructor-arg value="distributedSet"/>
 	</bean>

+	<bean id="distributedBulkSet" factory-bean="instance" factory-method="getSet">
+		<constructor-arg value="distributedBulkSet"/>
+	</bean>
+
 	<bean id="distributedQueue" factory-bean="instance" factory-method="getQueue">
 		<constructor-arg value="distributedQueue"/>
 	</bean>

+	<bean id="distributedBulkQueue" factory-bean="instance" factory-method="getQueue">
+		<constructor-arg value="distributedBulkQueue"/>
+	</bean>
+
+	<bean id="multiMap" factory-bean="instance" factory-method="getMultiMap">
+		<constructor-arg value="multiMap"/>
+	</bean>
+
+	<bean id="replicatedMap" factory-bean="instance" factory-method="getReplicatedMap">
+		<constructor-arg value="replicatedMap"/>
+	</bean>
+
+	<bean id="bulkReplicatedMap" factory-bean="instance" factory-method="getReplicatedMap">
+		<constructor-arg value="bulkReplicatedMap"/>
+	</bean>
+
+	<bean id="topic" factory-bean="instance" factory-method="getTopic">
+		<constructor-arg value="topic"/>
+	</bean>
+
+	<bean id="myLock" factory-bean="instance" factory-method="getLock">
+		<constructor-arg value="myLock"/>
+	</bean>
+
 	<bean id="instance" class="com.hazelcast.core.Hazelcast" factory-method="newHazelcastInstance"
 		  destroy-method="shutdown">
 		<constructor-arg>
--- 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<Integer, HazelcastIntegrationTestUser> multiMap;

 	@Resource
-	private Queue<?> distributedQueue;
+	private ReplicatedMap<Integer, HazelcastIntegrationTestUser> replicatedMap;
+
+	@Resource
+	private ReplicatedMap<Integer, HazelcastIntegrationTestUser> bulkReplicatedMap;
+
+	@Resource
+	private List<HazelcastIntegrationTestUser> distributedList;
+
+	@Resource
+	private List<HazelcastIntegrationTestUser> distributedBulkList;
+
+	@Resource
+	private Set<HazelcastIntegrationTestUser> distributedSet;
+
+	@Resource
+	private Set<HazelcastIntegrationTestUser> distributedBulkSet;
+
+	@Resource
+	private Queue<HazelcastIntegrationTestUser> distributedQueue;
+
+	@Resource
+	private Queue<HazelcastIntegrationTestUser> distributedBulkQueue;
+
+	@Resource
+	private ITopic<HazelcastIntegrationTestUser> 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<Integer, String> 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<Integer> list = (List<Integer>) fillCollectionByEntryCount(new ArrayList<Integer>());
-		listChannel.send(new GenericMessage<>(list));
-		verifyDistributedList();
+	public void testBulkWriteToDistributedMap() throws InterruptedException {
+		Map<Integer, HazelcastIntegrationTestUser> 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<Integer> set = (Set<Integer>) fillCollectionByEntryCount(new HashSet<Integer>());
-		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<Integer> queue = fillCollectionByEntryCount(new LinkedBlockingQueue<Integer>(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<Integer> 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<Integer> 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<Integer> 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<Integer> 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<Integer, String> createMapByEntryCount() {
-		Map<Integer, String> 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<Integer, HazelcastIntegrationTestUser> 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<Integer> fillCollectionByEntryCount(Collection<Integer> coll) {
-		for (int index = 0; index < DATA_COUNT; index++) {
-			coll.add(index);
+	@Test
+	public void testBulkWriteToDistributedList() throws InterruptedException {
+		List<HazelcastIntegrationTestUser> 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<HazelcastIntegrationTestUser> 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<HazelcastIntegrationTestUser> 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<HazelcastIntegrationTestUser> 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<HazelcastIntegrationTestUser> 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<HazelcastIntegrationTestUser> 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<HazelcastIntegrationTestUser> 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<Integer, HazelcastIntegrationTestUser> map) {
+		int index = 1;
+		assertNotNull(map);
+		assertEquals(true, map.size() == DATA_COUNT);
+		for (Entry<Integer, HazelcastIntegrationTestUser> entry : map.entrySet()) {
+			assertNotNull(entry);
+			assertEquals(index, entry.getKey().intValue());
+			verifyHazelcastIntegrationTestUser(entry.getValue(), index);
+			index++;
+		}
+	}
+
+	private void verifyMultiMapForPayload(
+			final MultiMap<Integer, HazelcastIntegrationTestUser> multiMap) {
+		int index = 1;
+		assertNotNull(multiMap);
+		assertEquals(true, multiMap.size() == DATA_COUNT);
+		SortedSet<Integer> 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<Integer, Message<HazelcastIntegrationTestUser>> map) {
+		int index = 1;
+		assertNotNull(map);
+		assertEquals(true, map.size() == DATA_COUNT);
+		for (Entry<Integer, Message<HazelcastIntegrationTestUser>> 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<HazelcastIntegrationTestUser> 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++;
 		}

 	}