INTEXT-142: Add Spring Integration Hazelcast

JIRA: https://jira.spring.io/browse/INTEXT-142

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/common/CacheEventType.java

@@ -0,0 +1,30 @@
+/*
+ * 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.common;
+
+/**
+ * Enumeration of Cache Event Types
+ *
+ * @author Eren Avsarogullari
+ * @since 1.0.0
+ * @see org.springframework.integration.hazelcast.inbound.AbstractHazelcastMessageProducer
+ */
+public enum CacheEventType {
+
+	ADDED, REMOVED, UPDATED, EVICTED, EVICT_ALL, CLEAR_ALL;
+
+}

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/common/CacheListeningPolicyType.java

@@ -0,0 +1,30 @@
+/*
+ * 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.common;
+
+/**
+ * Enumeration of Cache Listening Policy Type
+ *
+ * @author Eren Avsarogullari
+ * @since 1.0.0
+ * @see org.springframework.integration.hazelcast.inbound.AbstractHazelcastMessageProducer
+ */
+public enum CacheListeningPolicyType {
+
+	SINGLE, ALL;
+
+}

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/common/DistributedSQLIterationType.java

@@ -0,0 +1,30 @@
+/*
+ * 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.common;
+
+/**
+ * Enumeration of Distributed SQL Iteration Type
+ *
+ * @author Eren Avsarogullari
+ * @since 1.0.0
+ * @see org.springframework.integration.hazelcast.inbound.HazelcastDistributedSQLMessageSource
+ */
+public enum DistributedSQLIterationType {
+
+	ENTRY, KEY, LOCAL_KEY, VALUE
+
+}

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/common/HazelcastIntegrationDefinitionValidator.java

@@ -0,0 +1,112 @@
+/*
+ * 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.common;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Set;
+
+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;
+import com.hazelcast.core.ReplicatedMap;
+
+import reactor.util.CollectionUtils;
+import reactor.util.StringUtils;
+
+/**
+ * Common Validator for Hazelcast Integration. It validates cache types and events.
+ *
+ * @author Eren Avsarogullari
+ * @since 1.0.0
+ */
+public class HazelcastIntegrationDefinitionValidator {
+
+	public static <E extends Enum<E>> void validateEnumType(final Class<E> enumType, final String cacheEventTypes) {
+		Set<String> eventTypeSet = StringUtils.commaDelimitedListToSet(cacheEventTypes);
+		for (String eventType : eventTypeSet) {
+			Enum.valueOf(enumType, eventType);
+		}
+	}
+
+	public static void validateCacheTypeForEventDrivenMessageProducer(final DistributedObject distributedObject) {
+		if (!(distributedObject instanceof IMap
+				|| distributedObject instanceof MultiMap
+				|| distributedObject instanceof ReplicatedMap
+				|| distributedObject instanceof IList
+				|| distributedObject instanceof ISet
+				|| distributedObject instanceof IQueue
+				|| distributedObject instanceof ITopic)) {
+			throw new IllegalArgumentException(
+					"Invalid 'cache' type is set. IMap, MultiMap, ReplicatedMap, IList, ISet, IQueue and ITopic" +
+							" cache object types are acceptable for Hazelcast Inbound Channel Adapter.");
+		}
+	}
+
+	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);
+		if (!CollectionUtils.isEmpty(supportedCacheEventTypes)) {
+			validateCacheEventsByDistributedObject(distributedObject, cacheEventTypeSet, supportedCacheEventTypes);
+		}
+	}
+
+	private static List<String> getSupportedCacheEventTypes(final DistributedObject distributedObject) {
+		if ((distributedObject instanceof IList)
+				|| (distributedObject instanceof ISet)
+				|| (distributedObject instanceof IQueue)) {
+			return Arrays.asList(CacheEventType.ADDED.toString(), CacheEventType.REMOVED.toString());
+		}
+		else if (distributedObject instanceof MultiMap) {
+			return Arrays.asList(CacheEventType.ADDED.toString(),
+					CacheEventType.REMOVED.toString(),
+					CacheEventType.CLEAR_ALL.toString());
+		}
+		else if (distributedObject instanceof ReplicatedMap) {
+			return Arrays.asList(CacheEventType.ADDED.toString(),
+					CacheEventType.REMOVED.toString(),
+					CacheEventType.UPDATED.toString(),
+					CacheEventType.EVICTED.toString());
+		}
+
+		return null;
+	}
+
+	private static void validateCacheEventsByDistributedObject(DistributedObject distributedObject,
+			Set<String> cacheEventTypeSet, List<String> supportedCacheEventTypes) {
+		if (!supportedCacheEventTypes.containsAll(cacheEventTypeSet)) {
+			throw new IllegalArgumentException("'cache-events' attribute of "
+					+ distributedObject.getName() + " can be set as " + supportedCacheEventTypes);
+		}
+	}
+
+}

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/common/HazelcastLocalInstanceRegistrar.java

@@ -0,0 +1,78 @@
+/*
+ * 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.common;
+
+import java.net.SocketAddress;
+import java.util.concurrent.locks.Lock;
+
+import org.springframework.beans.factory.SmartInitializingSingleton;
+import org.springframework.integration.hazelcast.listener.HazelcastMembershipListener;
+
+import com.hazelcast.core.Hazelcast;
+import com.hazelcast.core.HazelcastInstance;
+import com.hazelcast.core.MultiMap;
+
+/**
+ * This class creates an internal configuration {@link MultiMap} to cache Hazelcast instances' socket
+ * address information which used Hazelcast event-driven inbound channel adapter(s). It
+ * also enables a Hazelcast {@link com.hazelcast.core.MembershipListener} to listen for
+ * membership updates.
+ *
+ * @author Eren Avsarogullari
+ * @since 1.0.0
+ */
+public class HazelcastLocalInstanceRegistrar implements SmartInitializingSingleton {
+
+	public static final String SPRING_INTEGRATION_INTERNAL_CLUSTER_MULTIMAP =
+			"SPRING_INTEGRATION_INTERNAL_CLUSTER_MULTIMAP";
+
+	public static final String SPRING_INTEGRATION_INTERNAL_CLUSTER_LOCK = "SPRING_INTEGRATION_INTERNAL_CLUSTER_LOCK";
+
+	@Override
+	public void afterSingletonsInstantiated() {
+		if (!Hazelcast.getAllHazelcastInstances().isEmpty()) {
+			HazelcastInstance hazelcastInstance = Hazelcast.getAllHazelcastInstances().iterator().next();
+			hazelcastInstance.getCluster().addMembershipListener(new HazelcastMembershipListener());
+			syncConfigurationMultiMap(hazelcastInstance);
+		}
+		else {
+			throw new IllegalStateException("No Active Local Hazelcast Instance found.");
+		}
+	}
+
+	private void syncConfigurationMultiMap(HazelcastInstance hazelcastInstance) {
+		Lock lock = hazelcastInstance.getLock(SPRING_INTEGRATION_INTERNAL_CLUSTER_LOCK);
+		lock.lock();
+		try {
+			MultiMap<SocketAddress, SocketAddress> multiMap = hazelcastInstance
+					.getMultiMap(SPRING_INTEGRATION_INTERNAL_CLUSTER_MULTIMAP);
+			for (HazelcastInstance localInstance : Hazelcast.getAllHazelcastInstances()) {
+				SocketAddress localInstanceSocketAddress = localInstance.getLocalEndpoint().getSocketAddress();
+				if (multiMap.size() == 0) {
+					multiMap.put(localInstanceSocketAddress, localInstanceSocketAddress);
+				}
+				else {
+					multiMap.put(multiMap.keySet().iterator().next(), localInstanceSocketAddress);
+				}
+			}
+		}
+		finally {
+			lock.unlock();
+		}
+	}
+
+}

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

