Rename ClusterEvent and ClusterEventListener for consistency with Clu…

…sterMembershipService.
atomix · Apr 26, 2018 · d14dadf · d14dadf
1 parent 08cd0ac
commit d14dadf
Show file tree

Hide file tree

Showing 13 changed files with 86 additions and 86 deletions.
diff --git a/.../java/io/atomix/cluster/ClusterEvent.java → ...tomix/cluster/ClusterMembershipEvent.java b/.../java/io/atomix/cluster/ClusterEvent.java → ...tomix/cluster/ClusterMembershipEvent.java
@@ -23,7 +23,7 @@
 /**
  * Describes cluster-related event.
  */
-public class ClusterEvent extends AbstractEvent<ClusterEvent.Type, Member> {
+public class ClusterMembershipEvent extends AbstractEvent<ClusterMembershipEvent.Type, Member> {
 
   /**
    * Type of cluster-related events.
@@ -57,7 +57,7 @@ public enum Type {
    * @param type     cluster event type
    * @param instance cluster device subject
    */
-  public ClusterEvent(Type type, Member instance) {
+  public ClusterMembershipEvent(Type type, Member instance) {
     super(type, instance);
   }
 
@@ -68,7 +68,7 @@ public ClusterEvent(Type type, Member instance) {
    * @param instance event device subject
    * @param time     occurrence time
    */
-  public ClusterEvent(Type type, Member instance, long time) {
+  public ClusterMembershipEvent(Type type, Member instance, long time) {
     super(type, instance, time);
   }
 
@@ -82,8 +82,8 @@ public boolean equals(Object obj) {
     if (this == obj) {
       return true;
     }
-    if (obj instanceof ClusterEvent) {
+    if (obj instanceof ClusterMembershipEvent) {
-      final ClusterEvent other = (ClusterEvent) obj;
+      final ClusterMembershipEvent other = (ClusterMembershipEvent) obj;
       return Objects.equals(this.type(), other.type()) &&
           Objects.equals(this.subject(), other.subject()) &&
           Objects.equals(this.time(), other.time());

diff --git a/.../atomix/cluster/ClusterEventListener.java → ...uster/ClusterMembershipEventListener.java b/.../atomix/cluster/ClusterEventListener.java → ...uster/ClusterMembershipEventListener.java
@@ -20,5 +20,5 @@
 /**
  * Entity capable of receiving device cluster-related events.
  */
-public interface ClusterEventListener extends EventListener<ClusterEvent> {
+public interface ClusterMembershipEventListener extends EventListener<ClusterMembershipEvent> {
 }
diff --git a/cluster/src/main/java/io/atomix/cluster/ClusterMembershipService.java b/cluster/src/main/java/io/atomix/cluster/ClusterMembershipService.java
@@ -23,7 +23,7 @@
  * Service for obtaining information about the individual members within
  * the cluster.
  */
-public interface ClusterMembershipService extends ListenerService<ClusterEvent, ClusterEventListener> {
+public interface ClusterMembershipService extends ListenerService<ClusterMembershipEvent, ClusterMembershipEventListener> {
 
   /**
    * Returns the local member.

diff --git a/cluster/src/main/java/io/atomix/cluster/impl/DefaultClusterMembershipService.java b/cluster/src/main/java/io/atomix/cluster/impl/DefaultClusterMembershipService.java
@@ -19,8 +19,8 @@
 import com.google.common.collect.Maps;
 import com.google.common.collect.Sets;
 import io.atomix.cluster.BootstrapMetadataService;
-import io.atomix.cluster.ClusterEvent;
+import io.atomix.cluster.ClusterMembershipEvent;
-import io.atomix.cluster.ClusterEventListener;
+import io.atomix.cluster.ClusterMembershipEventListener;
 import io.atomix.cluster.ClusterMembershipService;
 import io.atomix.cluster.ClusterMetadataEvent;
 import io.atomix.cluster.ClusterMetadataEventListener;
@@ -63,7 +63,7 @@
  * Default cluster implementation.
  */
 public class DefaultClusterMembershipService
-    extends AbstractListenerManager<ClusterEvent, ClusterEventListener>
+    extends AbstractListenerManager<ClusterMembershipEvent, ClusterMembershipEventListener>
     implements ManagedClusterMembershipService {
 
   private static final Logger LOGGER = getLogger(DefaultClusterMembershipService.class);
@@ -160,8 +160,8 @@ private void broadcastIdentity() {
   private void handleBroadcastMessage(byte[] message) {
     StatefulMember node = SERIALIZER.decode(message);
     if (nodes.putIfAbsent(node.id(), node) == null) {
-      post(new ClusterEvent(ClusterEvent.Type.NODE_ADDED, node));
+      post(new ClusterMembershipEvent(ClusterMembershipEvent.Type.NODE_ADDED, node));
-      post(new ClusterEvent(ClusterEvent.Type.NODE_ACTIVATED, node));
+      post(new ClusterMembershipEvent(ClusterMembershipEvent.Type.NODE_ACTIVATED, node));
       sendHeartbeats();
     }
   }
@@ -223,8 +223,8 @@ private CompletableFuture<Void> sendHeartbeat(Address address, byte[] payload) {
         boolean sendHeartbeats = false;
         for (StatefulMember node : nodes) {
           if (this.nodes.putIfAbsent(node.id(), node) == null) {
-            post(new ClusterEvent(ClusterEvent.Type.NODE_ADDED, node));
+            post(new ClusterMembershipEvent(ClusterMembershipEvent.Type.NODE_ADDED, node));
-            post(new ClusterEvent(ClusterEvent.Type.NODE_ACTIVATED, node));
+            post(new ClusterMembershipEvent(ClusterMembershipEvent.Type.NODE_ACTIVATED, node));
             sendHeartbeats = true;
           }
         }
@@ -275,8 +275,8 @@ private void activateNode(Member member) {
       LOGGER.info("{} - Node activated: {}", localMember.id(), statefulNode);
       statefulNode.setState(State.ACTIVE);
       nodes.put(statefulNode.id(), statefulNode);
-      post(new ClusterEvent(ClusterEvent.Type.NODE_ADDED, statefulNode));
+      post(new ClusterMembershipEvent(ClusterMembershipEvent.Type.NODE_ADDED, statefulNode));
-      post(new ClusterEvent(ClusterEvent.Type.NODE_ACTIVATED, statefulNode));
+      post(new ClusterMembershipEvent(ClusterMembershipEvent.Type.NODE_ACTIVATED, statefulNode));
       sendHeartbeat(member.address(), SERIALIZER.encode(new ClusterHeartbeat(
           localMember.id(),
           localMember.type(),
@@ -287,7 +287,7 @@ private void activateNode(Member member) {
     } else if (existingNode.getState() == State.INACTIVE) {
       LOGGER.info("{} - Node activated: {}", localMember.id(), existingNode);
       existingNode.setState(State.ACTIVE);
-      post(new ClusterEvent(ClusterEvent.Type.NODE_ACTIVATED, existingNode));
+      post(new ClusterMembershipEvent(ClusterMembershipEvent.Type.NODE_ACTIVATED, existingNode));
     }
   }
 
@@ -301,11 +301,11 @@ private void deactivateNode(Member member) {
       existingNode.setState(State.INACTIVE);
       switch (existingNode.type()) {
         case PERSISTENT:
-          post(new ClusterEvent(ClusterEvent.Type.NODE_DEACTIVATED, existingNode));
+          post(new ClusterMembershipEvent(ClusterMembershipEvent.Type.NODE_DEACTIVATED, existingNode));
           break;
         case EPHEMERAL:
-          post(new ClusterEvent(ClusterEvent.Type.NODE_DEACTIVATED, existingNode));
+          post(new ClusterMembershipEvent(ClusterMembershipEvent.Type.NODE_DEACTIVATED, existingNode));
-          post(new ClusterEvent(ClusterEvent.Type.NODE_REMOVED, existingNode));
+          post(new ClusterMembershipEvent(ClusterMembershipEvent.Type.NODE_REMOVED, existingNode));
           break;
         default:
           throw new AssertionError();
@@ -332,7 +332,7 @@ private void handleMetadataEvent(ClusterMetadataEvent event) {
                 node.host(),
                 node.tags());
             nodes.put(newNode.id(), newNode);
-            post(new ClusterEvent(ClusterEvent.Type.NODE_ADDED, newNode));
+            post(new ClusterMembershipEvent(ClusterMembershipEvent.Type.NODE_ADDED, newNode));
           }
           return node.id();
         }).collect(Collectors.toSet());
@@ -350,7 +350,7 @@ private void handleMetadataEvent(ClusterMetadataEvent event) {
     for (MemberId memberId : missingNodes) {
       StatefulMember existingNode = nodes.remove(memberId);
       if (existingNode != null) {
-        post(new ClusterEvent(ClusterEvent.Type.NODE_REMOVED, existingNode));
+        post(new ClusterMembershipEvent(ClusterMembershipEvent.Type.NODE_REMOVED, existingNode));
       }
     }
   }

diff --git a/core/src/test/java/io/atomix/core/AtomixTest.java b/core/src/test/java/io/atomix/core/AtomixTest.java
@@ -15,8 +15,8 @@
  */
 package io.atomix.core;
 
-import io.atomix.cluster.ClusterEvent;
+import io.atomix.cluster.ClusterMembershipEvent;
-import io.atomix.cluster.ClusterEventListener;
+import io.atomix.cluster.ClusterMembershipEventListener;
 import io.atomix.cluster.Member;
 import io.atomix.core.profile.Profile;
 import io.atomix.core.profile.Profiles;
@@ -187,58 +187,58 @@ public void testClientJoinLeaveCore() throws Exception {
     futures.add(startAtomix(Member.Type.PERSISTENT, 3, Arrays.asList(1, 2, 3), Profiles.CONSENSUS));
     Futures.allOf(futures).join();
 
-    TestClusterEventListener dataListener = new TestClusterEventListener();
+    TestClusterMembershipEventListener dataListener = new TestClusterMembershipEventListener();
     instances.get(0).membershipService().addListener(dataListener);
 
     Atomix client1 = startAtomix(Member.Type.EPHEMERAL, 4, Arrays.asList(1, 2, 3), Profiles.CLIENT).join();
     assertEquals(1, client1.partitionService().getPartitionGroups().size());
 
     // client1 added to data node
-    ClusterEvent event1 = dataListener.event();
+    ClusterMembershipEvent event1 = dataListener.event();
-    assertEquals(ClusterEvent.Type.NODE_ADDED, event1.type());
+    assertEquals(ClusterMembershipEvent.Type.NODE_ADDED, event1.type());
     event1 = dataListener.event();
-    assertEquals(ClusterEvent.Type.NODE_ACTIVATED, event1.type());
+    assertEquals(ClusterMembershipEvent.Type.NODE_ACTIVATED, event1.type());
 
     Thread.sleep(1000);
 
-    TestClusterEventListener clientListener = new TestClusterEventListener();
+    TestClusterMembershipEventListener clientListener = new TestClusterMembershipEventListener();
     client1.membershipService().addListener(clientListener);
 
     Atomix client2 = startAtomix(Member.Type.EPHEMERAL, 5, Arrays.asList(1, 2, 3), Profiles.CLIENT).join();
     assertEquals(1, client2.partitionService().getPartitionGroups().size());
 
     // client2 added to data node
-    ClusterEvent event2 = dataListener.event();
+    ClusterMembershipEvent event2 = dataListener.event();
-    assertEquals(ClusterEvent.Type.NODE_ADDED, event2.type());
+    assertEquals(ClusterMembershipEvent.Type.NODE_ADDED, event2.type());
     event2 = dataListener.event();
-    assertEquals(ClusterEvent.Type.NODE_ACTIVATED, event2.type());
+    assertEquals(ClusterMembershipEvent.Type.NODE_ACTIVATED, event2.type());
 
     // client2 added to client node
     event1 = clientListener.event();
-    assertEquals(ClusterEvent.Type.NODE_ADDED, event1.type());
+    assertEquals(ClusterMembershipEvent.Type.NODE_ADDED, event1.type());
     event1 = clientListener.event();
-    assertEquals(ClusterEvent.Type.NODE_ACTIVATED, event1.type());
+    assertEquals(ClusterMembershipEvent.Type.NODE_ACTIVATED, event1.type());
 
     client2.stop().join();
 
     // client2 removed from data node
     event1 = dataListener.event();
-    assertEquals(ClusterEvent.Type.NODE_DEACTIVATED, event1.type());
+    assertEquals(ClusterMembershipEvent.Type.NODE_DEACTIVATED, event1.type());
     event1 = dataListener.event();
-    assertEquals(ClusterEvent.Type.NODE_REMOVED, event1.type());
+    assertEquals(ClusterMembershipEvent.Type.NODE_REMOVED, event1.type());
 
     // client2 removed from client node
     event1 = clientListener.event();
-    assertEquals(ClusterEvent.Type.NODE_DEACTIVATED, event1.type());
+    assertEquals(ClusterMembershipEvent.Type.NODE_DEACTIVATED, event1.type());
     event1 = clientListener.event();
-    assertEquals(ClusterEvent.Type.NODE_REMOVED, event1.type());
+    assertEquals(ClusterMembershipEvent.Type.NODE_REMOVED, event1.type());
   }
 
-  private static class TestClusterEventListener implements ClusterEventListener {
+  private static class TestClusterMembershipEventListener implements ClusterMembershipEventListener {
-    private final BlockingQueue<ClusterEvent> queue = new LinkedBlockingQueue<>();
+    private final BlockingQueue<ClusterMembershipEvent> queue = new LinkedBlockingQueue<>();
 
     @Override
-    public void onEvent(ClusterEvent event) {
+    public void onEvent(ClusterMembershipEvent event) {
       try {
         queue.put(event);
       } catch (InterruptedException e) {
@@ -250,7 +250,7 @@ public boolean eventReceived() {
       return !queue.isEmpty();
     }
 
-    public ClusterEvent event() throws InterruptedException {
+    public ClusterMembershipEvent event() throws InterruptedException {
       return queue.take();
     }
   }

diff --git a/primitive/src/main/java/io/atomix/primitive/partition/impl/DefaultMemberGroupService.java b/primitive/src/main/java/io/atomix/primitive/partition/impl/DefaultMemberGroupService.java
@@ -15,7 +15,7 @@
  */
 package io.atomix.primitive.partition.impl;
 
-import io.atomix.cluster.ClusterEventListener;
+import io.atomix.cluster.ClusterMembershipEventListener;
 import io.atomix.cluster.ClusterMembershipService;
 import io.atomix.primitive.partition.ManagedMemberGroupService;
 import io.atomix.primitive.partition.MemberGroup;
@@ -39,7 +39,7 @@ public class DefaultMemberGroupService
   private final AtomicBoolean started = new AtomicBoolean();
   private final ClusterMembershipService membershipService;
   private final MemberGroupProvider memberGroupProvider;
-  private final ClusterEventListener clusterEventListener = event -> recomputeGroups();
+  private final ClusterMembershipEventListener membershipEventListener = event -> recomputeGroups();
   private volatile Collection<MemberGroup> memberGroups;
 
   public DefaultMemberGroupService(ClusterMembershipService membershipService, MemberGroupProvider memberGroupProvider) {
@@ -60,7 +60,7 @@ private void recomputeGroups() {
   public CompletableFuture<MemberGroupService> start() {
     if (started.compareAndSet(false, true)) {
       memberGroups = memberGroupProvider.getMemberGroups(membershipService.getMembers());
-      membershipService.addListener(clusterEventListener);
+      membershipService.addListener(membershipEventListener);
     }
     return CompletableFuture.completedFuture(this);
   }
@@ -73,7 +73,7 @@ public boolean isRunning() {
   @Override
   public CompletableFuture<Void> stop() {
     if (started.compareAndSet(true, false)) {
-      membershipService.removeListener(clusterEventListener);
+      membershipService.removeListener(membershipEventListener);
     }
     return CompletableFuture.completedFuture(null);
   }

diff --git a/primitive/src/main/java/io/atomix/primitive/partition/impl/HashBasedPrimaryElection.java b/primitive/src/main/java/io/atomix/primitive/partition/impl/HashBasedPrimaryElection.java
@@ -16,7 +16,7 @@
 package io.atomix.primitive.partition.impl;
 
 import com.google.common.hash.Hashing;
-import io.atomix.cluster.ClusterEventListener;
+import io.atomix.cluster.ClusterMembershipEventListener;
 import io.atomix.cluster.ClusterMembershipService;
 import io.atomix.cluster.Member;
 import io.atomix.primitive.partition.GroupMember;
@@ -43,15 +43,15 @@ public class HashBasedPrimaryElection
   private final PartitionId partitionId;
   private final ClusterMembershipService clusterMembershipService;
   private final HashBasedPrimaryElectionService electionService;
-  private final ClusterEventListener clusterEventListener = e -> recomputeTerm();
+  private final ClusterMembershipEventListener membershipEventListener = e -> recomputeTerm();
   private volatile PrimaryTerm currentTerm;
 
   public HashBasedPrimaryElection(PartitionId partitionId, ClusterMembershipService clusterMembershipService, HashBasedPrimaryElectionService electionService) {
     this.partitionId = partitionId;
     this.clusterMembershipService = clusterMembershipService;
     this.electionService = electionService;
     recomputeTerm();
-    clusterMembershipService.addListener(clusterEventListener);
+    clusterMembershipService.addListener(membershipEventListener);
   }
 
   @Override

diff --git a/...ols/primary-backup/src/main/java/io/atomix/protocols/backup/proxy/PrimaryBackupProxy.java b/...ols/primary-backup/src/main/java/io/atomix/protocols/backup/proxy/PrimaryBackupProxy.java
@@ -17,8 +17,8 @@
 
 import com.google.common.base.Throwables;
 import com.google.common.collect.Sets;
-import io.atomix.cluster.ClusterEvent;
+import io.atomix.cluster.ClusterMembershipEvent;
-import io.atomix.cluster.ClusterEventListener;
+import io.atomix.cluster.ClusterMembershipEventListener;
 import io.atomix.cluster.ClusterMembershipService;
 import io.atomix.primitive.PrimitiveException;
 import io.atomix.primitive.PrimitiveType;
@@ -65,7 +65,7 @@ public class PrimaryBackupProxy extends AbstractPrimitiveProxy {
   private final Set<Consumer<State>> stateChangeListeners = Sets.newIdentityHashSet();
   private final Set<Consumer<PrimitiveEvent>> eventListeners = Sets.newIdentityHashSet();
   private final PrimaryElectionEventListener primaryElectionListener = event -> changeReplicas(event.term());
-  private final ClusterEventListener clusterEventListener = this::handleClusterEvent;
+  private final ClusterMembershipEventListener membershipEventListener = this::handleClusterEvent;
   private PrimaryTerm term;
   private volatile State state = State.CLOSED;
 
@@ -212,8 +212,8 @@ private void changeReplicas(PrimaryTerm term) {
   /**
    * Handles a cluster event.
    */
-  private void handleClusterEvent(ClusterEvent event) {
+  private void handleClusterEvent(ClusterMembershipEvent event) {
-    if (event.type() == ClusterEvent.Type.NODE_DEACTIVATED && event.subject().id().equals(term.primary().memberId())) {
+    if (event.type() == ClusterMembershipEvent.Type.NODE_DEACTIVATED && event.subject().id().equals(term.primary().memberId())) {
       threadContext.execute(() -> {
         state = State.SUSPENDED;
         stateChangeListeners.forEach(l -> l.accept(state));
@@ -257,7 +257,7 @@ public CompletableFuture<Void> close() {
       protocol.close(term.primary().memberId(), new CloseRequest(descriptor, sessionId.id()))
           .whenCompleteAsync((response, error) -> {
             protocol.unregisterEventListener(sessionId);
-            clusterMembershipService.removeListener(clusterEventListener);
+            clusterMembershipService.removeListener(membershipEventListener);
             primaryElection.removeListener(primaryElectionListener);
             future.complete(null);
           }, threadContext);

diff --git a/...up/src/main/java/io/atomix/protocols/backup/service/impl/PrimaryBackupServiceContext.java b/...up/src/main/java/io/atomix/protocols/backup/service/impl/PrimaryBackupServiceContext.java
@@ -16,8 +16,8 @@
 
 package io.atomix.protocols.backup.service.impl;
 
-import io.atomix.cluster.ClusterEvent;
+import io.atomix.cluster.ClusterMembershipEvent;
-import io.atomix.cluster.ClusterEventListener;
+import io.atomix.cluster.ClusterMembershipEventListener;
 import io.atomix.cluster.ClusterMembershipService;
 import io.atomix.cluster.MemberId;
 import io.atomix.primitive.PrimitiveId;
@@ -104,7 +104,7 @@ public WallClockTimestamp getTime() {
     }
   };
   private PrimaryBackupRole role;
-  private final ClusterEventListener clusterEventListener = this::handleClusterEvent;
+  private final ClusterMembershipEventListener membershipEventListener = this::handleClusterEvent;
   private final PrimaryElectionEventListener primaryElectionListener = event -> changeRole(event.term());
 
   public PrimaryBackupServiceContext(
@@ -133,7 +133,7 @@ public PrimaryBackupServiceContext(
         .add("type", descriptor.type())
         .add("name", descriptor.name())
         .build());
-    clusterMembershipService.addListener(clusterEventListener);
+    clusterMembershipService.addListener(membershipEventListener);
     primaryElection.addListener(primaryElectionListener);
   }
 
@@ -517,8 +517,8 @@ public PrimaryBackupSession getOrCreateSession(long sessionId, MemberId memberId
   /**
    * Handles a cluster event.
    */
-  private void handleClusterEvent(ClusterEvent event) {
+  private void handleClusterEvent(ClusterMembershipEvent event) {
-    if (event.type() == ClusterEvent.Type.NODE_DEACTIVATED) {
+    if (event.type() == ClusterMembershipEvent.Type.NODE_DEACTIVATED) {
       for (Session session : sessions) {
         if (session.memberId().equals(event.subject().id())) {
           role.expire((PrimaryBackupSession) session);
@@ -575,7 +575,7 @@ private void changeRole(PrimaryTerm term) {
    * Closes the service.
    */
   public CompletableFuture<Void> close() {
-    clusterMembershipService.removeListener(clusterEventListener);
+    clusterMembershipService.removeListener(membershipEventListener);
     primaryElection.removeListener(primaryElectionListener);
     return CompletableFuture.completedFuture(null);
   }