chore(atomix): fix multiple local subscriptions on same topic

camunda · Apr 29, 2020 · 931101f · 931101f
1 parent 68df381
commit 931101f
Show file tree

Hide file tree

Showing 2 changed files with 100 additions and 9 deletions.
diff --git a/...ix/cluster/src/main/java/io/atomix/cluster/messaging/impl/DefaultClusterEventService.java b/...ix/cluster/src/main/java/io/atomix/cluster/messaging/impl/DefaultClusterEventService.java
@@ -300,7 +300,7 @@ void remove(final InternalSubscription subscription) {
   /** Internal topic. */
   private class InternalTopic {
     private final String topic;
-    private final InternalSubscriber subscribers = new InternalSubscriber();
+    private final InternalSubscriber localSubscribers = new InternalSubscriber();
     private final Set<MemberId> subscriptions = Sets.newCopyOnWriteArraySet();
 
     InternalTopic(final String topic) {
@@ -313,7 +313,7 @@ private class InternalTopic {
      * @return the local subscriber for the topic
      */
     InternalSubscriber localSubscriber() {
-      return subscribers;
+      return localSubscribers;
     }
 
     /**
@@ -390,11 +390,13 @@ <M> CompletableFuture<Subscription> subscribe(
      *
      * @param subscription the subscription to register
      */
-    private CompletableFuture<Subscription> addLocalSubscription(
+    private synchronized CompletableFuture<Subscription> addLocalSubscription(
         final InternalSubscription subscription) {
-      subscribers.add(subscription);
+      if (localSubscribers.subscriptions.isEmpty()) {
+        messagingService.registerHandler(subscription.topic(), localSubscribers);
+      }
+      localSubscribers.add(subscription);
       subscriptions.add(localMemberId);
-      messagingService.registerHandler(subscription.topic(), subscribers);
       return updateNodes().thenApply(v -> subscription);
     }
 
@@ -403,11 +405,11 @@ private CompletableFuture<Subscription> addLocalSubscription(
      *
      * @param subscription the subscription to unregister
      */
-    private CompletableFuture<Void> removeLocalSubscription(
+    private synchronized CompletableFuture<Void> removeLocalSubscription(
         final InternalSubscription subscription) {
-      subscribers.remove(subscription);
-      subscriptions.remove(localMemberId);
-      if (subscriptions.isEmpty()) {
+      localSubscribers.remove(subscription);
+      if (localSubscribers.subscriptions.isEmpty()) {
+        subscriptions.remove(localMemberId);
         messagingService.unregisterHandler(subscription.topic());
       }
       return updateNodes();

diff --git a/...luster/src/test/java/io/atomix/cluster/messaging/impl/DefaultClusterEventServiceTest.java b/...luster/src/test/java/io/atomix/cluster/messaging/impl/DefaultClusterEventServiceTest.java
@@ -246,6 +246,95 @@ public void shouldSubscribeMultipleTopics() throws InterruptedException {
     tearDown();
   }
 
+  @Test
+  public void shouldBroadcastToMultipleLocalSubscriptionsForSameTopic()
+      throws InterruptedException {
+    // given
+    this.membersDiscovered = new CountDownLatch(0);
+    final Collection<Node> bootstrapLocations = buildBootstrapNodes(1);
+    final ClusterEventService eventService1 = buildServices(1, bootstrapLocations);
+
+    final Set<Integer> events = new CopyOnWriteArraySet<>();
+    final CountDownLatch latch = new CountDownLatch(2);
+
+    final String topic = "test-topic1";
+
+    eventService1
+        .<String>subscribe(
+            topic,
+            SERIALIZER::decode,
+            message -> {
+              assertEquals("Hello world!", message);
+              events.add(1);
+              latch.countDown();
+            },
+            MoreExecutors.directExecutor())
+        .join();
+
+    eventService1
+        .<String>subscribe(
+            topic,
+            SERIALIZER::decode,
+            message -> {
+              assertEquals("Hello world!", message);
+              events.add(2);
+              latch.countDown();
+            },
+            MoreExecutors.directExecutor())
+        .join();
+
+    // when
+    eventService1.broadcast(topic, "Hello world!", SERIALIZER::encode);
+
+    // then
+    assertThat(latch.await(2, TimeUnit.SECONDS)).isTrue();
+    assertEquals(2, events.size());
+    assertThat(events).containsExactlyInAnyOrder(1, 2);
+
+    tearDown();
+  }
+
+  @Test
+  public void shouldNotCloseOtherSubscriptions() throws InterruptedException {
+    // given
+    this.membersDiscovered = new CountDownLatch(0);
+    final Collection<Node> bootstrapLocations = buildBootstrapNodes(1);
+    final ClusterEventService eventService1 = buildServices(1, bootstrapLocations);
+
+    final Set<Integer> events = new CopyOnWriteArraySet<>();
+    final CountDownLatch latch = new CountDownLatch(1);
+
+    final String topic = "test-topic1";
+
+    final var subscriptionToClose =
+        eventService1
+            .<String>subscribe(
+                topic, SERIALIZER::decode, message -> {}, MoreExecutors.directExecutor())
+            .join();
+
+    eventService1
+        .<String>subscribe(
+            topic,
+            SERIALIZER::decode,
+            message -> {
+              assertEquals("Hello world!", message);
+              events.add(2);
+              latch.countDown();
+            },
+            MoreExecutors.directExecutor())
+        .join();
+
+    // when
+    subscriptionToClose.close().join();
+    eventService1.broadcast(topic, "Hello world!", SERIALIZER::encode);
+
+    // then
+    assertThat(latch.await(2, TimeUnit.SECONDS)).isTrue();
+    assertEquals(1, events.size());
+
+    tearDown();
+  }
+
   @Test
   public void shouldBroadcastAfterRestart() throws InterruptedException {
     // given