From 5511987be258e6f7dbc3818237b059a832a92e7b Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 31 Jan 2017 18:24:45 -0500 Subject: [PATCH] GH-172: Do not require local HZ instances Fixes spring-projects/spring-integration-extensions#172 Since we can have an application based on the `HazelcastClient`, we don't need to require `Hazelcast.getAllHazelcastInstances()` be presented. * Rework `HazelcastLocalInstanceRegistrar` to be able to accept external `HazelcastInstance` for `MultiMap` and `MembershipListener` registration * If there is on local `Hazelcast.getAllHazelcastInstances()` just log a warn that we can't register `MembershipListener` * Allow for `AbstractHazelcastMessageProducer` to accept events in the `CacheListeningPolicyType.SINGLE` mode when there is no local `Hazelcast.getAllHazelcastInstances()` --- .../HazelcastLocalInstanceRegistrar.java | 45 ++++++++++++++++--- ...stIntegrationConfigurationInitializer.java | 11 +++-- .../AbstractHazelcastMessageProducer.java | 10 +++-- ...stIntegrationInboundTestConfiguration.java | 6 +++ 4 files changed, 57 insertions(+), 15 deletions(-) diff --git a/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/HazelcastLocalInstanceRegistrar.java b/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/HazelcastLocalInstanceRegistrar.java index cbb64ed..503723f 100644 --- a/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/HazelcastLocalInstanceRegistrar.java +++ b/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/HazelcastLocalInstanceRegistrar.java @@ -19,6 +19,9 @@ package org.springframework.integration.hazelcast; import java.net.SocketAddress; import java.util.concurrent.locks.Lock; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + import org.springframework.beans.factory.SmartInitializingSingleton; import org.springframework.integration.hazelcast.listener.HazelcastMembershipListener; @@ -33,10 +36,19 @@ import com.hazelcast.core.MultiMap; * membership updates. * * @author Eren Avsarogullari + * @author Artem Bilan + * * @since 1.0.0 */ public class HazelcastLocalInstanceRegistrar implements SmartInitializingSingleton { + private static final Log logger = LogFactory.getLog(HazelcastLocalInstanceRegistrar.class); + + /** + * The bean name for the {@link HazelcastLocalInstanceRegistrar} instance. + */ + public static final String BEAN_NAME = "hazelcastLocalInstanceRegistrar"; + /** * The name for the Hazelcast MultiMap used for membership registration. */ @@ -48,15 +60,38 @@ public class HazelcastLocalInstanceRegistrar implements SmartInitializingSinglet */ public static final String SPRING_INTEGRATION_INTERNAL_CLUSTER_LOCK = "SPRING_INTEGRATION_INTERNAL_CLUSTER_LOCK"; + private final HazelcastInstance hazelcastInstance; + + /** + * Construct {@link HazelcastLocalInstanceRegistrar} based on the local JVM {@link HazelcastInstance}s if any. + */ + public HazelcastLocalInstanceRegistrar() { + this.hazelcastInstance = null; + } + + /** + * Construct {@link HazelcastLocalInstanceRegistrar} based on the provided {@link HazelcastInstance}. + * @param hazelcastInstance the {@link HazelcastInstance} to use. + */ + public HazelcastLocalInstanceRegistrar(HazelcastInstance hazelcastInstance) { + this.hazelcastInstance = hazelcastInstance; + } + @Override public void afterSingletonsInstantiated() { - if (!Hazelcast.getAllHazelcastInstances().isEmpty()) { - HazelcastInstance hazelcastInstance = Hazelcast.getAllHazelcastInstances().iterator().next(); - hazelcastInstance.getCluster().addMembershipListener(new HazelcastMembershipListener()); - syncConfigurationMultiMap(hazelcastInstance); + if (this.hazelcastInstance == null) { + if (!Hazelcast.getAllHazelcastInstances().isEmpty()) { + HazelcastInstance hazelcastInstance = Hazelcast.getAllHazelcastInstances().iterator().next(); + hazelcastInstance.getCluster().addMembershipListener(new HazelcastMembershipListener()); + syncConfigurationMultiMap(hazelcastInstance); + } + else { + logger.warn("No HazelcastInstances for MembershipListener registration"); + } } else { - throw new IllegalStateException("No Active Local Hazelcast Instance found."); + syncConfigurationMultiMap(this.hazelcastInstance); + this.hazelcastInstance.getCluster().addMembershipListener(new HazelcastMembershipListener()); } } diff --git a/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/config/HazelcastIntegrationConfigurationInitializer.java b/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/config/HazelcastIntegrationConfigurationInitializer.java index d93ae9a..234787d 100644 --- a/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/config/HazelcastIntegrationConfigurationInitializer.java +++ b/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/config/HazelcastIntegrationConfigurationInitializer.java @@ -1,5 +1,5 @@ /* - * Copyright 2015 the original author or authors. + * Copyright 2015-2017 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -27,18 +27,17 @@ import org.springframework.integration.hazelcast.HazelcastLocalInstanceRegistrar * The Hazelcast Integration infrastructure {@code beanFactory} initializer. * * @author Eren Avsarogullari + * @author Artem Bilan + * * @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, + if (!beanDefinitionRegistry.containsBeanDefinition(HazelcastLocalInstanceRegistrar.BEAN_NAME)) { + beanDefinitionRegistry.registerBeanDefinition(HazelcastLocalInstanceRegistrar.BEAN_NAME, new RootBeanDefinition(HazelcastLocalInstanceRegistrar.class)); } } diff --git a/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/inbound/AbstractHazelcastMessageProducer.java b/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/inbound/AbstractHazelcastMessageProducer.java index b77ac9c..7796430 100644 --- a/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/inbound/AbstractHazelcastMessageProducer.java +++ b/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/inbound/AbstractHazelcastMessageProducer.java @@ -48,6 +48,7 @@ import com.hazelcast.core.MultiMap; * * @author Eren Avsarogullari * @author Artem Bilan + * * @since 1.0.0 */ public abstract class AbstractHazelcastMessageProducer extends MessageProducerSupport { @@ -116,10 +117,11 @@ public abstract class AbstractHazelcastMessageProducer extends MessageProducerSu private boolean isEventAcceptable(final InetSocketAddress socketAddress) { final Set hazelcastInstanceSet = Hazelcast.getAllHazelcastInstances(); final Set localSocketAddressesSet = getLocalSocketAddresses(hazelcastInstanceSet); - return (!localSocketAddressesSet.isEmpty()) - && (localSocketAddressesSet.contains(socketAddress) || - isEventComingFromNonRegisteredHazelcastInstance(hazelcastInstanceSet.iterator().next(), - localSocketAddressesSet, socketAddress)); + return localSocketAddressesSet.isEmpty() || + (!localSocketAddressesSet.isEmpty() + && (localSocketAddressesSet.contains(socketAddress) || + isEventComingFromNonRegisteredHazelcastInstance(hazelcastInstanceSet.iterator().next(), + localSocketAddressesSet, socketAddress))); } diff --git a/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/inbound/config/HazelcastIntegrationInboundTestConfiguration.java b/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/inbound/config/HazelcastIntegrationInboundTestConfiguration.java index 5f5de3f..e831a0f 100644 --- a/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/inbound/config/HazelcastIntegrationInboundTestConfiguration.java +++ b/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/inbound/config/HazelcastIntegrationInboundTestConfiguration.java @@ -24,6 +24,7 @@ import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.config.EnableIntegration; import org.springframework.integration.hazelcast.DistributedSQLIterationType; import org.springframework.integration.hazelcast.HazelcastIntegrationTestUser; +import org.springframework.integration.hazelcast.HazelcastLocalInstanceRegistrar; import org.springframework.integration.hazelcast.inbound.HazelcastClusterMonitorMessageProducer; import org.springframework.integration.hazelcast.inbound.HazelcastContinuousQueryMessageProducer; import org.springframework.integration.hazelcast.inbound.HazelcastDistributedSQLMessageSource; @@ -215,6 +216,11 @@ public class HazelcastIntegrationInboundTestConfiguration { return Hazelcast.newHazelcastInstance(); } + @Bean(HazelcastLocalInstanceRegistrar.BEAN_NAME) + public HazelcastLocalInstanceRegistrar hazelcastLocalInstanceRegistrar() { + return new HazelcastLocalInstanceRegistrar(testHazelcastInstance()); + } + @Bean public HazelcastEventDrivenMessageProducer hazelcastEventDrivenMessageProducer() { final HazelcastEventDrivenMessageProducer producer =