Refactoring

src/main/java/org/redisson/RedissonLock.java

@@ -179,7 +179,7 @@ public class RedissonLock extends RedissonExpirable implements RLock {
     }
 
 
-    Long tryLockInner(final long leaseTime, final TimeUnit unit) {
+    Long tryLockInner(long leaseTime, TimeUnit unit) {
         internalLockLeaseTime = unit.toMillis(leaseTime);
 
         return commandExecutor.evalWrite(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_LONG,

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

@@ -17,9 +17,6 @@ package org.redisson.pubsub;
 
 import org.redisson.RedissonCountDownLatch;
 import org.redisson.RedissonCountDownLatchEntry;
-import org.redisson.client.BaseRedisPubSubListener;
-import org.redisson.client.RedisPubSubListener;
-import org.redisson.client.protocol.pubsub.PubSubType;
 
 import io.netty.util.concurrent.Promise;
 
@@ -31,15 +28,7 @@ public class CountDownLatchPubSub extends PublishSubscribe<RedissonCountDownLatc
     }
 
     @Override
-    protected RedisPubSubListener<Long> createListener(final String channelName, final RedissonCountDownLatchEntry value) {
-        RedisPubSubListener<Long> listener = new BaseRedisPubSubListener<Long>() {
-
-            @Override
-            public void onMessage(String channel, Long message) {
-                if (!channelName.equals(channel)) {
-                    return;
-                }
-
+    protected void onMessage(RedissonCountDownLatchEntry value, Long message) {
         if (message.equals(RedissonCountDownLatch.zeroCountMessage)) {
             value.getLatch().open();
         }
@@ -48,22 +37,4 @@ public class CountDownLatchPubSub extends PublishSubscribe<RedissonCountDownLatc
         }
     }
 
-            @Override
-            public boolean onStatus(PubSubType type, String channel) {
-                if (!channelName.equals(channel)) {
-                    return false;
-                }
-
-                if (type == PubSubType.SUBSCRIBE) {
-                    value.getPromise().trySuccess(value);
-                    return true;
-                }
-                return false;
-            }
-
-        };
-        return listener;
-    }
-
-
 }

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

@@ -17,9 +17,6 @@ package org.redisson.pubsub;
 
 import org.redisson.RedissonLock;
 import org.redisson.RedissonLockEntry;
-import org.redisson.client.BaseRedisPubSubListener;
-import org.redisson.client.RedisPubSubListener;
-import org.redisson.client.protocol.pubsub.PubSubType;
 
 import io.netty.util.concurrent.Promise;
 
@@ -31,36 +28,10 @@ public class LockPubSub extends PublishSubscribe<RedissonLockEntry> {
     }
 
     @Override
-    protected RedisPubSubListener<Long> createListener(final String channelName, final RedissonLockEntry value) {
-        RedisPubSubListener<Long> listener = new BaseRedisPubSubListener<Long>() {
-
-            @Override
-            public void onMessage(String channel, Long message) {
-                if (!channelName.equals(channel)) {
-                    return;
-                }
-
+    protected void onMessage(RedissonLockEntry value, Long message) {
         if (message.equals(RedissonLock.unlockMessage)) {
             value.getLatch().release();
         }
     }
 
-            @Override
-            public boolean onStatus(PubSubType type, String channel) {
-                if (!channelName.equals(channel)) {
-                    return false;
-                }
-
-                if (type == PubSubType.SUBSCRIBE) {
-                    value.getPromise().trySuccess(value);
-                    return true;
-                }
-                return false;
-            }
-
-        };
-        return listener;
-    }
-
-
 }

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

@@ -18,8 +18,10 @@ package org.redisson.pubsub;
 import java.util.concurrent.ConcurrentMap;
 
 import org.redisson.PubSubEntry;
+import org.redisson.client.BaseRedisPubSubListener;
 import org.redisson.client.RedisPubSubListener;
 import org.redisson.client.codec.LongCodec;
+import org.redisson.client.protocol.pubsub.PubSubType;
 import org.redisson.connection.ConnectionManager;
 
 import io.netty.util.concurrent.Future;
@@ -72,5 +74,35 @@ abstract class PublishSubscribe<E extends PubSubEntry<E>> {
 
     protected abstract E createEntry(Promise<E> newPromise);
 
-    protected abstract RedisPubSubListener<Long> createListener(String channelName, E value);
+    protected abstract void onMessage(E value, Long message);
+
+    private RedisPubSubListener<Long> createListener(final String channelName, final E value) {
+        RedisPubSubListener<Long> listener = new BaseRedisPubSubListener<Long>() {
+
+            @Override
+            public void onMessage(String channel, Long message) {
+                if (!channelName.equals(channel)) {
+                    return;
+                }
+
+                PublishSubscribe.this.onMessage(value, message);
+            }
+
+            @Override
+            public boolean onStatus(PubSubType type, String channel) {
+                if (!channelName.equals(channel)) {
+                    return false;
+                }
+
+                if (type == PubSubType.SUBSCRIBE) {
+                    value.getPromise().trySuccess(value);
+                    return true;
+                }
+                return false;
+            }
+
+        };
+        return listener;
+    }
+
 }