Refactor MembershipListenerAdapter away from using java.util.function.Consumer implementations to delegate the processing of membership event towards using the new MembershipEvent types along with overloaded membership event handler methods.
Move MembershipListenerAdapter from org.springframework.geode.distributed to org.springframework.geode.distributed.event. Edit Javadoc.
This commit is contained in:
John Blum
@@ -1,198 +0,0 @@
|
||||
/*
|
||||
* Copyright 2020 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
|
||||
*
|
||||
* https://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.geode.distributed;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Optional;
|
||||
import java.util.Set;
|
||||
import java.util.function.BiConsumer;
|
||||
import java.util.function.Consumer;
|
||||
|
||||
import org.apache.geode.cache.Cache;
|
||||
import org.apache.geode.distributed.internal.DistributionManager;
|
||||
import org.apache.geode.distributed.internal.InternalDistributedSystem;
|
||||
import org.apache.geode.distributed.internal.MembershipListener;
|
||||
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
|
||||
|
||||
import org.springframework.geode.util.function.QuadConsumer;
|
||||
import org.springframework.geode.util.function.TriConsumer;
|
||||
|
||||
/**
|
||||
* A {@link MembershipListener} implementation using the
|
||||
* <a href="https://en.wikipedia.org/wiki/Adapter_pattern">Adapter Software Design Pattern</a>
|
||||
* to delegate membership event callbacks to a {@link Consumer} of those membership events.
|
||||
*
|
||||
* @author John Blum
|
||||
* @see java.util.function.BiConsumer
|
||||
* @see java.util.function.Consumer
|
||||
* @see org.apache.geode.distributed.internal.DistributionManager
|
||||
* @see org.apache.geode.distributed.internal.MembershipListener
|
||||
* @see org.apache.geode.distributed.internal.membership.InternalDistributedMember
|
||||
* @since 1.3.0
|
||||
*/
|
||||
@SuppressWarnings("unused")
|
||||
public class MembershipListenerAdapter implements MembershipListener {
|
||||
|
||||
/**
|
||||
* Factory method used to construct a new instance of the {@link MembershipListenerAdapter}.
|
||||
*
|
||||
* @return a new instance of {@link MembershipListenerAdapter}.
|
||||
*/
|
||||
public static MembershipListenerAdapter create() {
|
||||
return new MembershipListenerAdapter();
|
||||
}
|
||||
|
||||
private TriConsumer<DistributionManager, InternalDistributedMember, Boolean> memberDepartedConsumer =
|
||||
(manager, member, crashed) -> {};
|
||||
|
||||
private BiConsumer<DistributionManager, InternalDistributedMember> memberJoinedConsumer = (manager, member) -> {};
|
||||
|
||||
private QuadConsumer<DistributionManager, InternalDistributedMember, InternalDistributedMember, String> memberSuspectConsumer =
|
||||
(manage, member, suspect, reason) -> {};
|
||||
|
||||
private TriConsumer<DistributionManager, Set<InternalDistributedMember>, List<InternalDistributedMember>> quorumLostConsumer =
|
||||
(manager, failures, remaining) -> {};
|
||||
|
||||
/**
|
||||
* @inheritDoc
|
||||
*/
|
||||
@Override
|
||||
public void memberDeparted(DistributionManager distributionManager, InternalDistributedMember distributedMember,
|
||||
boolean crashed) {
|
||||
|
||||
this.memberDepartedConsumer.accept(distributionManager, distributedMember, crashed);
|
||||
}
|
||||
|
||||
/**
|
||||
* @inheritDoc
|
||||
*/
|
||||
@Override
|
||||
public void memberJoined(DistributionManager distributionManager, InternalDistributedMember distributedMember) {
|
||||
this.memberJoinedConsumer.accept(distributionManager, distributedMember);
|
||||
}
|
||||
|
||||
/**
|
||||
* @inheritDoc
|
||||
*/
|
||||
@Override
|
||||
public void memberSuspect(DistributionManager distributionManager, InternalDistributedMember distributedMember,
|
||||
InternalDistributedMember whoSuspected, String reason) {
|
||||
|
||||
this.memberSuspectConsumer.accept(distributionManager, distributedMember, whoSuspected, reason);
|
||||
}
|
||||
|
||||
/**
|
||||
* @inheritDoc
|
||||
*/
|
||||
@Override
|
||||
public void quorumLost(DistributionManager distributionManager, Set<InternalDistributedMember> failures,
|
||||
List<InternalDistributedMember> remaining) {
|
||||
|
||||
this.quorumLostConsumer.accept(distributionManager, failures, remaining);
|
||||
}
|
||||
|
||||
/**
|
||||
* Registers this {@link MembershipListener} with the given {@literal peer} {@link Cache}.
|
||||
*
|
||||
* @param peerCache {@literal peer} {@link Cache} on which to register this {@link MembershipListener}.
|
||||
* @return this {@link MembershipListenerAdapter}.
|
||||
* @see org.apache.geode.cache.Cache
|
||||
*/
|
||||
public MembershipListenerAdapter register(Cache peerCache) {
|
||||
|
||||
Optional.ofNullable(peerCache)
|
||||
.map(Cache::getDistributedSystem)
|
||||
.filter(InternalDistributedSystem.class::isInstance)
|
||||
.map(InternalDistributedSystem.class::cast)
|
||||
.map(InternalDistributedSystem::getDistributionManager)
|
||||
.ifPresent(distributionManager -> distributionManager
|
||||
.addMembershipListener(this));
|
||||
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Null-safe builder method used to add a {@link #memberDeparted(DistributionManager, InternalDistributedMember, boolean)}
|
||||
* {@link TriConsumer} event handler.
|
||||
*
|
||||
* @param memberDepartedConsumer {@link TriConsumer} handling {@literal memberDeparted} events.
|
||||
* @return this {@link MembershipListenerAdapter}.
|
||||
* @see org.springframework.geode.util.function.TriConsumer
|
||||
*/
|
||||
public MembershipListenerAdapter withMemberDepartedConsumer(
|
||||
TriConsumer<DistributionManager, InternalDistributedMember, Boolean> memberDepartedConsumer) {
|
||||
|
||||
if (memberDepartedConsumer != null) {
|
||||
this.memberDepartedConsumer = memberDepartedConsumer;
|
||||
}
|
||||
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Null-safe builder method used to add a {@link #memberJoined(DistributionManager, InternalDistributedMember)}
|
||||
* {@link BiConsumer} event handler.
|
||||
*
|
||||
* @param memberJoinedConsumer {@link BiConsumer} handling {@literal memberJoined} events.
|
||||
* @return this {@link MembershipListenerAdapter}.
|
||||
* @see java.util.function.BiConsumer
|
||||
*/
|
||||
public MembershipListenerAdapter withMemberJoinedConsumer(
|
||||
BiConsumer<DistributionManager, InternalDistributedMember> memberJoinedConsumer) {
|
||||
|
||||
if (memberJoinedConsumer != null) {
|
||||
this.memberJoinedConsumer = memberJoinedConsumer;
|
||||
}
|
||||
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Null-safe builder method used to add a {@link #memberSuspect(DistributionManager, InternalDistributedMember, InternalDistributedMember, String)}
|
||||
* {@link QuadConsumer} event handler.
|
||||
*
|
||||
* @param memberSuspectConsumer {@link QuadConsumer} handling {@literal memberSuspect} events.
|
||||
* @return this {@link MembershipListenerAdapter}.
|
||||
* @see org.springframework.geode.util.function.QuadConsumer
|
||||
*/
|
||||
public MembershipListenerAdapter withMemberSuspectConsumer(
|
||||
QuadConsumer<DistributionManager, InternalDistributedMember, InternalDistributedMember, String> memberSuspectConsumer) {
|
||||
|
||||
if (memberSuspectConsumer != null) {
|
||||
this.memberSuspectConsumer = memberSuspectConsumer;
|
||||
}
|
||||
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Null-safe build method used to add a {@link #quorumLost(DistributionManager, Set, List)} {@link TriConsumer}
|
||||
* event handler.
|
||||
*
|
||||
* @param quorumLostConsumer {@link TriConsumer} handling {@literal quorumLost} events.
|
||||
* @return this {@link MembershipListenerAdapter}.
|
||||
* @see org.springframework.geode.util.function.TriConsumer
|
||||
*/
|
||||
public MembershipListenerAdapter withQuorumLostConsumer(
|
||||
TriConsumer<DistributionManager, Set<InternalDistributedMember>, List<InternalDistributedMember>> quorumLostConsumer) {
|
||||
|
||||
if (quorumLostConsumer != null) {
|
||||
this.quorumLostConsumer = quorumLostConsumer;
|
||||
}
|
||||
|
||||
return this;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,135 @@
|
||||
/*
|
||||
* Copyright 2020 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
|
||||
*
|
||||
* https://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.geode.distributed.event;
|
||||
|
||||
import java.util.EventListener;
|
||||
import java.util.List;
|
||||
import java.util.Optional;
|
||||
import java.util.Set;
|
||||
|
||||
import org.apache.geode.cache.Cache;
|
||||
import org.apache.geode.distributed.internal.DistributionManager;
|
||||
import org.apache.geode.distributed.internal.InternalDistributedSystem;
|
||||
import org.apache.geode.distributed.internal.MembershipListener;
|
||||
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
|
||||
|
||||
import org.springframework.geode.distributed.event.support.MemberDepartedEvent;
|
||||
import org.springframework.geode.distributed.event.support.MemberJoinedEvent;
|
||||
import org.springframework.geode.distributed.event.support.MemberSuspectEvent;
|
||||
import org.springframework.geode.distributed.event.support.QuorumLostEvent;
|
||||
|
||||
/**
|
||||
* An abstract {@link MembershipListener} implementation using the
|
||||
* <a href="https://en.wikipedia.org/wiki/Adapter_pattern">Adapter Software Design Pattern</a>
|
||||
* to delegate membership event callbacks to event handlers for those membership events.
|
||||
*
|
||||
* @author John Blum
|
||||
* @param <T> specific {@link Class sub-type} of this {@link MembershipListenerAdapter}.
|
||||
* @see java.util.EventListener
|
||||
* @see org.apache.geode.cache.Cache
|
||||
* @see org.apache.geode.distributed.internal.DistributionManager
|
||||
* @see org.apache.geode.distributed.internal.InternalDistributedSystem
|
||||
* @see org.apache.geode.distributed.internal.MembershipListener
|
||||
* @see org.apache.geode.distributed.internal.membership.InternalDistributedMember
|
||||
* @since 1.3.0
|
||||
*/
|
||||
@SuppressWarnings("unused")
|
||||
public abstract class MembershipListenerAdapter<T extends MembershipListenerAdapter<T>>
|
||||
implements EventListener, MembershipListener {
|
||||
|
||||
/**
|
||||
* @inheritDoc
|
||||
*/
|
||||
@Override
|
||||
public final void memberDeparted(DistributionManager manager, InternalDistributedMember member, boolean crashed) {
|
||||
|
||||
MemberDepartedEvent event = new MemberDepartedEvent(manager)
|
||||
.withMember(member)
|
||||
.crashed(crashed);
|
||||
|
||||
handleMemberDeparted(event);
|
||||
}
|
||||
|
||||
public void handleMemberDeparted(MemberDepartedEvent event) { }
|
||||
|
||||
/**
|
||||
* @inheritDoc
|
||||
*/
|
||||
@Override
|
||||
public final void memberJoined(DistributionManager manager, InternalDistributedMember member) {
|
||||
|
||||
MemberJoinedEvent event = new MemberJoinedEvent(manager)
|
||||
.withMember(member);
|
||||
|
||||
handleMemberJoined(event);
|
||||
}
|
||||
|
||||
public void handleMemberJoined(MemberJoinedEvent event) { }
|
||||
|
||||
/**
|
||||
* @inheritDoc
|
||||
*/
|
||||
@Override
|
||||
public final void memberSuspect(DistributionManager manager, InternalDistributedMember member,
|
||||
InternalDistributedMember suspectMember, String reason) {
|
||||
|
||||
MemberSuspectEvent event = new MemberSuspectEvent(manager)
|
||||
.withMember(member)
|
||||
.withReason(reason)
|
||||
.withSuspect(suspectMember);
|
||||
|
||||
handleMemberSuspect(event);
|
||||
}
|
||||
|
||||
public void handleMemberSuspect(MemberSuspectEvent event) { }
|
||||
|
||||
/**
|
||||
* @inheritDoc
|
||||
*/
|
||||
@Override
|
||||
public final void quorumLost(DistributionManager manager, Set<InternalDistributedMember> failedMembers,
|
||||
List<InternalDistributedMember> remainingMembers) {
|
||||
|
||||
QuorumLostEvent event = new QuorumLostEvent(manager)
|
||||
.withFailedMembers(failedMembers)
|
||||
.withRemainingMembers(remainingMembers);
|
||||
|
||||
handleQuorumLost(event);
|
||||
}
|
||||
|
||||
public void handleQuorumLost(QuorumLostEvent event) { }
|
||||
|
||||
/**
|
||||
* Registers this {@link MembershipListener} with the given {@literal peer} {@link Cache}.
|
||||
*
|
||||
* @param peerCache {@literal peer} {@link Cache} on which to register this {@link MembershipListener}.
|
||||
* @return this {@link MembershipListenerAdapter}.
|
||||
* @see org.apache.geode.cache.Cache
|
||||
*/
|
||||
@SuppressWarnings("unchecked")
|
||||
public T register(Cache peerCache) {
|
||||
|
||||
Optional.ofNullable(peerCache)
|
||||
.map(Cache::getDistributedSystem)
|
||||
.filter(InternalDistributedSystem.class::isInstance)
|
||||
.map(InternalDistributedSystem.class::cast)
|
||||
.map(InternalDistributedSystem::getDistributionManager)
|
||||
.ifPresent(distributionManager -> distributionManager
|
||||
.addMembershipListener(this));
|
||||
|
||||
return (T) this;
|
||||
}
|
||||
}
|
||||
@@ -1,235 +0,0 @@
|
||||
/*
|
||||
* Copyright 2020 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
|
||||
*
|
||||
* https://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.geode.distributed;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.mockito.ArgumentMatchers.eq;
|
||||
import static org.mockito.Mockito.doReturn;
|
||||
import static org.mockito.Mockito.mock;
|
||||
import static org.mockito.Mockito.times;
|
||||
import static org.mockito.Mockito.verify;
|
||||
import static org.mockito.Mockito.verifyNoInteractions;
|
||||
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
import java.util.Set;
|
||||
import java.util.function.BiConsumer;
|
||||
|
||||
import org.junit.Test;
|
||||
import org.junit.runner.RunWith;
|
||||
import org.mockito.Mock;
|
||||
import org.mockito.junit.MockitoJUnitRunner;
|
||||
|
||||
import org.apache.geode.cache.Cache;
|
||||
import org.apache.geode.distributed.DistributedSystem;
|
||||
import org.apache.geode.distributed.internal.DistributionManager;
|
||||
import org.apache.geode.distributed.internal.InternalDistributedSystem;
|
||||
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
|
||||
|
||||
import org.springframework.geode.util.function.QuadConsumer;
|
||||
import org.springframework.geode.util.function.TriConsumer;
|
||||
|
||||
/**
|
||||
* Unit Tests for {@link MembershipListenerAdapter}.
|
||||
*
|
||||
* @author John Blum
|
||||
* @see org.junit.Test
|
||||
* @see org.mockito.Mockito
|
||||
* @see org.mockito.junit.MockitoJUnitRunner
|
||||
* @see org.apache.geode.distributed.internal.DistributionManager
|
||||
* @see org.apache.geode.distributed.internal.membership.InternalDistributedMember
|
||||
* @see org.springframework.geode.distributed.MembershipListenerAdapter
|
||||
* @since 1.3.0
|
||||
*/
|
||||
@RunWith(MockitoJUnitRunner.class)
|
||||
public class MembershipListenerAdapterUnitTests {
|
||||
|
||||
@Mock
|
||||
private DistributionManager mockDistributionManager;
|
||||
|
||||
@Mock
|
||||
private InternalDistributedMember mockDistributedMember;
|
||||
|
||||
@Test
|
||||
public void createConstructsNewMembershipListenerAdapterWithNoOpEventHandlers() {
|
||||
|
||||
InternalDistributedMember mockSuspectMember = mock(InternalDistributedMember.class);
|
||||
|
||||
MembershipListenerAdapter membershipListener = MembershipListenerAdapter.create();
|
||||
|
||||
assertThat(membershipListener).isNotNull();
|
||||
|
||||
membershipListener.memberDeparted(this.mockDistributionManager, this.mockDistributedMember, true);
|
||||
membershipListener.memberJoined(this.mockDistributionManager, this.mockDistributedMember);
|
||||
membershipListener.memberSuspect(this.mockDistributionManager, this.mockDistributedMember, mockSuspectMember,
|
||||
"System is unstable!!");
|
||||
membershipListener.quorumLost(this.mockDistributionManager, Collections.singleton(mockSuspectMember),
|
||||
Collections.singletonList(this.mockDistributedMember));
|
||||
|
||||
verifyNoInteractions(this.mockDistributedMember);
|
||||
verifyNoInteractions(this.mockDistributedMember);
|
||||
verifyNoInteractions(mockSuspectMember);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void registersListenerWithPeerCache() {
|
||||
|
||||
Cache mockCache = mock(Cache.class);
|
||||
|
||||
DistributionManager mockDistributionManager = mock(DistributionManager.class);
|
||||
|
||||
InternalDistributedSystem mockDistributedSystem = mock(InternalDistributedSystem.class);
|
||||
|
||||
doReturn(mockDistributedSystem).when(mockCache).getDistributedSystem();
|
||||
doReturn(mockDistributionManager).when(mockDistributedSystem).getDistributionManager();
|
||||
|
||||
MembershipListenerAdapter membershipListener = new MembershipListenerAdapter();
|
||||
|
||||
assertThat(membershipListener.register(mockCache)).isSameAs(membershipListener);
|
||||
|
||||
verify(mockCache, times(1)).getDistributedSystem();
|
||||
verify(mockDistributedSystem, times(1)).getDistributionManager();
|
||||
verify(mockDistributionManager, times(1)).addMembershipListener(eq(membershipListener));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void registerListenerWithNullCacheIsNullSafe() {
|
||||
|
||||
MembershipListenerAdapter membershipListener = new MembershipListenerAdapter();
|
||||
|
||||
assertThat(membershipListener.register(null)).isSameAs(membershipListener);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void registerListenerWithNullDistributedSystemIsNullSafe() {
|
||||
|
||||
Cache mockCache = mock(Cache.class);
|
||||
|
||||
MembershipListenerAdapter membershipListener = new MembershipListenerAdapter();
|
||||
|
||||
assertThat(membershipListener.register(mockCache)).isSameAs(membershipListener);
|
||||
|
||||
verify(mockCache, times(1)).getDistributedSystem();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void registerListenerWithNonInternalDistributedSystemIsSafe() {
|
||||
|
||||
Cache mockCache = mock(Cache.class);
|
||||
|
||||
DistributedSystem mockDistributedSystem = mock(DistributedSystem.class);
|
||||
|
||||
MembershipListenerAdapter membershipListener = new MembershipListenerAdapter();
|
||||
|
||||
assertThat(membershipListener.register(mockCache)).isSameAs(membershipListener);
|
||||
|
||||
verify(mockCache, times(1)).getDistributedSystem();
|
||||
verifyNoInteractions(mockDistributedSystem);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void registerListenerWithNullDistributionManagerIsNullSafe() {
|
||||
|
||||
Cache mockCache = mock(Cache.class);
|
||||
|
||||
InternalDistributedSystem mockDistributedSystem = mock(InternalDistributedSystem.class);
|
||||
|
||||
doReturn(mockDistributedSystem).when(mockCache).getDistributedSystem();
|
||||
|
||||
MembershipListenerAdapter membershipListener = new MembershipListenerAdapter();
|
||||
|
||||
assertThat(membershipListener.register(mockCache)).isSameAs(membershipListener);
|
||||
|
||||
verify(mockCache, times(1)).getDistributedSystem();
|
||||
verify(mockDistributedSystem, times(1)).getDistributionManager();
|
||||
}
|
||||
|
||||
@Test
|
||||
@SuppressWarnings("unchecked")
|
||||
public void withMemberDepartedEventHandler() {
|
||||
|
||||
TriConsumer<DistributionManager, InternalDistributedMember, Boolean> mockConsumer = mock(TriConsumer.class);
|
||||
|
||||
MembershipListenerAdapter membershipListener = MembershipListenerAdapter.create()
|
||||
.withMemberDepartedConsumer(mockConsumer);
|
||||
|
||||
assertThat(membershipListener).isNotNull();
|
||||
|
||||
membershipListener.memberDeparted(this.mockDistributionManager, this.mockDistributedMember, true);
|
||||
|
||||
verify(mockConsumer, times(1))
|
||||
.accept(eq(this.mockDistributionManager), eq(this.mockDistributedMember), eq(true));
|
||||
}
|
||||
|
||||
@Test
|
||||
@SuppressWarnings("unchecked")
|
||||
public void withMemberJoinedEventHandler() {
|
||||
|
||||
BiConsumer<DistributionManager, InternalDistributedMember> mockConsumer = mock(BiConsumer.class);
|
||||
|
||||
MembershipListenerAdapter membershipListener = MembershipListenerAdapter.create()
|
||||
.withMemberJoinedConsumer(mockConsumer);
|
||||
|
||||
assertThat(membershipListener).isNotNull();
|
||||
|
||||
membershipListener.memberJoined(this.mockDistributionManager, this.mockDistributedMember);
|
||||
|
||||
verify(mockConsumer, times(1))
|
||||
.accept(eq(this.mockDistributionManager), eq(this.mockDistributedMember));
|
||||
}
|
||||
|
||||
@Test
|
||||
@SuppressWarnings("unchecked")
|
||||
public void withMemberSuspectEventHandler() {
|
||||
|
||||
QuadConsumer<DistributionManager, InternalDistributedMember, InternalDistributedMember, String> mockConsumer =
|
||||
mock(QuadConsumer.class);
|
||||
|
||||
InternalDistributedMember mockSuspectMember = mock(InternalDistributedMember.class);
|
||||
|
||||
MembershipListenerAdapter membershipListener = MembershipListenerAdapter.create()
|
||||
.withMemberSuspectConsumer(mockConsumer);
|
||||
|
||||
assertThat(membershipListener).isNotNull();
|
||||
|
||||
membershipListener.memberSuspect(this.mockDistributionManager, this.mockDistributedMember, mockSuspectMember,
|
||||
"System is a lost cause!!");
|
||||
|
||||
verify(mockConsumer, times(1))
|
||||
.accept(eq(this.mockDistributionManager), eq(this.mockDistributedMember), eq(mockSuspectMember),
|
||||
eq("System is a lost cause!!"));
|
||||
}
|
||||
|
||||
@Test
|
||||
@SuppressWarnings("unchecked")
|
||||
public void withQuorumLostEventHandler() {
|
||||
|
||||
TriConsumer<DistributionManager, Set<InternalDistributedMember>, List<InternalDistributedMember>> mockConsumer
|
||||
= mock(TriConsumer.class);
|
||||
|
||||
MembershipListenerAdapter membershipListener = MembershipListenerAdapter.create()
|
||||
.withQuorumLostConsumer(mockConsumer);
|
||||
|
||||
assertThat(membershipListener).isNotNull();
|
||||
|
||||
membershipListener.quorumLost(this.mockDistributionManager, Collections.singleton(this.mockDistributedMember),
|
||||
Collections.singletonList(this.mockDistributedMember));
|
||||
|
||||
verify(mockConsumer, times(1)).accept(eq(this.mockDistributionManager),
|
||||
eq(Collections.singleton(this.mockDistributedMember)),
|
||||
eq(Collections.singletonList(this.mockDistributedMember)));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,259 @@
|
||||
/*
|
||||
* Copyright 2020 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
|
||||
*
|
||||
* https://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.geode.distributed.event;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.mockito.ArgumentMatchers.any;
|
||||
import static org.mockito.ArgumentMatchers.eq;
|
||||
import static org.mockito.ArgumentMatchers.isA;
|
||||
import static org.mockito.Mockito.doAnswer;
|
||||
import static org.mockito.Mockito.doReturn;
|
||||
import static org.mockito.Mockito.mock;
|
||||
import static org.mockito.Mockito.spy;
|
||||
import static org.mockito.Mockito.times;
|
||||
import static org.mockito.Mockito.verify;
|
||||
import static org.mockito.Mockito.verifyNoInteractions;
|
||||
|
||||
import java.util.Arrays;
|
||||
import java.util.Collections;
|
||||
import java.util.HashSet;
|
||||
import java.util.Set;
|
||||
|
||||
import org.junit.Test;
|
||||
import org.junit.runner.RunWith;
|
||||
import org.mockito.Mock;
|
||||
import org.mockito.junit.MockitoJUnitRunner;
|
||||
|
||||
import org.apache.geode.cache.Cache;
|
||||
import org.apache.geode.distributed.DistributedSystem;
|
||||
import org.apache.geode.distributed.internal.DistributionManager;
|
||||
import org.apache.geode.distributed.internal.InternalDistributedSystem;
|
||||
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
|
||||
|
||||
import org.springframework.geode.distributed.event.support.MemberDepartedEvent;
|
||||
import org.springframework.geode.distributed.event.support.MemberJoinedEvent;
|
||||
import org.springframework.geode.distributed.event.support.MemberSuspectEvent;
|
||||
import org.springframework.geode.distributed.event.support.QuorumLostEvent;
|
||||
|
||||
/**
|
||||
* Unit Tests for {@link MembershipListenerAdapter}.
|
||||
*
|
||||
* @author John Blum
|
||||
* @see org.junit.Test
|
||||
* @see org.mockito.Mockito
|
||||
* @see org.mockito.junit.MockitoJUnitRunner
|
||||
* @see org.apache.geode.cache.Cache
|
||||
* @see org.apache.geode.distributed.DistributedMember
|
||||
* @see org.apache.geode.distributed.DistributedSystem
|
||||
* @see org.apache.geode.distributed.internal.DistributionManager
|
||||
* @see org.apache.geode.distributed.internal.membership.InternalDistributedMember
|
||||
* @see MembershipListenerAdapter
|
||||
* @since 1.3.0
|
||||
*/
|
||||
@RunWith(MockitoJUnitRunner.class)
|
||||
public class MembershipListenerAdapterUnitTests {
|
||||
|
||||
@Mock
|
||||
private DistributionManager mockDistributionManager;
|
||||
|
||||
@Mock
|
||||
private InternalDistributedMember mockDistributedMember;
|
||||
|
||||
@Mock
|
||||
private InternalDistributedSystem mockDistributedSystem;
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
private <T> Set<T> asSet(T... elements) {
|
||||
return new HashSet<>(Arrays.asList(elements));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void memberDepartedCallsHandleMemberDeparted() {
|
||||
|
||||
MembershipListenerAdapter<?> listener = spy(new TestMembershipListener());
|
||||
|
||||
doAnswer(invocation -> {
|
||||
|
||||
MemberDepartedEvent event = invocation.getArgument(0);
|
||||
|
||||
assertThat(event).isNotNull();
|
||||
assertThat(event.isCrashed()).isTrue();
|
||||
assertThat(event.getDistributedMember().orElse(null)).isEqualTo(this.mockDistributedMember);
|
||||
assertThat(event.getDistributionManager()).isEqualTo(this.mockDistributionManager);
|
||||
assertThat(event.getType()).isEqualTo(MembershipEvent.Type.MEMBER_DEPARTED);
|
||||
|
||||
return null;
|
||||
|
||||
}).when(listener).handleMemberDeparted(any(MemberDepartedEvent.class));
|
||||
|
||||
listener.memberDeparted(this.mockDistributionManager, this.mockDistributedMember, true);
|
||||
|
||||
verify(listener, times(1)).handleMemberDeparted(isA(MemberDepartedEvent.class));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void memberJoinedCallsHandleMemberJoined() {
|
||||
|
||||
MembershipListenerAdapter<?> listener = spy(new TestMembershipListener());
|
||||
|
||||
doAnswer(invocation -> {
|
||||
|
||||
MemberJoinedEvent event = invocation.getArgument(0);
|
||||
|
||||
assertThat(event).isNotNull();
|
||||
assertThat(event.getDistributedMember().orElse(null)).isEqualTo(this.mockDistributedMember);
|
||||
assertThat(event.getDistributionManager()).isEqualTo(this.mockDistributionManager);
|
||||
assertThat(event.getType()).isEqualTo(MembershipEvent.Type.MEMBER_JOINED);
|
||||
|
||||
return null;
|
||||
|
||||
}).when(listener).handleMemberJoined(any(MemberJoinedEvent.class));
|
||||
|
||||
listener.memberJoined(this.mockDistributionManager, this.mockDistributedMember);
|
||||
|
||||
verify(listener, times(1)).handleMemberJoined(isA(MemberJoinedEvent.class));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void memberSuspectCallsHandleMemberSuspect() {
|
||||
|
||||
InternalDistributedMember suspectMember = mock(InternalDistributedMember.class);
|
||||
|
||||
MembershipListenerAdapter<?> listener = spy(new TestMembershipListener());
|
||||
|
||||
doAnswer(invocation -> {
|
||||
|
||||
MemberSuspectEvent event = invocation.getArgument(0);
|
||||
|
||||
assertThat(event).isNotNull();
|
||||
assertThat(event.getDistributedMember().orElse(null)).isEqualTo(this.mockDistributedMember);
|
||||
assertThat(event.getDistributionManager()).isEqualTo(this.mockDistributionManager);
|
||||
assertThat(event.getReason().orElse(null)).isEqualTo("The system sucks!");
|
||||
assertThat(event.getSuspectMember().orElse(null)).isEqualTo(suspectMember);
|
||||
assertThat(event.getType()).isEqualTo(MembershipEvent.Type.MEMBER_SUSPECT);
|
||||
|
||||
return null;
|
||||
|
||||
}).when(listener).handleMemberSuspect(any(MemberSuspectEvent.class));
|
||||
|
||||
listener.memberSuspect(this.mockDistributionManager, this.mockDistributedMember, suspectMember,
|
||||
"The system sucks!");
|
||||
|
||||
verify(listener, times(1)).handleMemberSuspect(isA(MemberSuspectEvent.class));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void quorumLostCallsHandleQuorumLost() {
|
||||
|
||||
InternalDistributedMember mockMemberOne = mock(InternalDistributedMember.class);
|
||||
InternalDistributedMember mockMemberTwo = mock(InternalDistributedMember.class);
|
||||
|
||||
MembershipListenerAdapter<?> listener = spy(new TestMembershipListener());
|
||||
|
||||
doAnswer(invocation -> {
|
||||
|
||||
QuorumLostEvent event = invocation.getArgument(0);
|
||||
|
||||
assertThat(event).isNotNull();
|
||||
assertThat(event.getDistributedMember().orElse(null)).isNull();
|
||||
assertThat(event.getDistributionManager()).isEqualTo(this.mockDistributionManager);
|
||||
assertThat(event.getFailedMembers()).isEqualTo(asSet(mockMemberOne, mockMemberTwo));
|
||||
assertThat(event.getRemainingMembers()).isEqualTo(Collections.singletonList(this.mockDistributedMember));
|
||||
assertThat(event.getType()).isEqualTo(MembershipEvent.Type.QUORUM_LOST);
|
||||
|
||||
return null;
|
||||
|
||||
}).when(listener).handleQuorumLost(any(QuorumLostEvent.class));
|
||||
|
||||
listener.quorumLost(this.mockDistributionManager, asSet(mockMemberOne, mockMemberTwo),
|
||||
Collections.singletonList(this.mockDistributedMember));
|
||||
|
||||
verify(listener, times(1)).handleQuorumLost(isA(QuorumLostEvent.class));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void registersListenerWithPeerCache() {
|
||||
|
||||
Cache mockCache = mock(Cache.class);
|
||||
|
||||
doReturn(this.mockDistributedSystem).when(mockCache).getDistributedSystem();
|
||||
doReturn(this.mockDistributionManager).when(this.mockDistributedSystem).getDistributionManager();
|
||||
|
||||
MembershipListenerAdapter<?> listener = new TestMembershipListener();
|
||||
|
||||
assertThat(listener.register(mockCache)).isSameAs(listener);
|
||||
|
||||
verify(mockCache, times(1)).getDistributedSystem();
|
||||
verify(this.mockDistributedSystem, times(1)).getDistributionManager();
|
||||
verify(this.mockDistributionManager, times(1))
|
||||
.addMembershipListener(eq(listener));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void registerListenerWithNullCacheIsNullSafe() {
|
||||
|
||||
MembershipListenerAdapter<?> listener = new TestMembershipListener();
|
||||
|
||||
assertThat(listener.register(null)).isSameAs(listener);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void registerListenerWithNullDistributedSystemIsNullSafe() {
|
||||
|
||||
Cache mockCache = mock(Cache.class);
|
||||
|
||||
MembershipListenerAdapter<?> listener = new TestMembershipListener();
|
||||
|
||||
assertThat(listener.register(mockCache)).isSameAs(listener);
|
||||
|
||||
verify(mockCache, times(1)).getDistributedSystem();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void registerListenerWithNonInternalDistributedSystemIsSafe() {
|
||||
|
||||
Cache mockCache = mock(Cache.class);
|
||||
|
||||
DistributedSystem mockDistributedSystem = mock(DistributedSystem.class);
|
||||
|
||||
doReturn(mockDistributedSystem).when(mockCache).getDistributedSystem();
|
||||
|
||||
MembershipListenerAdapter<?> listener = new TestMembershipListener();
|
||||
|
||||
assertThat(listener.register(mockCache)).isSameAs(listener);
|
||||
|
||||
verify(mockCache, times(1)).getDistributedSystem();
|
||||
verifyNoInteractions(mockDistributedSystem);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void registerListenerWithNullDistributionManagerIsNullSafe() {
|
||||
|
||||
Cache mockCache = mock(Cache.class);
|
||||
|
||||
doReturn(this.mockDistributedSystem).when(mockCache).getDistributedSystem();
|
||||
|
||||
MembershipListenerAdapter<?> listener = new TestMembershipListener();
|
||||
|
||||
assertThat(listener.register(mockCache)).isSameAs(listener);
|
||||
|
||||
verify(mockCache, times(1)).getDistributedSystem();
|
||||
verify(this.mockDistributedSystem, times(1)).getDistributionManager();
|
||||
}
|
||||
|
||||
static class TestMembershipListener extends MembershipListenerAdapter<TestMembershipListener> { }
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user