@@ -0,0 +1,4 @@
+/**
+ * Provides common used types and classes.
+ */
+package org.springframework.integration.hazelcast.common;

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/config/HazelcastIntegrationConfigurationInitializer.java

@@ -0,0 +1,46 @@
+/*
+ * 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.config;
+
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
+import org.springframework.beans.factory.support.BeanDefinitionRegistry;
+import org.springframework.beans.factory.support.RootBeanDefinition;
+import org.springframework.integration.config.IntegrationConfigurationInitializer;
+import org.springframework.integration.hazelcast.common.HazelcastLocalInstanceRegistrar;
+
+/**
+ * The Hazelcast Integration infrastructure {@code beanFactory} initializer.
+ *
+ * @author Eren Avsarogullari
+ * @since 1.0.0
+ */
+public class HazelcastIntegrationConfigurationInitializer implements IntegrationConfigurationInitializer {
+
+	private static final String HAZELCAST_LOCAL_INSTANCE_REGISTRAR_BEAN_NAME =
+			HazelcastLocalInstanceRegistrar.class.getName();
+
+	@Override
+	public void initialize(ConfigurableListableBeanFactory beanFactory) throws BeansException {
+		BeanDefinitionRegistry beanDefinitionRegistry = (BeanDefinitionRegistry) beanFactory;
+		if (!beanDefinitionRegistry.containsBeanDefinition(HAZELCAST_LOCAL_INSTANCE_REGISTRAR_BEAN_NAME)) {
+			beanDefinitionRegistry.registerBeanDefinition(HAZELCAST_LOCAL_INSTANCE_REGISTRAR_BEAN_NAME,
+					new RootBeanDefinition(HazelcastLocalInstanceRegistrar.class));
+		}
+	}
+
+}

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

@@ -0,0 +1,4 @@
+/**
+ * Provides classes for configuration.
+ */
+package org.springframework.integration.hazelcast.config;

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/config/xml/HazelcastContinuousQueryInboundChannelAdapterParser.java

@@ -0,0 +1,109 @@
+/*
+ * 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.config.xml;
+
+import org.w3c.dom.Element;
+
+import org.springframework.beans.factory.BeanDefinitionStoreException;
+import org.springframework.beans.factory.support.AbstractBeanDefinition;
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
+import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
+import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
+import org.springframework.integration.hazelcast.inbound.HazelcastContinuousQueryMessageProducer;
+
+import reactor.util.StringUtils;
+
+/**
+ * Hazelcast Continuous Query Inbound Channel Adapter Parser parses
+ * {@code <int-hazelcast:cq-inbound-channel-adapter/>} configuration.
+ *
+ * @author Eren Avsarogullari
+ * @since 1.0.0
+ */
+public class HazelcastContinuousQueryInboundChannelAdapterParser extends AbstractSingleBeanDefinitionParser {
+
+	private static final String CHANNEL_ATTRIBUTE = "channel";
+
+	private static final String CACHE_ATTRIBUTE = "cache";
+
+	private static final String CACHE_EVENTS_ATTRIBUTE = "cache-events";
+
+	private static final String PREDICATE_ATTRIBUTE = "predicate";
+
+	private static final String INCLUDE_VALUE_ATTRIBUTE = "include-value";
+
+	private static final String CACHE_LISTENING_POLICY_ATTRIBUTE = "cache-listening-policy";
+
+	private static final String OUTPUT_CHANNEL = "outputChannel";
+
+	private static final String CACHE_EVENT_TYPES = "cacheEventTypes";
+
+	@Override
+	protected Class<?> getBeanClass(Element element) {
+		return HazelcastContinuousQueryMessageProducer.class;
+	}
+
+	@Override
+	protected String resolveId(Element element, AbstractBeanDefinition definition, ParserContext parserContext)
+			throws BeanDefinitionStoreException {
+		String id = super.resolveId(element, definition, parserContext);
+
+		if (!element.hasAttribute(CHANNEL_ATTRIBUTE)) {
+			id = id + ".adapter";
+		}
+
+		if (!StringUtils.hasText(id)) {
+			id = BeanDefinitionReaderUtils.generateBeanName(definition, parserContext.getRegistry());
+		}
+
+		return id;
+	}
+
+	@Override
+	protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
+		String channelName = element.getAttribute(CHANNEL_ATTRIBUTE);
+		if (!StringUtils.hasText(channelName)) {
+			channelName = IntegrationNamespaceUtils.createDirectChannel(element, parserContext);
+		}
+
+		if (!StringUtils.hasText(element.getAttribute(CACHE_ATTRIBUTE))) {
+			parserContext.getReaderContext().error("'" + CACHE_ATTRIBUTE + "' attribute is required.", element);
+		}
+		else if (!StringUtils.hasText(element.getAttribute(CACHE_EVENTS_ATTRIBUTE))) {
+			parserContext.getReaderContext().error("'" + CACHE_EVENTS_ATTRIBUTE + "' attribute is required.", element);
+		}
+		else if (!StringUtils.hasText(element.getAttribute(PREDICATE_ATTRIBUTE))) {
+			parserContext.getReaderContext().error("'" + PREDICATE_ATTRIBUTE + "' attribute is required.", element);
+		}
+		else if (!StringUtils.hasText(element.getAttribute(CACHE_LISTENING_POLICY_ATTRIBUTE))) {
+			parserContext.getReaderContext().error("'" + CACHE_LISTENING_POLICY_ATTRIBUTE + "' attribute is required.",
+					element);
+		}
+
+		builder.addPropertyReference(OUTPUT_CHANNEL, channelName);
+		builder.addConstructorArgReference(element.getAttribute(CACHE_ATTRIBUTE));
+		builder.addConstructorArgValue(element.getAttribute(PREDICATE_ATTRIBUTE));
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, CACHE_EVENTS_ATTRIBUTE, CACHE_EVENT_TYPES);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, INCLUDE_VALUE_ATTRIBUTE);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, CACHE_LISTENING_POLICY_ATTRIBUTE);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IntegrationNamespaceUtils.AUTO_STARTUP);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IntegrationNamespaceUtils.PHASE);
+	}
+
+}

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/config/xml/HazelcastDistributedSQLInboundChannelAdapterParser.java

