Spring/spring-integration-extensions
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Start version 2.0.0

Browse Source 
Fixes https://github.com/spring-projects/spring-integration-extensions/issues/197

* Upgrade dependencies
* Fix deprecations; logic according new Hazelcast
* Fix tests for CP Subsystem
* Upgrade Gradle
* Prepare for release
This commit is contained in:
Artem Bilan
2019-11-05 14:50:30 -05:00
parent f6f686f128
commit fe6ffee60c
50 changed files with 475 additions and 274 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/HazelcastLocalInstanceRegistrar.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2017 the original author or authors.
* Copyright 2015-2019 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.
@@ -96,7 +96,7 @@ public class HazelcastLocalInstanceRegistrar implements SmartInitializingSinglet
}
private void syncConfigurationMultiMap(HazelcastInstance hazelcastInstance) {
Lock lock = hazelcastInstance.getLock(SPRING_INTEGRATION_INTERNAL_CLUSTER_LOCK);
Lock lock = hazelcastInstance.getCPSubsystem().getLock(SPRING_INTEGRATION_INTERNAL_CLUSTER_LOCK);
lock.lock();
try {
MultiMap<SocketAddress, SocketAddress> multiMap = hazelcastInstance

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

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2017 the original author or authors.
* Copyright 2015-2019 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.
@@ -42,6 +42,7 @@ import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.MapEvent;
import com.hazelcast.core.MultiMap;
import com.hazelcast.instance.EndpointQualifier;
/**
* Hazelcast Base Event-Driven Message Producer.
@@ -118,10 +119,9 @@ public abstract class AbstractHazelcastMessageProducer extends MessageProducerSu
final Set<HazelcastInstance> hazelcastInstanceSet = Hazelcast.getAllHazelcastInstances();
final Set<SocketAddress> localSocketAddressesSet = getLocalSocketAddresses(hazelcastInstanceSet);
return localSocketAddressesSet.isEmpty() ||
(!localSocketAddressesSet.isEmpty()
&& (localSocketAddressesSet.contains(socketAddress) ||
isEventComingFromNonRegisteredHazelcastInstance(hazelcastInstanceSet.iterator().next(),
localSocketAddressesSet, socketAddress)));
localSocketAddressesSet.contains(socketAddress)
|| isEventComingFromNonRegisteredHazelcastInstance(hazelcastInstanceSet.iterator().next(),
localSocketAddressesSet, socketAddress);
}
@@ -193,19 +193,20 @@ public abstract class AbstractHazelcastMessageProducer extends MessageProducerSu
if (AbstractHazelcastMessageProducer.this.logger.isDebugEnabled()) {
AbstractHazelcastMessageProducer.this.logger.debug("Received Event : " + event);
}
sendMessage(event, event.getMember().getSocketAddress(), getCacheListeningPolicy());
sendMessage(event,
event.getMember().getSocketAddress(EndpointQualifier.MEMBER), getCacheListeningPolicy());
}
}
@Override
@SuppressWarnings("unchecked")
protected Message<?> toMessage(AbstractIMapEvent event) {
final Map<String, Object> headers = new HashMap<String, Object>();
final Map<String, Object> headers = new HashMap<>();
headers.put(HazelcastHeaders.EVENT_TYPE, event.getEventType().name());
headers.put(HazelcastHeaders.MEMBER, event.getMember().getSocketAddress());
headers.put(HazelcastHeaders.MEMBER, event.getMember().getSocketAddress(EndpointQualifier.MEMBER));
headers.put(HazelcastHeaders.CACHE_NAME, event.getName());
if (event instanceof EntryEvent) {
@SuppressWarnings("unchecked")
EntryEvent<K, V> entryEvent = (EntryEvent<K, V>) event;
EntryEventMessagePayload<K, V> messagePayload = new EntryEventMessagePayload<>(entryEvent.getKey(),
entryEvent.getValue(), entryEvent.getOldValue());

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

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2018 the original author or authors.
* Copyright 2015-2019 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.
@@ -35,6 +35,7 @@ import com.hazelcast.core.Message;
import com.hazelcast.core.MessageListener;
import com.hazelcast.core.MultiMap;
import com.hazelcast.core.ReplicatedMap;
import com.hazelcast.instance.EndpointQualifier;
import com.hazelcast.map.listener.MapListener;
/**
@@ -140,7 +141,8 @@ public class HazelcastEventDrivenMessageProducer extends AbstractHazelcastMessag
@Override
protected void processEvent(ItemEvent<E> event) {
if (getCacheEvents().contains(event.getEventType().toString())) {
sendMessage(event, event.getMember().getSocketAddress(), getCacheListeningPolicy());
sendMessage(event,
event.getMember().getSocketAddress(EndpointQualifier.MEMBER), getCacheListeningPolicy());
}
if (logger.isDebugEnabled()) {
@@ -152,7 +154,7 @@ public class HazelcastEventDrivenMessageProducer extends AbstractHazelcastMessag
protected org.springframework.messaging.Message<?> toMessage(ItemEvent<E> event) {
final Map<String, Object> headers = new HashMap<>();
headers.put(HazelcastHeaders.EVENT_TYPE, event.getEventType().name());
headers.put(HazelcastHeaders.MEMBER, event.getMember().getSocketAddress());
headers.put(HazelcastHeaders.MEMBER, event.getMember().getSocketAddress(EndpointQualifier.MEMBER));
return getMessageBuilderFactory().withPayload(event.getItem()).copyHeaders(headers).build();
}
@@ -169,7 +171,8 @@ public class HazelcastEventDrivenMessageProducer extends AbstractHazelcastMessag
@Override
protected void processEvent(Message<E> event) {
sendMessage(event, event.getPublishingMember().getSocketAddress(), getCacheListeningPolicy());
sendMessage(event,
event.getPublishingMember().getSocketAddress(EndpointQualifier.MEMBER), getCacheListeningPolicy());
if (logger.isDebugEnabled()) {
logger.debug("Received Message : " + event);
@@ -181,7 +184,8 @@ public class HazelcastEventDrivenMessageProducer extends AbstractHazelcastMessag
Assert.notNull(event.getMessageObject(), "message must not be null");
final Map<String, Object> headers = new HashMap<>();
headers.put(HazelcastHeaders.MEMBER, event.getPublishingMember().getSocketAddress());
headers.put(HazelcastHeaders.MEMBER,
event.getPublishingMember().getSocketAddress(EndpointQualifier.MEMBER));
headers.put(HazelcastHeaders.CACHE_NAME, event.getSource());
headers.put(HazelcastHeaders.PUBLISHING_TIME, event.getPublishTime());

8
spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/leader/LeaderInitiator.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2018 the original author or authors.
* Copyright 2015-2019 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.
@@ -39,7 +39,7 @@ import org.springframework.integration.support.leader.LockRegistryLeaderInitiato
import org.springframework.util.Assert;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.ILock;
import com.hazelcast.cp.lock.FencedLock;
/**
* Bootstrap leadership {@link org.springframework.integration.leader.Candidate candidates}
@@ -106,7 +106,7 @@ public class LeaderInitiator implements SmartLifecycle, DisposableBean, Applicat
/**
* Hazelcast distributed lock.
*/
private volatile ILock lock;
private volatile FencedLock lock;
private boolean customPublisher = false;
@@ -210,7 +210,7 @@ public class LeaderInitiator implements SmartLifecycle, DisposableBean, Applicat
@Override
public synchronized void start() {
if (!this.running) {
this.lock = this.client.getLock(this.candidate.getRole());
this.lock = this.client.getCPSubsystem().getLock(this.candidate.getRole());
this.leaderSelector = new LeaderSelector();
this.running = true;
this.future = this.executorService.submit(this.leaderSelector);

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

@@ -1,5 +1,5 @@
/*
* Copyright 2015 the original author or authors.
* Copyright 2015-2019 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,23 +27,28 @@ import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.MembershipAdapter;
import com.hazelcast.core.MembershipEvent;
import com.hazelcast.core.MultiMap;
import com.hazelcast.instance.EndpointQualifier;
/**
* Hazelcast {@link MembershipAdapter} in order to listen for membership updates in the cluster.
*
* @author Eren Avsarogullari
* @author Artem Bilan
*
* @since 1.0.0
*/
public class HazelcastMembershipListener extends MembershipAdapter {
@Override
public void memberRemoved(MembershipEvent membershipEvent) {
SocketAddress removedMemberSocketAddress = membershipEvent.getMember().getSocketAddress();
SocketAddress removedMemberSocketAddress =
membershipEvent.getMember().getSocketAddress(EndpointQualifier.MEMBER);
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 =
hazelcastInstance.getCPSubsystem()
.getLock(HazelcastLocalInstanceRegistrar.SPRING_INTEGRATION_INTERNAL_CLUSTER_LOCK);
lock.lock();
try {
MultiMap<SocketAddress, SocketAddress> configMultiMap = hazelcastInstance

4
spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/lock/HazelcastLockRegistry.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017 the original author or authors.
* Copyright 2017-2019 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.
@@ -40,7 +40,7 @@ public class HazelcastLockRegistry implements LockRegistry {
@Override
public Lock obtain(Object lockKey) {
Assert.isInstanceOf(String.class, lockKey);
return this.client.getLock((String) lockKey);
return this.client.getCPSubsystem().getLock((String) lockKey);
}
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2015 the original author or authors.
* Copyright 2015-2019 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.
@@ -37,6 +37,7 @@ import com.hazelcast.core.MultiMap;
*
* @author Eren Avsarogullari
* @author Artem Bilan
*
* @since 1.0.0
*/
public class HazelcastCacheWritingMessageHandler extends AbstractMessageHandler {
@@ -71,14 +72,14 @@ public class HazelcastCacheWritingMessageHandler extends AbstractMessageHandler
}
@Override
protected void onInit() throws Exception {
protected void onInit() {
super.onInit();
this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(getBeanFactory());
}
@Override
@SuppressWarnings({"unchecked", "rawtypes"})
protected void handleMessageInternal(final Message<?> message) throws Exception {
protected void handleMessageInternal(final Message<?> message) {
Object objectToStore = message;
if (this.extractPayload) {
objectToStore = message.getPayload();

12
spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/store/HazelcastMessageStore.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017 the original author or authors.
* Copyright 2017-2019 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.
@@ -57,6 +57,16 @@ public class HazelcastMessageStore extends AbstractKeyValueMessageStore {
this.map.put(id, objectToStore);
}
@Override
protected void doStoreIfAbsent(Object id, Object objectToStore) {
this.map.putIfAbsent(id, objectToStore);
}
@Override
protected void doRemoveAll(Collection<Object> ids) {
this.map.removeAll((mapEntry) -> ids.contains(mapEntry.getKey()));
}
@Override
protected Object doRemove(Object id) {
return this.map.remove(id);