RSemaphore, RPermitExpirableSemaphore miss release messages and wait a defined timeout before a next attempt or hang #5762

RLock, RFencedLock, RReadWriteLock miss unlock messages and wait a defined timeout before a next attempt or hang #5761

redisson/src/main/java/org/redisson/RedissonLockEntry.java

@@ -56,14 +56,31 @@ public class RedissonLockEntry implements PubSubEntry<RedissonLockEntry> {
 
     public void addListener(Runnable listener) {
         listeners.add(listener);
+
+        if (latch.tryAcquire()) {
+            tryRunListener();
+        }
     }
 
-    public boolean removeListener(Runnable listener) {
-        return listeners.remove(listener);
+    public void tryRunListener() {
+        Runnable runnableToExecute = listeners.poll();
+        if (runnableToExecute != null) {
+            runnableToExecute.run();
+        }
     }
 
-    public ConcurrentLinkedQueue<Runnable> getListeners() {
-        return listeners;
+    public void tryRunAllListeners() {
+        while (true) {
+            Runnable runnableToExecute = listeners.poll();
+            if (runnableToExecute == null) {
+                break;
+            }
+            runnableToExecute.run();
+        }
+    }
+
+    public boolean removeListener(Runnable listener) {
+        return listeners.remove(listener);
     }
 
     public Semaphore getLatch() {

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

@@ -41,20 +41,11 @@ public class LockPubSub extends PublishSubscribe<RedissonLockEntry> {
     @Override
     protected void onMessage(RedissonLockEntry value, Long message) {
         if (message.equals(UNLOCK_MESSAGE)) {
-            Runnable runnableToExecute = value.getListeners().poll();
-            if (runnableToExecute != null) {
-                runnableToExecute.run();
-            }
+            value.tryRunListener();
 
             value.getLatch().release();
         } else if (message.equals(READ_UNLOCK_MESSAGE)) {
-            while (true) {
-                Runnable runnableToExecute = value.getListeners().poll();
-                if (runnableToExecute == null) {
-                    break;
-                }
-                runnableToExecute.run();
-            }
+            value.tryRunAllListeners();
 
             value.getLatch().release(value.getLatch().getQueueLength());
         }

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

@@ -37,10 +37,7 @@ public class SemaphorePubSub extends PublishSubscribe<RedissonLockEntry> {
 
     @Override
     protected void onMessage(RedissonLockEntry value, Long message) {
-        Runnable runnableToExecute = value.getListeners().poll();
-        if (runnableToExecute != null) {
-            runnableToExecute.run();
-        }
+        value.tryRunListener();
 
         value.getLatch().release(Math.min(value.acquired(), message.intValue()));
     }