@@ -0,0 +1,69 @@
+/*
+ * 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.config.xml;
+
+import org.w3c.dom.Element;
+
+import org.springframework.beans.BeanMetadataElement;
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.config.xml.AbstractPollingInboundChannelAdapterParser;
+import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
+import org.springframework.integration.hazelcast.inbound.HazelcastDistributedSQLMessageSource;
+
+import reactor.util.StringUtils;
+
+/**
+ * Hazelcast Distributed SQL Inbound Channel Adapter Parser parses
+ * {@code <int-hazelcast:cq-inbound-channel-adapter/>} configuration.
+ *
+ * @author Eren Avsarogullari
+ * @since 1.0.0
+ */
+public class HazelcastDistributedSQLInboundChannelAdapterParser extends AbstractPollingInboundChannelAdapterParser {
+
+	private static final String CACHE_ATTRIBUTE = "cache";
+
+	private static final String DISTRIBUTED_SQL_ATTRIBUTE = "distributed-sql";
+
+	private static final String ITERATION_TYPE_ATTRIBUTE = "iteration-type";
+
+	@Override
+	protected BeanMetadataElement parseSource(Element element, ParserContext parserContext) {
+		if (!StringUtils.hasText(element.getAttribute(CACHE_ATTRIBUTE))) {
+			parserContext.getReaderContext().error("'" + CACHE_ATTRIBUTE + "' attribute is required.", element);
+		}
+		else if (!StringUtils.hasText(element.getAttribute(DISTRIBUTED_SQL_ATTRIBUTE))) {
+			parserContext.getReaderContext().error("'" + DISTRIBUTED_SQL_ATTRIBUTE + "' attribute is required.",
+					element);
+		}
+		else if (!StringUtils.hasText(element.getAttribute(ITERATION_TYPE_ATTRIBUTE))) {
+			parserContext.getReaderContext().error("'" + ITERATION_TYPE_ATTRIBUTE + "' attribute is required.",
+					element);
+		}
+
+		BeanDefinitionBuilder builder = BeanDefinitionBuilder
+				.genericBeanDefinition(HazelcastDistributedSQLMessageSource.class.getName());
+
+		builder.addConstructorArgReference(element.getAttribute(CACHE_ATTRIBUTE));
+		builder.addConstructorArgValue(element.getAttribute(DISTRIBUTED_SQL_ATTRIBUTE));
+
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, ITERATION_TYPE_ATTRIBUTE);
+
+		return builder.getBeanDefinition();
+	}
+}

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/config/xml/HazelcastEventDrivenInboundChannelAdapterParser.java

@@ -0,0 +1,100 @@
+/*
+ * 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.config.xml;
+
+import org.w3c.dom.Element;
+
+import org.springframework.beans.factory.BeanDefinitionStoreException;
+import org.springframework.beans.factory.support.AbstractBeanDefinition;
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
+import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
+import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
+import org.springframework.integration.hazelcast.inbound.HazelcastEventDrivenMessageProducer;
+
+import reactor.util.StringUtils;
+
+/**
+ * Hazelcast Event Driven Inbound Channel Adapter Parser parses
+ * {@code <int-hazelcast:inbound-channel-adapter />} configuration.
+ *
+ * @author Eren Avsarogullari
+ * @since 1.0.0
+ */
+public class HazelcastEventDrivenInboundChannelAdapterParser extends AbstractSingleBeanDefinitionParser {
+
+	private static final String CHANNEL_ATTRIBUTE = "channel";
+
+	private static final String CACHE_ATTRIBUTE = "cache";
+
+	private static final String CACHE_EVENTS_ATTRIBUTE = "cache-events";
+
+	private static final String CACHE_LISTENING_POLICY_ATTRIBUTE = "cache-listening-policy";
+
+	private static final String OUTPUT_CHANNEL = "outputChannel";
+
+	private static final String CACHE_EVENT_TYPES = "cacheEventTypes";
+
+	@Override
+	protected Class<?> getBeanClass(Element element) {
+		return HazelcastEventDrivenMessageProducer.class;
+	}
+
+	@Override
+	protected String resolveId(Element element, AbstractBeanDefinition definition, ParserContext parserContext)
+			throws BeanDefinitionStoreException {
+		String id = super.resolveId(element, definition, parserContext);
+
+		if (!element.hasAttribute(CHANNEL_ATTRIBUTE)) {
+			id = id + ".adapter";
+		}
+
+		if (!StringUtils.hasText(id)) {
+			id = BeanDefinitionReaderUtils.generateBeanName(definition, parserContext.getRegistry());
+		}
+
+		return id;
+	}
+
+	@Override
+	protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
+		String channelName = element.getAttribute(CHANNEL_ATTRIBUTE);
+		if (!StringUtils.hasText(channelName)) {
+			channelName = IntegrationNamespaceUtils.createDirectChannel(element, parserContext);
+		}
+
+		if (!StringUtils.hasText(element.getAttribute(CACHE_ATTRIBUTE))) {
+			parserContext.getReaderContext().error("'" + CACHE_ATTRIBUTE + "' attribute is required.", element);
+		}
+		else if (!StringUtils.hasText(element.getAttribute(CACHE_EVENTS_ATTRIBUTE))) {
+			parserContext.getReaderContext().error("'" + CACHE_EVENTS_ATTRIBUTE + "' attribute is required.", element);
+		}
+		else if (!StringUtils.hasText(element.getAttribute(CACHE_LISTENING_POLICY_ATTRIBUTE))) {
+			parserContext.getReaderContext().error("'" + CACHE_LISTENING_POLICY_ATTRIBUTE + "' attribute is required.",
+					element);
+		}
+
+		builder.addPropertyReference(OUTPUT_CHANNEL, channelName);
+		builder.addConstructorArgReference(element.getAttribute(CACHE_ATTRIBUTE));
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, CACHE_EVENTS_ATTRIBUTE, CACHE_EVENT_TYPES);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, CACHE_LISTENING_POLICY_ATTRIBUTE);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IntegrationNamespaceUtils.AUTO_STARTUP);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IntegrationNamespaceUtils.PHASE);
+	}
+
+}

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/config/xml/HazelcastIntegrationNamespaceHandler.java

