INTEXT-167: Add Cluster Monitor Channel Adapter

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

Some refactorings are done.

Hazelcast Cluster Monitor Unit Test Refactorings are done.

Some minor bug fixes are done.

Redundant assert is removed.

HazelcastInstance lifecycle check is added

Migration Event UT Case is updated.

Code style polishing and upgrade to the `spring.io.plugin-0.4.0`

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/ClusterMonitorType.java

@@ -0,0 +1,35 @@
+/*
+ * 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;
+
+/**
+ * Enumeration of Hazelcast Cluster Monitor Types
+ *
+ * @author Eren Avsarogullari
+ * @since 1.0.0
+ * @see org.springframework.integration.hazelcast.inbound.HazelcastClusterMonitorMessageProducer
+ * @see com.hazelcast.core.MembershipListener
+ * @see com.hazelcast.core.DistributedObjectListener
+ * @see com.hazelcast.core.MigrationListener
+ * @see com.hazelcast.core.LifecycleListener
+ * @see com.hazelcast.core.ClientListener
+ */
+public enum ClusterMonitorType {
+
+	MEMBERSHIP, DISTRIBUTED_OBJECT, MIGRATION, LIFECYCLE, CLIENT;
+
+}

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

@@ -40,11 +40,13 @@ import com.hazelcast.core.ReplicatedMap;
  */
 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 <E extends Enum<E>> Set<String> validateEnumType(final Class<E> enumType, final String types) {
+		Set<String> typeSet = StringUtils.commaDelimitedListToSet(types);
+		for (String type : typeSet) {
+			Enum.valueOf(enumType, type);
 		}
+
+		return typeSet;
 	}
 
 	public static void validateCacheTypeForEventDrivenMessageProducer(final DistributedObject distributedObject) {

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

@@ -0,0 +1,99 @@
+/*
+ * 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.HazelcastClusterMonitorMessageProducer;
+
+import org.springframework.util.StringUtils;
+
+/**
+ * Parser for the {@code <int-hazelcast:cm-inbound-channel-adapter />} component.
+ *
+ * @author Eren Avsarogullari
+ * @since 1.0.0
+ */
+public class HazelcastClusterMonitorInboundChannelAdapterParser extends
+		AbstractSingleBeanDefinitionParser {
+
+	private static final String CHANNEL_ATTRIBUTE = "channel";
+
+	private static final String HAZELCAST_INSTANCE_ATTRIBUTE = "hazelcast-instance";
+
+	private static final String MONITOR_TYPES_ATTRIBUTE = "monitor-types";
+
+	private static final String OUTPUT_CHANNEL = "outputChannel";
+
+	private static final String MONITOR_EVENT_TYPES = "monitorEventTypes";
+
+	@Override
+	protected Class<?> getBeanClass(Element element) {
+		return HazelcastClusterMonitorMessageProducer.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(HAZELCAST_INSTANCE_ATTRIBUTE))) {
+			parserContext.getReaderContext().error(
+					"'" + HAZELCAST_INSTANCE_ATTRIBUTE + "' attribute is required.",
+					element);
+		}
+
+		builder.addPropertyReference(OUTPUT_CHANNEL, channelName);
+		builder.addConstructorArgReference(element
+				.getAttribute(HAZELCAST_INSTANCE_ATTRIBUTE));
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
+				MONITOR_TYPES_ATTRIBUTE, MONITOR_EVENT_TYPES);
+		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

@@ -32,6 +32,7 @@ public class HazelcastIntegrationNamespaceHandler extends AbstractIntegrationNam
 		registerBeanDefinitionParser("outbound-channel-adapter", new HazelcastOutboundChannelAdapterParser());
 		registerBeanDefinitionParser("cq-inbound-channel-adapter", new HazelcastContinuousQueryInboundChannelAdapterParser());
 		registerBeanDefinitionParser("ds-inbound-channel-adapter", new HazelcastDistributedSQLInboundChannelAdapterParser());
+		registerBeanDefinitionParser("cm-inbound-channel-adapter", new HazelcastClusterMonitorInboundChannelAdapterParser());
 	}
 
 }

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

