Refactoring

redisson · Aug 5, 2016 · c3c190e · c3c190e
1 parent 1d0d87b
commit c3c190e
Show file tree

Hide file tree

Showing 10 changed files with 52 additions and 48 deletions.
diff --git a/src/main/java/org/redisson/Redisson.java b/src/main/java/org/redisson/Redisson.java
@@ -79,6 +79,7 @@
 import org.redisson.liveobject.provider.DefaultCodecProvider;
 import org.redisson.liveobject.provider.DefaultResolverProvider;
 import org.redisson.liveobject.provider.ResolverProvider;
+import org.redisson.pubsub.SemaphorePubSub;
 
 import io.netty.util.concurrent.Future;
 import io.netty.util.internal.PlatformDependent;
@@ -100,6 +101,7 @@ public class Redisson implements RedissonClient {
     protected final CodecProvider liveObjectDefaultCodecProvider = new DefaultCodecProvider();
     protected final ResolverProvider liveObjectDefaultResolverProvider = new DefaultResolverProvider();
     protected final Config config;
+    protected final SemaphorePubSub semaphorePubSub = new SemaphorePubSub();
 
     protected final UUID id = UUID.randomUUID();
 
@@ -504,7 +506,7 @@ public RBitSet getBitSet(String name) {
 
     @Override
     public RSemaphore getSemaphore(String name) {
-        return new RedissonSemaphore(commandExecutor, name, id);
+        return new RedissonSemaphore(commandExecutor, name, semaphorePubSub);
     }
 
     @Override

diff --git a/src/main/java/org/redisson/RedissonBlockingDeque.java b/src/main/java/org/redisson/RedissonBlockingDeque.java
@@ -26,6 +26,8 @@
 import org.redisson.command.CommandAsyncExecutor;
 
 import io.netty.util.concurrent.Future;
+import io.netty.util.concurrent.FutureListener;
+import io.netty.util.concurrent.Promise;
 
 /**
  * <p>Distributed and concurrent implementation of {@link java.util.concurrent.BlockingDeque}.
@@ -50,8 +52,19 @@ protected RedissonBlockingDeque(Codec codec, CommandAsyncExecutor commandExecuto
     }
 
     @Override
-    public Future<Boolean> putAsync(V e) {
-        return offerAsync(e);
+    public Future<Void> putAsync(V e) {
+        final Promise<Void> result = commandExecutor.getConnectionManager().newPromise();
+        offerAsync(e).addListener(new FutureListener<Boolean>() {
+            @Override
+            public void operationComplete(Future<Boolean> future) throws Exception {
+                if (!future.isSuccess()) {
+                    result.setFailure(future.cause());
+                    return; 
+                }
+                result.setSuccess(null);
+            }
+        });
+        return result;
     }
 
     /*
@@ -60,7 +73,7 @@ public Future<Boolean> putAsync(V e) {
      */
     @Override
     public void put(V e) throws InterruptedException {
-        offer(e);
+        get(putAsync(e));
     }
 
     @Override

diff --git a/src/main/java/org/redisson/RedissonBlockingQueue.java b/src/main/java/org/redisson/RedissonBlockingQueue.java
@@ -29,6 +29,8 @@
 import org.redisson.connection.decoder.ListDrainToDecoder;
 
 import io.netty.util.concurrent.Future;
+import io.netty.util.concurrent.FutureListener;
+import io.netty.util.concurrent.Promise;
 
 /**
  * <p>Distributed and concurrent implementation of {@link java.util.concurrent.BlockingQueue}.
@@ -50,8 +52,19 @@ protected RedissonBlockingQueue(Codec codec, CommandAsyncExecutor commandExecuto
     }
 
     @Override
-    public Future<Boolean> putAsync(V e) {
-        return offerAsync(e);
+    public Future<Void> putAsync(V e) {
+        final Promise<Void> result = commandExecutor.getConnectionManager().newPromise();
+        offerAsync(e).addListener(new FutureListener<Boolean>() {
+            @Override
+            public void operationComplete(Future<Boolean> future) throws Exception {
+                if (!future.isSuccess()) {
+                    result.setFailure(future.cause());
+                    return; 
+                }
+                result.setSuccess(null);
+            }
+        });
+        return result;
     }
 
     /*
@@ -60,7 +73,7 @@ public Future<Boolean> putAsync(V e) {
      */
     @Override
     public void put(V e) throws InterruptedException {
-        offer(e);
+        get(putAsync(e));
     }
 
     @Override

diff --git a/src/main/java/org/redisson/RedissonBloomFilter.java b/src/main/java/org/redisson/RedissonBloomFilter.java
@@ -167,16 +167,6 @@ public boolean contains(T object) {
         }
     }
 
-    private byte[] encode(T object) {
-        byte[] state = null;
-        try {
-            state = codec.getValueEncoder().encode(object);
-        } catch (IOException e) {
-            throw new IllegalArgumentException(e);
-        }
-        return state;
-    }
-
     private void addConfigCheck(int hashIterations, long size, CommandBatchService executorService) {
         executorService.evalReadAsync(getConfigName(), codec, RedisCommands.EVAL_VOID,
                 "local size = redis.call('hget', KEYS[1], 'size');" +

diff --git a/src/main/java/org/redisson/RedissonObject.java b/src/main/java/org/redisson/RedissonObject.java
@@ -15,6 +15,7 @@
  */
 package org.redisson;
 
+import java.io.IOException;
 import java.util.concurrent.TimeUnit;
 
 import org.redisson.api.RObject;
@@ -132,5 +133,13 @@ public Future<Boolean> isExistsAsync() {
     public Codec getCodec() {
         return codec;
     }
+
+    protected byte[] encode(Object value) {
+        try {
+            return codec.getValueEncoder().encode(value);
+        } catch (IOException e) {
+            throw new IllegalArgumentException(e);
+        }
+    }
 
 }
diff --git a/src/main/java/org/redisson/RedissonQueue.java b/src/main/java/org/redisson/RedissonQueue.java
@@ -43,7 +43,7 @@ protected RedissonQueue(Codec codec, CommandAsyncExecutor commandExecutor, Strin
 
     @Override
     public boolean offer(V e) {
-        return add(e);
+        return get(offerAsync(e));
     }
 
     @Override

diff --git a/src/main/java/org/redisson/RedissonScoredSortedSet.java b/src/main/java/org/redisson/RedissonScoredSortedSet.java
@@ -330,14 +330,6 @@ public boolean retainAll(Collection<?> c) {
         return get(retainAllAsync(c));
     }
 
-    private byte[] encode(V value) {
-        try {
-            return codec.getValueEncoder().encode(value);
-        } catch (IOException e) {
-            throw new IllegalArgumentException(e);
-        }
-    }
-
     @Override
     public Future<Boolean> retainAllAsync(Collection<?> c) {
         if (c.isEmpty()) {

diff --git a/src/main/java/org/redisson/RedissonSemaphore.java b/src/main/java/org/redisson/RedissonSemaphore.java
@@ -17,7 +17,6 @@
 
 import java.util.Arrays;
 import java.util.Collections;
-import java.util.UUID;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
@@ -45,20 +44,14 @@
  */
 public class RedissonSemaphore extends RedissonExpirable implements RSemaphore {
 
-    final UUID id;
-
-    private static final SemaphorePubSub PUBSUB = new SemaphorePubSub();
+    private final SemaphorePubSub semaphorePubSub;
 
     final CommandExecutor commandExecutor;
 
-    protected RedissonSemaphore(CommandExecutor commandExecutor, String name, UUID id) {
+    protected RedissonSemaphore(CommandExecutor commandExecutor, String name, SemaphorePubSub semaphorePubSub) {
         super(commandExecutor, name);
         this.commandExecutor = commandExecutor;
-        this.id = id;
-    }
-
-    private String getEntryName() {
-        return id + ":" + getName();
+        this.semaphorePubSub = semaphorePubSub;
     }
 
     String getChannelName() {
@@ -375,15 +368,15 @@ public void run() {
 
 
     private RedissonLockEntry getEntry() {
-        return PUBSUB.getEntry(getEntryName());
+        return semaphorePubSub.getEntry(getName());
     }
 
     private Future<RedissonLockEntry> subscribe() {
-        return PUBSUB.subscribe(getEntryName(), getChannelName(), commandExecutor.getConnectionManager());
+        return semaphorePubSub.subscribe(getName(), getChannelName(), commandExecutor.getConnectionManager());
     }
 
     private void unsubscribe(Future<RedissonLockEntry> future) {
-        PUBSUB.unsubscribe(future.getNow(), getEntryName(), getChannelName(), commandExecutor.getConnectionManager());
+        semaphorePubSub.unsubscribe(future.getNow(), getName(), getChannelName(), commandExecutor.getConnectionManager());
     }
 
     @Override

diff --git a/src/main/java/org/redisson/RedissonSetCache.java b/src/main/java/org/redisson/RedissonSetCache.java
@@ -209,14 +209,6 @@ public Future<Boolean> addAsync(V value, long ttl, TimeUnit unit) {
                 Arrays.<Object>asList(getName()), System.currentTimeMillis(), timeoutDate, objectState);
     }
 
-    private byte[] encode(V value) {
-        try {
-            return codec.getValueEncoder().encode(value);
-        } catch (IOException e) {
-            throw new IllegalArgumentException(e);
-        }
-    }
-
     @Override
     public Future<Boolean> addAsync(V value) {
         return addAsync(value, 92233720368547758L - System.currentTimeMillis(), TimeUnit.MILLISECONDS);

diff --git a/src/main/java/org/redisson/api/RBlockingQueueAsync.java b/src/main/java/org/redisson/api/RBlockingQueueAsync.java
@@ -53,6 +53,6 @@ public interface RBlockingQueueAsync<V> extends RQueueAsync<V> {
 
     Future<V> takeAsync();
 
-    Future<Boolean> putAsync(V e);
+    Future<Void> putAsync(V e);
 
 }