Fixed exception during channel re-subscription. #87

redisson · Nov 8, 2014 · f80a679 · f80a679
1 parent 57dc4fc
commit f80a679
Show file tree

Hide file tree

Showing 3 changed files with 17 additions and 15 deletions.
diff --git a/src/main/java/org/redisson/RedissonCountDownLatch.java b/src/main/java/org/redisson/RedissonCountDownLatch.java
@@ -49,7 +49,7 @@ public class RedissonCountDownLatch extends RedissonObject implements RCountDown
     private static final Integer newCountMessage = 1;
 
     private static final ConcurrentMap<String, RedissonCountDownLatchEntry> ENTRIES = new ConcurrentHashMap<String, RedissonCountDownLatchEntry>();
-    
+
     private final UUID id;
 
     RedissonCountDownLatch(ConnectionManager connectionManager, String name, UUID id) {
@@ -74,12 +74,13 @@ private Future<Boolean> subscribe() {
             }
             return oldPromise;
         }
-        
+
         RedisPubSubAdapter<Integer> listener = new RedisPubSubAdapter<Integer>() {
 
             @Override
             public void subscribed(String channel, long count) {
-                if (getChannelName().equals(channel)) {
+                if (getChannelName().equals(channel)
+                        && !value.getPromise().isSuccess()) {
                     value.getPromise().setSuccess(true);
                 }
             }
@@ -112,7 +113,7 @@ private void release() {
             RedissonCountDownLatchEntry newEntry = new RedissonCountDownLatchEntry(entry);
             newEntry.release();
             if (ENTRIES.replace(getEntryName(), entry, newEntry)) {
-                if (newEntry.isFree() 
+                if (newEntry.isFree()
                         && ENTRIES.remove(getEntryName(), newEntry)) {
                     Future future = connectionManager.unsubscribe(getChannelName());
                     future.awaitUninterruptibly();
@@ -121,7 +122,7 @@ private void release() {
             }
         }
     }
-    
+
     private Promise<Boolean> aquire() {
         while (true) {
             RedissonCountDownLatchEntry entry = ENTRIES.get(getEntryName());
@@ -141,7 +142,7 @@ public void await() throws InterruptedException {
         Future<Boolean> promise = subscribe();
         try {
             promise.await();
-            
+
             while (getCountInner() > 0) {
                 // waiting for open state
                 RedissonCountDownLatchEntry entry = ENTRIES.get(getEntryName());
@@ -162,7 +163,7 @@ public boolean await(long time, TimeUnit unit) throws InterruptedException {
             if (!promise.await(time, unit)) {
                 return false;
             }
-            
+
             time = unit.toMillis(time);
             while (getCountInner() > 0) {
                 if (time <= 0) {
@@ -178,7 +179,7 @@ public boolean await(long time, TimeUnit unit) throws InterruptedException {
                 long elapsed = System.currentTimeMillis() - current;
                 time = time - elapsed;
             }
-            
+
             return true;
         } finally {
             release();
@@ -213,7 +214,7 @@ public Void execute(RedisConnection<Object, Object> conn) {
     private String getEntryName() {
         return id + getName();
     }
-    
+
     private String getChannelName() {
         return groupName + getName();
     }
@@ -230,7 +231,7 @@ protected Future<Number> execute(RedisAsyncConnection<Object, Number> async) {
                 return async.get(getName());
             }
         });
-        
+
         if (val == null) {
             return 0;
         }
@@ -256,7 +257,7 @@ public Boolean execute(RedisConnection<Object, Object> conn) {
             }
         });
     }
-    
+
     @Override
     public void delete() {
         connectionManager.write(new SyncOperation<Object, Void>() {

diff --git a/src/main/java/org/redisson/RedissonLock.java b/src/main/java/org/redisson/RedissonLock.java
@@ -178,7 +178,8 @@ private Future<Boolean> subscribe() {
 
             @Override
             public void subscribed(String channel, long count) {
-                if (getChannelName().equals(channel)) {
+                if (getChannelName().equals(channel)
+                        && !value.getPromise().isSuccess()) {
                     value.getPromise().setSuccess(true);
                 }
             }

diff --git a/src/test/java/org/redisson/RedissonLockTest.java b/src/test/java/org/redisson/RedissonLockTest.java
@@ -33,15 +33,15 @@ public void after() {
     public void testExpire() throws InterruptedException {
         RLock lock = redisson.getLock("lock");
         lock.lock(2, TimeUnit.SECONDS);
-        
+
         final long startTime = System.currentTimeMillis();
         final CountDownLatch latch = new CountDownLatch(1);
         new Thread() {
             public void run() {
                 RLock lock1 = redisson.getLock("lock");
                 lock1.lock();
                 long spendTime = System.currentTimeMillis() - startTime;
-                Assert.assertTrue(spendTime < 2005);
+                Assert.assertTrue(spendTime < 2010);
                 lock1.unlock();
                 latch.countDown();
             };
@@ -51,7 +51,7 @@ public void run() {
 
         lock.unlock();
     }
-    
+
     @Test
     public void testGetHoldCount() {
         RLock lock = redisson.getLock("lock");