@@ -33,7 +33,6 @@ import org.springframework.integration.hazelcast.HazelcastLocalInstanceRegistrar
 import org.springframework.integration.hazelcast.message.EntryEventMessagePayload;
 import org.springframework.messaging.Message;
 import org.springframework.util.Assert;
-import org.springframework.util.StringUtils;
 
 import com.hazelcast.core.AbstractIMapEvent;
 import com.hazelcast.core.DistributedObject;
@@ -71,11 +70,11 @@ public abstract class AbstractHazelcastMessageProducer extends MessageProducerSu
 	}
 
 	public void setCacheEventTypes(String cacheEventTypes) {
-		HazelcastIntegrationDefinitionValidator.validateEnumType(CacheEventType.class, cacheEventTypes);
-		Set<String> cacheEvents = StringUtils.commaDelimitedListToSet(cacheEventTypes);
+		Set<String> cacheEvents =
+				HazelcastIntegrationDefinitionValidator.validateEnumType(CacheEventType.class, cacheEventTypes);
 		Assert.notEmpty(cacheEvents, "'cacheEvents' must have elements");
-		HazelcastIntegrationDefinitionValidator.validateCacheEventsByDistributedObject(
-				this.distributedObject, cacheEvents);
+		HazelcastIntegrationDefinitionValidator.validateCacheEventsByDistributedObject(this.distributedObject,
+				cacheEvents);
 		this.cacheEvents = cacheEvents;
 	}
 