@@ -0,0 +1,37 @@
+/*
+ * 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.config.xml;
+
+import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHandler;
+
+/**
+ * Namespace handler for the Hazelcast schema.
+ *
+ * @author Eren Avsarogullari
+ * @since 1.0.0
+ */
+public class HazelcastIntegrationNamespaceHandler extends AbstractIntegrationNamespaceHandler {
+
+	@Override
+	public void init() {
+		registerBeanDefinitionParser("inbound-channel-adapter", new HazelcastEventDrivenInboundChannelAdapterParser());
+		registerBeanDefinitionParser("outbound-channel-adapter", new HazelcastOutboundChannelAdapterParser());
+		registerBeanDefinitionParser("cq-inbound-channel-adapter", new HazelcastContinuousQueryInboundChannelAdapterParser());
+		registerBeanDefinitionParser("ds-inbound-channel-adapter", new HazelcastDistributedSQLInboundChannelAdapterParser());
+	}
+
+}

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/config/xml/HazelcastOutboundChannelAdapterParser.java

@@ -0,0 +1,53 @@
+/*
+ * 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.config.xml;
+
+import org.w3c.dom.Element;
+
+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.hazelcast.outbound.HazelcastCacheWritingMessageHandler;
+
+import reactor.util.StringUtils;
+
+/**
+ * Hazelcast Outbound Channel Adapter Parser for {@code <int-hazelcast:inbound-channel-adapter />}.
+ *
+ * @author Eren Avsarogullari
+ * @since 1.0.0
+ */
+public class HazelcastOutboundChannelAdapterParser extends AbstractOutboundChannelAdapterParser {
+
+	private static final String CACHE_ATTRIBUTE = "cache";
+
+	@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);
+		}
+
+		builder.addConstructorArgReference(element.getAttribute(CACHE_ATTRIBUTE));
+
+		return builder.getBeanDefinition();
+	}
+
+}

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

@@ -0,0 +1,4 @@
+/**
+ * Provides classes for parsers and namespace handlers.
+ */
+package org.springframework.integration.hazelcast.config.xml;

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/inbound/AbstractHazelcastMessageProducer.java

@@ -0,0 +1,191 @@
+/*
+ * 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.inbound;
+
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.util.Collections;
+import java.util.EventObject;
+import java.util.HashSet;
+import java.util.Set;
+
+import org.springframework.integration.endpoint.MessageProducerSupport;
+import org.springframework.integration.hazelcast.common.CacheEventType;
+import org.springframework.integration.hazelcast.common.CacheListeningPolicyType;
+import org.springframework.integration.hazelcast.common.HazelcastIntegrationDefinitionValidator;
+import org.springframework.integration.hazelcast.common.HazelcastLocalInstanceRegistrar;
+import org.springframework.util.Assert;
+
+import com.hazelcast.core.AbstractIMapEvent;
+import com.hazelcast.core.DistributedObject;
+import com.hazelcast.core.EntryEvent;
+import com.hazelcast.core.EntryListener;
+import com.hazelcast.core.Hazelcast;
+import com.hazelcast.core.HazelcastInstance;
+import com.hazelcast.core.MapEvent;
+import com.hazelcast.core.MultiMap;
+
+import reactor.util.StringUtils;
+
+/**
+ * Hazelcast Base Event-Driven Message Producer.
+ *
+ * @author Eren Avsarogullari
+ * @since 1.0.0
+ */
+public abstract class AbstractHazelcastMessageProducer extends MessageProducerSupport {
+
+	protected final DistributedObject distributedObject;
+
+	private CacheListeningPolicyType cacheListeningPolicy = CacheListeningPolicyType.SINGLE;
+
+	private String hazelcastRegisteredEventListenerId;
+
+	private Set<String> cacheEvents = Collections.singleton(CacheEventType.ADDED.name());
+
+	protected AbstractHazelcastMessageProducer(DistributedObject distributedObject) {
+		Assert.notNull(distributedObject, "cache must not be null");
+		this.distributedObject = distributedObject;
+	}
+
+	protected Set<String> getCacheEvents() {
+		return cacheEvents;
+	}
+
+	public void setCacheEventTypes(String cacheEventTypes) {
+		HazelcastIntegrationDefinitionValidator.validateEnumType(CacheEventType.class, cacheEventTypes);
+		final Set<String> cacheEvents = StringUtils.commaDelimitedListToSet(cacheEventTypes);
+		Assert.notEmpty(cacheEvents, "cacheEvents must have elements");
+		HazelcastIntegrationDefinitionValidator.validateCacheEventsByDistributedObject(
+				this.distributedObject, cacheEvents);
+		this.cacheEvents = cacheEvents;
+	}
+
+	protected CacheListeningPolicyType getCacheListeningPolicy() {
+		return cacheListeningPolicy;
+	}
+
+	public void setCacheListeningPolicy(CacheListeningPolicyType cacheListeningPolicy) {
+		Assert.notNull(cacheListeningPolicy, "cacheListeningPolicy must not be null");
+		this.cacheListeningPolicy = cacheListeningPolicy;
+	}
+
+	protected String getHazelcastRegisteredEventListenerId() {
+		return hazelcastRegisteredEventListenerId;
+	}
+
+	protected void setHazelcastRegisteredEventListenerId(String hazelcastRegisteredEventListenerId) {
+		this.hazelcastRegisteredEventListenerId = hazelcastRegisteredEventListenerId;
+	}
+
+	protected abstract class AbstractHazelcastEventListener {
+
+		protected abstract void processEvent(EventObject event);
+
+		protected void sendMessage(final EventObject event, final InetSocketAddress socketAddress,
+				final CacheListeningPolicyType cacheListeningPolicyType) {
+			if (CacheListeningPolicyType.ALL == cacheListeningPolicyType || isEventAcceptable(socketAddress)) {
+				AbstractHazelcastMessageProducer.this.sendMessage(getMessageBuilderFactory().withPayload(event).build());
+			}
+		}
+
+		private boolean isEventAcceptable(final InetSocketAddress socketAddress) {
+			final Set<HazelcastInstance> hazelcastInstanceSet = Hazelcast.getAllHazelcastInstances();
+			final Set<SocketAddress> localSocketAddressesSet = getLocalSocketAddresses(hazelcastInstanceSet);
+			if ((!localSocketAddressesSet.isEmpty())
+					&& (localSocketAddressesSet.contains(socketAddress) ||
+					isEventComingFromNonRegisteredHazelcastInstance(hazelcastInstanceSet.iterator().next(),
+							localSocketAddressesSet, socketAddress))) {
+				return true;
+			}
+
+			return false;
+		}
+
+		private Set<SocketAddress> getLocalSocketAddresses(final Set<HazelcastInstance> hazelcastInstanceSet) {
+			final Set<SocketAddress> localSocketAddressesSet = new HashSet<>();
+			for (HazelcastInstance hazelcastInstance : hazelcastInstanceSet) {
+				localSocketAddressesSet.add(hazelcastInstance.getLocalEndpoint().getSocketAddress());
+			}
+
+			return localSocketAddressesSet;
+		}
+
+		private boolean isEventComingFromNonRegisteredHazelcastInstance(
+				final HazelcastInstance hazelcastInstance,
+				final Set<SocketAddress> localSocketAddressesSet,
+				final InetSocketAddress socketAddressOfEvent) {
+			final MultiMap<SocketAddress, SocketAddress> configMultiMap = hazelcastInstance
+					.getMultiMap(HazelcastLocalInstanceRegistrar.SPRING_INTEGRATION_INTERNAL_CLUSTER_MULTIMAP);
+			return configMultiMap.size() > 0
+					&& !configMultiMap.values().contains(socketAddressOfEvent)
+					&& localSocketAddressesSet.contains(configMultiMap.keySet().iterator().next());
+
+		}
+
+	}
+
+	protected final class HazelcastEntryListener<K, V> extends
+			AbstractHazelcastEventListener implements EntryListener<K, V> {
+
+		@Override
+		public void entryAdded(EntryEvent<K, V> event) {
+			processEvent(event);
+		}
+
+		@Override
+		public void entryRemoved(EntryEvent<K, V> event) {
+			processEvent(event);
+		}
+
+		@Override
+		public void entryUpdated(EntryEvent<K, V> event) {
+			processEvent(event);
+		}
+
+		@Override
+		public void entryEvicted(EntryEvent<K, V> event) {
+			processEvent(event);
+		}
+
+		@Override
+		public void mapEvicted(MapEvent event) {
+			processEvent(event);
+		}
+
+		@Override
+		public void mapCleared(MapEvent event) {
+			processEvent(event);
+		}
+
+		@Override
+		protected void processEvent(EventObject event) {
+			Assert.notNull(event, "event must not be null");
+
+			if (getCacheEvents().contains(((AbstractIMapEvent) event).getEventType().toString())) {
+				sendMessage(event, ((AbstractIMapEvent) event).getMember().getSocketAddress(),
+						getCacheListeningPolicy());
+			}
+
+			if (logger.isDebugEnabled()) {
+				logger.debug("Received Event : " + event);
+			}
+		}
+
+	}
+
+}

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/inbound/HazelcastContinuousQueryMessageProducer.java

