Fix RCountDownLatch only notifying the first async listener after countdown reaches 0.

https://github.com/redisson/redisson/issues/5349

Signed-off-by: Sin Li <sinbios@gmail.com>

redisson/src/main/java/org/redisson/pubsub/CountDownLatchPubSub.java

@@ -42,8 +42,9 @@ public class CountDownLatchPubSub extends PublishSubscribe<RedissonCountDownLatc
     protected void onMessage(RedissonCountDownLatchEntry value, Long message) {
         if (message.equals(ZERO_COUNT_MESSAGE)) {
             Runnable runnableToExecute = value.getListeners().poll();
-            if (runnableToExecute != null) {
+            while (runnableToExecute != null) {
                 runnableToExecute.run();
+                runnableToExecute = value.getListeners().poll();
             }
 
             value.getLatch().open();

redisson/src/test/java/org/redisson/pubsub/CountDownLatchPubSubTest.java

@@ -0,0 +1,32 @@
+package org.redisson.pubsub;
+
+import mockit.Injectable;
+import mockit.Mocked;
+import mockit.Tested;
+import mockit.Verifications;
+
+import org.junit.jupiter.api.Test;
+import org.redisson.RedissonCountDownLatchEntry;
+
+public class CountDownLatchPubSubTest {
+
+  @Tested
+  private CountDownLatchPubSub countDownLatchPubSub;
+
+  @Injectable
+  private PublishSubscribeService publishSubscribeService;
+
+  @Test
+  public void testOnZeroMessageAllListenersExecuted(@Mocked Runnable listener) {
+    int listenCount = 10;
+    RedissonCountDownLatchEntry entry = new RedissonCountDownLatchEntry(null);
+    for (int i = 0; i < listenCount; i++) {
+      entry.addListener(listener);
+    }
+    countDownLatchPubSub.onMessage(entry, CountDownLatchPubSub.ZERO_COUNT_MESSAGE);
+    new Verifications() {{
+      listener.run();
+      times = listenCount;
+    }};
+  }
+}