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()`
This commit is contained in:
Artem Bilan
2017-01-31 18:24:45 -05:00
parent 894a60b8bb
commit 5511987be2
4 changed files with 57 additions and 15 deletions

View File

@@ -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());
}
}

View File

@@ -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));
}
}

View File

@@ -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<HazelcastInstance> hazelcastInstanceSet = Hazelcast.getAllHazelcastInstances();
final Set<SocketAddress> 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)));
}

View File

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