@@ -0,0 +1,67 @@
+/*
+ * 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.inbound;
+
+import org.springframework.util.Assert;
+
+import com.hazelcast.core.IMap;
+import com.hazelcast.query.SqlPredicate;
+
+/**
+ * Hazelcast Continuous Query Message Producer is a message producer which enables
+ * {@link AbstractHazelcastMessageProducer.HazelcastEntryListener} with a
+ * {@link SqlPredicate} in order to listen related distributed map events in the light of
+ * defined predicate and sends events to related channel.
+ *
+ * @author Eren Avsarogullari
+ * @since 1.0.0
+ */
+public class HazelcastContinuousQueryMessageProducer extends AbstractHazelcastMessageProducer {
+
+	private final String predicate;
+
+	private boolean includeValue = true;
+
+	@SuppressWarnings("rawtypes")
+	public HazelcastContinuousQueryMessageProducer(IMap distributedMap, String predicate) {
+		super(distributedMap);
+		Assert.hasText(predicate, "predicate must not be null");
+		this.predicate = predicate;
+	}
+
+	public void setIncludeValue(boolean includeValue) {
+		this.includeValue = includeValue;
+	}
+
+	@SuppressWarnings({"rawtypes", "unchecked"})
+	@Override
+	protected void doStart() {
+		setHazelcastRegisteredEventListenerId(((IMap<?, ?>) this.distributedObject)
+				.addEntryListener(new HazelcastEntryListener(), new SqlPredicate(this.predicate), this.includeValue));
+	}
+
+	@Override
+	protected void doStop() {
+		((IMap<?, ?>) this.distributedObject).removeEntryListener(getHazelcastRegisteredEventListenerId());
+	}
+
+	@Override
+	public String getComponentType() {
+		return "hazelcast:cq-inbound-channel-adapter";
+	}
+
+}

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/inbound/HazelcastDistributedSQLMessageSource.java

@@ -0,0 +1,91 @@
+/*
+ * 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.inbound;
+
+import java.util.Collection;
+import java.util.Collections;
+
+import org.springframework.integration.endpoint.AbstractMessageSource;
+import org.springframework.integration.hazelcast.common.DistributedSQLIterationType;
+import org.springframework.util.Assert;
+import org.springframework.util.CollectionUtils;
+
+import com.hazelcast.core.IMap;
+import com.hazelcast.query.SqlPredicate;
+
+/**
+ * Hazelcast Distributed SQL Message Source is a message source which runs defined
+ * distributed query in the cluster and returns results in the light of iteration type.
+ *
+ * @author Eren Avsarogullari
+ * @since 1.0.0
+ */
+@SuppressWarnings("rawtypes")
+public class HazelcastDistributedSQLMessageSource extends AbstractMessageSource {
+
+	private final IMap<?, ?> distributedMap;
+
+	private final String distributedSQL;
+
+	private DistributedSQLIterationType iterationType = DistributedSQLIterationType.VALUE;
+
+	public HazelcastDistributedSQLMessageSource(IMap distributedMap, String distributedSQL) {
+		Assert.notNull(distributedMap, "cache must not be null");
+		Assert.hasText(distributedSQL, "distributed-sql must not be null");
+		this.distributedMap = distributedMap;
+		this.distributedSQL = distributedSQL;
+	}
+
+	public void setIterationType(DistributedSQLIterationType iterationType) {
+		Assert.notNull(this.iterationType, "iterationType must not be null");
+		this.iterationType = iterationType;
+	}
+
+	@Override
+	public String getComponentType() {
+		return "hazelcast:ds-inbound-channel-adapter";
+	}
+
+	@Override
+	protected Collection<?> doReceive() {
+		switch (this.iterationType) {
+			case ENTRY:
+				return getDistributedSQLResultSet(Collections
+						.unmodifiableCollection(this.distributedMap.entrySet(new SqlPredicate(this.distributedSQL))));
+
+			case KEY:
+				return getDistributedSQLResultSet(Collections
+						.unmodifiableCollection(this.distributedMap.keySet(new SqlPredicate(this.distributedSQL))));
+
+			case LOCAL_KEY:
+				return getDistributedSQLResultSet(Collections
+						.unmodifiableCollection(this.distributedMap.localKeySet(new SqlPredicate(this.distributedSQL))));
+
+			default:
+				return getDistributedSQLResultSet(this.distributedMap.values(new SqlPredicate(this.distributedSQL)));
+		}
+	}
+
+	private Collection<?> getDistributedSQLResultSet(Collection<?> collection) {
+		if (CollectionUtils.isEmpty(collection)) {
+			return null;
+		}
+
+		return collection;
+	}
+
+}

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/inbound/HazelcastEventDrivenMessageProducer.java