@@ -103,7 +102,7 @@ public abstract class AbstractHazelcastMessageProducer extends MessageProducerSu
 		protected abstract Message<?> toMessage(E event);
 
 		protected void sendMessage(E event, InetSocketAddress socketAddress,
-				CacheListeningPolicyType cacheListeningPolicyType) {
+								   CacheListeningPolicyType cacheListeningPolicyType) {
 			if (CacheListeningPolicyType.ALL == cacheListeningPolicyType || isEventAcceptable(socketAddress)) {
 				AbstractHazelcastMessageProducer.this.sendMessage(toMessage(event));
 			}
@@ -178,12 +177,11 @@ public abstract class AbstractHazelcastMessageProducer extends MessageProducerSu
 		@Override
 		protected void processEvent(AbstractIMapEvent event) {
 			if (getCacheEvents().contains(event.getEventType().toString())) {
+				if (logger.isDebugEnabled()) {
+					logger.debug("Received Event : " + event);
+				}
 				sendMessage(event, event.getMember().getSocketAddress(), getCacheListeningPolicy());
 			}
-
-			if (logger.isDebugEnabled()) {
-				logger.debug("Received Event : " + event);
-			}
 		}
 
 		@Override

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

@@ -0,0 +1,210 @@
+/*
+ * 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.Collections;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.springframework.integration.endpoint.MessageProducerSupport;
+import org.springframework.integration.hazelcast.ClusterMonitorType;
+import org.springframework.integration.hazelcast.HazelcastIntegrationDefinitionValidator;
+import org.springframework.util.Assert;
+
+import com.hazelcast.core.Client;
+import com.hazelcast.core.ClientListener;
+import com.hazelcast.core.DistributedObjectEvent;
+import com.hazelcast.core.DistributedObjectListener;
+import com.hazelcast.core.HazelcastInstance;
+import com.hazelcast.core.LifecycleEvent;
+import com.hazelcast.core.LifecycleListener;
+import com.hazelcast.core.MemberAttributeEvent;
+import com.hazelcast.core.MembershipEvent;
+import com.hazelcast.core.MembershipListener;
+import com.hazelcast.core.MigrationEvent;
+import com.hazelcast.core.MigrationListener;
+
+/**
+ * Hazelcast Cluster Monitor Event Driven Message Producer is a message producer which
+ * enables {@link HazelcastClusterMonitorMessageProducer.HazelcastClusterMonitorListener}
+ * listener in order to listen cluster related events and sends events to related channel.
+ *
+ * @author Eren Avsarogullari
+ * @since 1.0.0
+ */
+public class HazelcastClusterMonitorMessageProducer extends MessageProducerSupport {
+
+	private final HazelcastInstance hazelcastInstance;
+
+	private Set<String> monitorTypes = Collections.singleton(ClusterMonitorType.MEMBERSHIP.name());
+
+	private final Map<ClusterMonitorType, String> hazelcastRegisteredListenerIdMap = new ConcurrentHashMap<>(5);
+
+	public HazelcastClusterMonitorMessageProducer(HazelcastInstance hazelcastInstance) {
+		Assert.notNull(hazelcastInstance, "'hazelcastInstance' must not be null");
+		this.hazelcastInstance = hazelcastInstance;
+	}
+
+	public void setMonitorEventTypes(String monitorEventTypes) {
+		final Set<String> monitorTypes =
+				HazelcastIntegrationDefinitionValidator.validateEnumType(ClusterMonitorType.class, monitorEventTypes);
+		Assert.notEmpty(monitorTypes, "'monitorTypes' must have elements");
+		this.monitorTypes = monitorTypes;
+	}
+
+	@Override
+	protected void doStart() {
+		final HazelcastClusterMonitorListener clusterMonitorListener = new HazelcastClusterMonitorListener();
+
+		if (this.monitorTypes.contains(ClusterMonitorType.MEMBERSHIP.name())) {
+			final String registrationId = this.hazelcastInstance.getCluster()
+					.addMembershipListener(clusterMonitorListener);
+			this.hazelcastRegisteredListenerIdMap.put(ClusterMonitorType.MEMBERSHIP, registrationId);
+		}
+
+		if (this.monitorTypes.contains(ClusterMonitorType.DISTRIBUTED_OBJECT.name())) {
+			final String registrationId = this.hazelcastInstance
+					.addDistributedObjectListener(clusterMonitorListener);
+			this.hazelcastRegisteredListenerIdMap.put(ClusterMonitorType.DISTRIBUTED_OBJECT, registrationId);
+		}
+
+		if (this.monitorTypes.contains(ClusterMonitorType.MIGRATION.name())) {
+			final String registrationId = this.hazelcastInstance.getPartitionService()
+					.addMigrationListener(clusterMonitorListener);
+			this.hazelcastRegisteredListenerIdMap.put(ClusterMonitorType.MIGRATION,	registrationId);
+		}
+
+		if (this.monitorTypes.contains(ClusterMonitorType.LIFECYCLE.name())) {
+			final String registrationId = this.hazelcastInstance.getLifecycleService()
+					.addLifecycleListener(clusterMonitorListener);
+			this.hazelcastRegisteredListenerIdMap.put(ClusterMonitorType.LIFECYCLE,	registrationId);
+		}
+
+		if (this.monitorTypes.contains(ClusterMonitorType.CLIENT.name())) {
+			final String registrationId = this.hazelcastInstance.getClientService()
+					.addClientListener(clusterMonitorListener);
+			this.hazelcastRegisteredListenerIdMap.put(ClusterMonitorType.CLIENT, registrationId);
+		}
+	}
+
+	@Override
+	protected void doStop() {
+		if (this.hazelcastInstance.getLifecycleService().isRunning()) {
+			String id = this.hazelcastRegisteredListenerIdMap.remove(ClusterMonitorType.MEMBERSHIP);
+			if (id != null) {
+				this.hazelcastInstance.getCluster().removeMembershipListener(id);
+			}
+
+			id = this.hazelcastRegisteredListenerIdMap.remove(ClusterMonitorType.DISTRIBUTED_OBJECT);
+			if (id != null) {
+				this.hazelcastInstance.removeDistributedObjectListener(id);
+			}
+
+			id = this.hazelcastRegisteredListenerIdMap.remove(ClusterMonitorType.MIGRATION);
+			if (id != null) {
+				this.hazelcastInstance.getPartitionService().removeMigrationListener(id);
+			}
+
+			id = this.hazelcastRegisteredListenerIdMap.remove(ClusterMonitorType.LIFECYCLE);
+			if (id != null) {
+				this.hazelcastInstance.getLifecycleService().removeLifecycleListener(id);
+			}
+
+			id = this.hazelcastRegisteredListenerIdMap.remove(ClusterMonitorType.CLIENT);
+			if (id != null) {
+				this.hazelcastInstance.getClientService().removeClientListener(id);
+			}
+		}
+	}
+
+	@Override
+	public String getComponentType() {
+		return "hazelcast:cm-inbound-channel-adapter";
+	}
+
+	private void processEvent(Object event) {
+		Assert.notNull(event, "'hazelcast event' must not be null");
+		if (logger.isDebugEnabled()) {
+			logger.debug("Received Cluster Monitor Event : " + event);
+		}
+		this.sendMessage(getMessageBuilderFactory().withPayload(event).build());
+
+	}
+
+	private final class HazelcastClusterMonitorListener implements MembershipListener,
+			DistributedObjectListener, MigrationListener, LifecycleListener,
+			ClientListener {
+
+		@Override
+		public void memberAdded(MembershipEvent membershipEvent) {
+			processEvent(membershipEvent);
+		}
+
+		@Override
+		public void memberRemoved(MembershipEvent membershipEvent) {
+			processEvent(membershipEvent);
+		}
+
+		@Override
+		public void memberAttributeChanged(MemberAttributeEvent memberAttributeEvent) {
+			processEvent(memberAttributeEvent);
+		}
+
+		@Override
+		public void distributedObjectCreated(DistributedObjectEvent event) {
+			processEvent(event);
+		}
+
+		@Override
+		public void distributedObjectDestroyed(DistributedObjectEvent event) {
+			processEvent(event);
+		}
+
+		@Override
+		public void migrationStarted(MigrationEvent migrationEvent) {
+			processEvent(migrationEvent);
+		}
+
+		@Override
+		public void migrationCompleted(MigrationEvent migrationEvent) {
+			processEvent(migrationEvent);
+		}
+
+		@Override
+		public void migrationFailed(MigrationEvent migrationEvent) {
+			processEvent(migrationEvent);
+		}
+
+		@Override
+		public void stateChanged(LifecycleEvent event) {
+			processEvent(event);
+		}
+
+		@Override
+		public void clientConnected(Client client) {
+			processEvent(client);
+		}
+
+		@Override
+		public void clientDisconnected(Client client) {
+			processEvent(client);
+		}
+
+	}
+
+}

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

@@ -22,7 +22,6 @@ import java.util.Map;
 import org.springframework.expression.EvaluationContext;
 import org.springframework.expression.Expression;
 import org.springframework.integration.expression.ExpressionUtils;
-import org.springframework.integration.expression.IntegrationEvaluationContextAware;
 import org.springframework.integration.handler.AbstractMessageHandler;
 import org.springframework.integration.hazelcast.HazelcastHeaders;
 import org.springframework.messaging.Message;

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

@@ -257,4 +257,47 @@
 		</xsd:complexType>
 
 	</xsd:element>
+	
+	<xsd:element name="cm-inbound-channel-adapter">
+
+		<xsd:annotation>
+			<xsd:documentation>
+				Configures Hazelcast Cluster Monitor Inbound Channel Adapter
+			</xsd:documentation>
+		</xsd:annotation>
+
+		<xsd:complexType>
+
+			<xsd:attributeGroup ref="integration:channelAdapterAttributes"/>
+
+			<xsd:attribute name="hazelcast-instance" use="required" type="xsd:string">
+				<xsd:annotation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:expected-type type="com.hazelcast.core.HazelcastInstance" />
+						</tool:annotation>
+					</xsd:appinfo>
+					<xsd:documentation>
+						<![CDATA[ Specifies hazelcast instance reference to listen ]]>
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+
+			<xsd:attribute name="monitor-types" type="xsd:string" use="optional" default="MEMBERSHIP">
+				<xsd:annotation>
+					<xsd:appinfo>
+						<tool:annotation kind="value">
+							<tool:expected-type
+								type="org.springframework.integration.hazelcast.ClusterMonitorType" />
+						</tool:annotation>
+					</xsd:appinfo>
+					<xsd:documentation>
+						<![CDATA[ Specifies cluster monitor types ]]>
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+		</xsd:complexType>
+
+	</xsd:element>
+	
 </xsd:schema>