@@ -0,0 +1,168 @@
+/*
+ * 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.inbound;
+
+import java.util.EventObject;
+
+import org.springframework.integration.hazelcast.common.HazelcastIntegrationDefinitionValidator;
+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.ItemEvent;
+import com.hazelcast.core.ItemListener;
+import com.hazelcast.core.Message;
+import com.hazelcast.core.MessageListener;
+import com.hazelcast.core.MultiMap;
+import com.hazelcast.core.ReplicatedMap;
+
+/**
+ * Hazelcast Event Driven Message Producer is a message producer which enables
+ * {@link AbstractHazelcastMessageProducer.HazelcastEntryListener},
+ * {@link HazelcastEventDrivenMessageProducer.HazelcastItemListener} and
+ * {@link HazelcastEventDrivenMessageProducer.HazelcastMessageListener} listeners in order
+ * to listen related cache events and sends events to related channel.
+ *
+ * @author Eren Avsarogullari
+ * @since 1.0.0
+ */
+@SuppressWarnings({ "unchecked", "rawtypes" })
+public class HazelcastEventDrivenMessageProducer extends AbstractHazelcastMessageProducer {
+
+	public HazelcastEventDrivenMessageProducer(DistributedObject distributedObject) {
+		super(distributedObject);
+	}
+
+	@Override
+	protected void onInit() {
+		super.onInit();
+		HazelcastIntegrationDefinitionValidator.validateCacheTypeForEventDrivenMessageProducer(this.distributedObject);
+	}
+
+	@Override
+	protected void doStart() {
+		if(this.distributedObject instanceof IMap) {
+			setHazelcastRegisteredEventListenerId(((IMap<?, ?>) this.distributedObject)
+					.addEntryListener(new HazelcastEntryListener(), true));
+		}
+		else if(this.distributedObject instanceof MultiMap) {
+			setHazelcastRegisteredEventListenerId(((MultiMap<?, ?>) this.distributedObject)
+					.addEntryListener(new HazelcastEntryListener(), true));
+		}
+		else if(this.distributedObject instanceof ReplicatedMap) {
+			setHazelcastRegisteredEventListenerId(((ReplicatedMap<?, ?>) this.distributedObject)
+					.addEntryListener(new HazelcastEntryListener()));
+		}
+		else if(this.distributedObject instanceof IList) {
+			setHazelcastRegisteredEventListenerId(((IList<?>) this.distributedObject)
+					.addItemListener(new HazelcastItemListener(), true));
+		}
+		else if(this.distributedObject instanceof ISet) {
+			setHazelcastRegisteredEventListenerId(((ISet<?>) this.distributedObject)
+					.addItemListener(new HazelcastItemListener(), true));
+		}
+		else if(this.distributedObject instanceof IQueue) {
+			setHazelcastRegisteredEventListenerId(((IQueue<?>) this.distributedObject)
+					.addItemListener(new HazelcastItemListener(), true));
+		}
+		else if(this.distributedObject instanceof ITopic) {
+			setHazelcastRegisteredEventListenerId(((ITopic<?>) this.distributedObject)
+					.addMessageListener(new HazelcastMessageListener()));
+		}
+	}
+
+	@Override
+	protected void doStop() {
+		if(this.distributedObject instanceof IMap) {
+			((IMap<?, ?>) this.distributedObject).removeEntryListener(getHazelcastRegisteredEventListenerId());
+		}
+		else if(this.distributedObject instanceof MultiMap) {
+			((MultiMap<?, ?>) this.distributedObject).removeEntryListener(getHazelcastRegisteredEventListenerId());
+		}
+		else if(this.distributedObject instanceof ReplicatedMap) {
+			((ReplicatedMap<?, ?>) this.distributedObject).removeEntryListener(getHazelcastRegisteredEventListenerId());
+		}
+		else if(this.distributedObject instanceof IList) {
+			((IList<?>) this.distributedObject).removeItemListener(getHazelcastRegisteredEventListenerId());
+		}
+		else if(this.distributedObject instanceof ISet) {
+			((ISet<?>) this.distributedObject).removeItemListener(getHazelcastRegisteredEventListenerId());
+		}
+		else if(this.distributedObject instanceof IQueue) {
+			((IQueue<?>) this.distributedObject).removeItemListener(getHazelcastRegisteredEventListenerId());
+		}
+		else if(this.distributedObject instanceof ITopic) {
+			((ITopic<?>) this.distributedObject).removeMessageListener(getHazelcastRegisteredEventListenerId());
+		}
+	}
+
+	@Override
+	public String getComponentType() {
+		return "hazelcast:inbound-channel-adapter";
+	}
+
+	private class HazelcastItemListener<E> extends AbstractHazelcastEventListener implements ItemListener<E> {
+
+		@Override
+		public void itemAdded(ItemEvent<E> item) {
+			processEvent(item);
+		}
+
+		@Override
+		public void itemRemoved(ItemEvent<E> item) {
+			processEvent(item);
+		}
+
+		@Override
+		protected void processEvent(EventObject event) {
+			Assert.notNull(event, "event must not be null");
+
+			if (getCacheEvents().contains(((ItemEvent<E>) event).getEventType().toString())) {
+				sendMessage(event, ((ItemEvent<E>) event).getMember().getSocketAddress(), getCacheListeningPolicy());
+			}
+
+			if (logger.isDebugEnabled()){
+				logger.debug("Received ItemEvent : " + event);
+			}
+		}
+
+	}
+
+	private class HazelcastMessageListener<E> extends AbstractHazelcastEventListener implements MessageListener<E> {
+
+		@Override
+		public void onMessage(Message<E> message) {
+			processEvent(message);
+		}
+
+		@Override
+		protected void processEvent(EventObject event) {
+			Assert.notNull(event, "event must not be null");
+			sendMessage(event, ((Message<E>) event).getPublishingMember().getSocketAddress(), null);
+
+			if (logger.isDebugEnabled()){
+				logger.debug("Received Message : " + event);
+			}
+		}
+
+	}
+
+}

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

@@ -0,0 +1,4 @@
+/**
+ * Provides classes supporting inbound endpoints.
+ */
+package org.springframework.integration.hazelcast.inbound;

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/listener/HazelcastMembershipListener.java

@@ -0,0 +1,83 @@
+/*
+ * 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.listener;
+
+import java.net.SocketAddress;
+import java.util.Set;
+import java.util.concurrent.locks.Lock;
+
+import org.springframework.integration.hazelcast.common.HazelcastLocalInstanceRegistrar;
+
+import com.hazelcast.core.Hazelcast;
+import com.hazelcast.core.HazelcastInstance;
+import com.hazelcast.core.MembershipAdapter;
+import com.hazelcast.core.MembershipEvent;
+import com.hazelcast.core.MultiMap;
+
+/**
+ * Hazelcast {@link MembershipAdapter} in order to listen for membership updates in the cluster.
+ *
+ * @author Eren Avsarogullari
+ * @since 1.0.0
+ */
+public class HazelcastMembershipListener extends MembershipAdapter {
+
+	@Override
+	public void memberRemoved(MembershipEvent membershipEvent) {
+		SocketAddress removedMemberSocketAddress = membershipEvent.getMember().getSocketAddress();
+		Set<HazelcastInstance> hazelcastLocalInstanceSet = Hazelcast.getAllHazelcastInstances();
+		if (!hazelcastLocalInstanceSet.isEmpty()) {
+			HazelcastInstance hazelcastInstance = hazelcastLocalInstanceSet.iterator().next();
+			Lock lock = hazelcastInstance
+					.getLock(HazelcastLocalInstanceRegistrar.SPRING_INTEGRATION_INTERNAL_CLUSTER_LOCK);
+			lock.lock();
+			try {
+				MultiMap<SocketAddress, SocketAddress> configMultiMap = hazelcastInstance
+						.getMultiMap(HazelcastLocalInstanceRegistrar.SPRING_INTEGRATION_INTERNAL_CLUSTER_MULTIMAP);
+
+				if (configMultiMap.containsKey(removedMemberSocketAddress)) {
+					SocketAddress newAdminSocketAddress = getNewAdminInstanceSocketAddress(
+							configMultiMap, removedMemberSocketAddress);
+					for (SocketAddress socketAddress : configMultiMap.values()) {
+						if (!socketAddress.equals(removedMemberSocketAddress)) {
+							configMultiMap.put(newAdminSocketAddress, socketAddress);
+						}
+					}
+					configMultiMap.remove(removedMemberSocketAddress);
+				}
+				else {
+					configMultiMap.remove(configMultiMap.keySet().iterator().next(), removedMemberSocketAddress);
+				}
+			}
+			finally {
+				lock.unlock();
+			}
+		}
+	}
+
+	private SocketAddress getNewAdminInstanceSocketAddress(
+			MultiMap<SocketAddress, SocketAddress> configMultiMap, SocketAddress removedMemberSocketAddress) {
+		for (SocketAddress socketAddress : configMultiMap.values()) {
+			if (!socketAddress.equals(removedMemberSocketAddress)) {
+				return socketAddress;
+			}
+		}
+
+		throw new IllegalStateException("No Active Hazelcast Instance Found.");
+	}
+
+}

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

@@ -0,0 +1,4 @@
+/**
+ * Provides classes for listeners.
+ */
+package org.springframework.integration.hazelcast.listener;

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/outbound/HazelcastCacheWritingMessageHandler.java

@@ -0,0 +1,80 @@
+/*
+ * 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.outbound;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
+import java.util.Set;
+
+import org.springframework.integration.handler.AbstractMessageHandler;
+import org.springframework.integration.hazelcast.common.HazelcastIntegrationDefinitionValidator;
+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;
+
+/**
+ * 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.
+ *
+ * @author Eren Avsarogullari
+ * @since 1.0.0
+ */
+public class HazelcastCacheWritingMessageHandler extends AbstractMessageHandler {
+
+	private final DistributedObject distributedObject;
+
+	public HazelcastCacheWritingMessageHandler(DistributedObject distributedObject) {
+		Assert.notNull(distributedObject, "cache must not be null");
+		this.distributedObject = distributedObject;
+	}
+
+	@Override
+	protected void onInit() throws Exception {
+		super.onInit();
+		HazelcastIntegrationDefinitionValidator.validateCacheTypeForCacheWritingMessageHandler(this.distributedObject);
+	}
+
+	@Override
+	protected void handleMessageInternal(Message<?> message) throws Exception {
+		writeToCache(message);
+	}
+
+	@SuppressWarnings({ "unchecked", "rawtypes" })
+	private void writeToCache(Message<?> message) {
+		if (this.distributedObject instanceof IMap) {
+			((IMap<?, ?>) this.distributedObject).putAll((Map) message.getPayload());
+		}
+		else if (this.distributedObject instanceof IList) {
+			((IList<?>) this.distributedObject).addAll((List) message.getPayload());
+		}
+		else if (this.distributedObject instanceof ISet) {
+			((ISet<?>) this.distributedObject).addAll((Set) message.getPayload());
+		}
+		else if (this.distributedObject instanceof IQueue) {
+			((IQueue<?>) this.distributedObject).addAll((Queue) message.getPayload());
+		}
+	}
+
+}

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

@@ -0,0 +1,4 @@
+/**
+ * Provides classes supporting outbound endpoints.
+ */
+package org.springframework.integration.hazelcast.outbound;

spring-integration-hazelcast/src/main/resources/META-INF/spring.factories

@@ -0,0 +1,2 @@
+org.springframework.integration.config.IntegrationConfigurationInitializer=\
+org.springframework.integration.hazelcast.config.HazelcastIntegrationConfigurationInitializer

spring-integration-hazelcast/src/main/resources/META-INF/spring.handlers

@@ -0,0 +1 @@
+http\://www.springframework.org/schema/integration/hazelcast=org.springframework.integration.hazelcast.config.xml.HazelcastIntegrationNamespaceHandler

spring-integration-hazelcast/src/main/resources/META-INF/spring.schemas

@@ -0,0 +1,2 @@
+http\://www.springframework.org/schema/integration/hazelcast/spring-integration-hazelcast-1.0.xsd=org/springframework/integration/hazelcast/config/xml/spring-integration-hazelcast-1.0.xsd
+http\://www.springframework.org/schema/integration/hazelcast/spring-integration-hazelcast.xsd=org/springframework/integration/hazelcast/config/xml/spring-integration-hazelcast-1.0.xsd

spring-integration-hazelcast/src/main/resources/org/springframework/integration/hazelcast/config/xml/spring-integration-hazelcast-1.0.xsd

@@ -0,0 +1,235 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<xsd:schema xmlns="http://www.springframework.org/schema/integration/hazelcast"
+	xmlns:xsd="http://www.w3.org/2001/XMLSchema"
+	xmlns:beans="http://www.springframework.org/schema/beans"
+	xmlns:tool="http://www.springframework.org/schema/tool"
+	xmlns:integration="http://www.springframework.org/schema/integration"
+	targetNamespace="http://www.springframework.org/schema/integration/hazelcast"
+	elementFormDefault="qualified"
+	attributeFormDefault="unqualified">
+	
+	<xsd:import namespace="http://www.springframework.org/schema/beans"
+				schemaLocation="http://www.springframework.org/schema/beans/spring-beans.xsd"/>
+	<xsd:import namespace="http://www.springframework.org/schema/tool"/>
+	<xsd:import namespace="http://www.springframework.org/schema/integration"
+				schemaLocation="http://www.springframework.org/schema/integration/spring-integration.xsd"/>
+
+	<xsd:element name="inbound-channel-adapter">
+	
+		<xsd:annotation>
+			<xsd:documentation>
+				Configures Hazelcast Event-Driven Inbound Channel Adapter
+			</xsd:documentation>
+		</xsd:annotation>
+		
+		<xsd:complexType>
+			<xsd:attributeGroup ref="integration:channelAdapterAttributes"/>
+			
+			<xsd:attribute name="cache" use="required" type="xsd:string">
+				<xsd:annotation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:expected-type type="com.hazelcast.core.DistributedObject" />
+						</tool:annotation>
+					</xsd:appinfo>
+					<xsd:documentation>
+						<![CDATA[ Specifies cache reference to listen ]]>
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+
+			<xsd:attribute name="cache-events" type="xsd:string" use="optional" default="ADDED">
+				<xsd:annotation>
+					<xsd:appinfo>
+						<tool:annotation kind="value">
+							<tool:expected-type
+								type="org.springframework.integration.hazelcast.common.CacheEventType" />
+						</tool:annotation>
+					</xsd:appinfo>
+					<xsd:documentation>
+						<![CDATA[ Specifies cache entry event types ]]>
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+			
+			<xsd:attribute name="cache-listening-policy" default="SINGLE" use="optional">
+				<xsd:annotation>
+					<xsd:documentation><![CDATA[ Specifies cache listening policy. ]]></xsd:documentation>
+				</xsd:annotation>
+				<xsd:simpleType>
+					<xsd:restriction base="xsd:string">
+						<xsd:enumeration value="SINGLE" />
+						<xsd:enumeration value="ALL" />
+					</xsd:restriction>
+				</xsd:simpleType>
+			</xsd:attribute>
+					
+		</xsd:complexType>
+	</xsd:element>
+	
+	<xsd:element name="outbound-channel-adapter">
+		
+		<xsd:annotation>
+			<xsd:documentation>
+				Configures Hazelcast Outbound Channel Adapter
+			</xsd:documentation>
+		</xsd:annotation>
+		
+		<xsd:complexType>
+			<xsd:all>
+				<xsd:element name="request-handler-advice-chain" type="integration:handlerAdviceChainType"
+							minOccurs="0" maxOccurs="1" />
+			</xsd:all>
+ 			
+ 			<xsd:attributeGroup ref="integration:channelAdapterAttributes"/>
+ 	
+			<xsd:attribute name="cache" use="required" type="xsd:string">
+				<xsd:annotation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:expected-type type="com.hazelcast.core.DistributedObject" />
+						</tool:annotation>
+					</xsd:appinfo>
+					<xsd:documentation>
+						<![CDATA[ Specifies cache reference to listen ]]>
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+			
+			<xsd:attribute name="order" type="xsd:string" use="optional">
+                <xsd:annotation>
+                    <xsd:documentation><![CDATA[
+                    Specifies the order for invocation when this endpoint is connected as a
+                    subscriber to a SubscribableChannel.
+                ]]></xsd:documentation>
+                </xsd:annotation>
+            </xsd:attribute>
+		
+		</xsd:complexType>
+	</xsd:element>
+	
+	<xsd:element name="cq-inbound-channel-adapter">
+	
+		<xsd:annotation>
+			<xsd:documentation>
+				Configures Hazelcast Continuous Query Inbound Channel Adapter
+			</xsd:documentation>
+		</xsd:annotation>
+		
+		<xsd:complexType>
+
+			<xsd:attributeGroup ref="integration:channelAdapterAttributes"/>
+			
+			<xsd:attribute name="cache" use="required" type="xsd:string">
+				<xsd:annotation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:expected-type type="com.hazelcast.core.IMap" />
+						</tool:annotation>
+					</xsd:appinfo>
+					<xsd:documentation>
+						<![CDATA[ Specifies cache reference to listen ]]>
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+
+			<xsd:attribute name="cache-events" type="xsd:string" use="optional" default="ADDED">
+				<xsd:annotation>
+					<xsd:appinfo>
+						<tool:annotation kind="value">
+							<tool:expected-type
+								type="org.springframework.integration.hazelcast.common.CacheEventType" />
+						</tool:annotation>
+					</xsd:appinfo>
+					<xsd:documentation>
+						<![CDATA[ Specifies cache entry event types ]]>
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+			
+			<xsd:attribute name="predicate" type="xsd:string" use="required">
+				<xsd:annotation>
+					<xsd:documentation>
+						<![CDATA[ Specifies predicate for continuous query ]]>
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+			
+			<xsd:attribute name="include-value" type="xsd:boolean" default="true">
+				<xsd:annotation>
+					<xsd:documentation>
+						<![CDATA[ Specifies including of value and oldValue in continuous query result ]]>
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+			
+			<xsd:attribute name="cache-listening-policy" default="SINGLE" use="optional">
+				<xsd:annotation>
+					<xsd:documentation><![CDATA[ Specifies cache listening policy. ]]></xsd:documentation>
+				</xsd:annotation>
+				<xsd:simpleType>
+					<xsd:restriction base="xsd:string">
+						<xsd:enumeration value="SINGLE" />
+						<xsd:enumeration value="ALL" />
+					</xsd:restriction>
+				</xsd:simpleType>
+			</xsd:attribute>
+			
+		</xsd:complexType>
+	</xsd:element>
+	
+	<xsd:element name="ds-inbound-channel-adapter">
+	
+		<xsd:annotation>
+			<xsd:documentation>
+				Configures Hazelcast Distributed SQL Inbound Channel Adapter
+			</xsd:documentation>
+		</xsd:annotation>
+		
+		<xsd:complexType>
+		
+			<xsd:sequence>
+				<xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1"/>
+			</xsd:sequence>
+			
+			<xsd:attributeGroup ref="integration:channelAdapterAttributes"/>
+
+			<xsd:attribute name="cache" use="required" type="xsd:string">
+				<xsd:annotation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:expected-type type="com.hazelcast.core.IMap" />
+						</tool:annotation>
+					</xsd:appinfo>
+					<xsd:documentation>
+						<![CDATA[ Specifies cache reference to listen ]]>
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+			
+			<xsd:attribute name="iteration-type" default="VALUE" use="optional">
+				<xsd:annotation>
+					<xsd:documentation><![CDATA[ Specifies Distributed-SQL Iteration Types. ]]></xsd:documentation>
+				</xsd:annotation>
+				<xsd:simpleType>
+					<xsd:restriction base="xsd:string">
+						<xsd:enumeration value="ENTRY" />
+						<xsd:enumeration value="KEY" />
+						<xsd:enumeration value="LOCAL_KEY" />
+						<xsd:enumeration value="VALUE" />
+					</xsd:restriction>
+				</xsd:simpleType>
+			</xsd:attribute>
+			
+			<xsd:attribute name="distributed-sql" type="xsd:string" use="required">
+				<xsd:annotation>
+					<xsd:documentation>
+						<![CDATA[ Specifies Distributed-SQL ]]>
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+		
+		</xsd:complexType>
+		
+	</xsd:element>
+</xsd:schema>