Data encoding should be executed on client thread only #1083

redisson · Oct 2, 2017 · a41d7db · a41d7db
1 parent c45fbde
commit a41d7db
Show file tree

Hide file tree

Showing 33 changed files with 332 additions and 569 deletions.
diff --git a/redisson/src/main/java/org/redisson/RedissonBoundedBlockingQueue.java b/redisson/src/main/java/org/redisson/RedissonBoundedBlockingQueue.java
@@ -176,7 +176,7 @@ public RFuture<Boolean> removeAllAsync(Collection<?> c) {
         }
 
         String channelName = RedissonSemaphore.getChannelName(getSemaphoreName());
-        return commandExecutor.evalWriteAsync(getName(), codec, RedisCommands.EVAL_BOOLEAN_WITH_VALUES_6,
+        return commandExecutor.evalWriteAsync(getName(), codec, RedisCommands.EVAL_BOOLEAN,
                         "local count = 0; " +
                         "for i = 1, #ARGV, 1 do "
                             + "if redis.call('lrem', KEYS[1], 0, ARGV[i]) == 1 then "
@@ -189,7 +189,7 @@ public RFuture<Boolean> removeAllAsync(Collection<?> c) {
                             + "return 1;"
                         + "end;"
                        + "return 0 ",
-                       Arrays.<Object>asList(getName(), getSemaphoreName(), channelName), c.toArray());
+                       Arrays.<Object>asList(getName(), getSemaphoreName(), channelName), encode(c).toArray());
     }
 
     @Override

diff --git a/redisson/src/main/java/org/redisson/RedissonBuckets.java b/redisson/src/main/java/org/redisson/RedissonBuckets.java
@@ -71,7 +71,7 @@ public <V> Map<String, V> get(String... keys) {
             return Collections.emptyMap();
         }
 
-        RedisCommand<Map<Object, Object>> command = new RedisCommand<Map<Object, Object>>("MGET", new MapGetAllDecoder(Arrays.<Object>asList(keys), 0), ValueType.OBJECTS);
+        RedisCommand<Map<Object, Object>> command = new RedisCommand<Map<Object, Object>>("MGET", new MapGetAllDecoder(Arrays.<Object>asList(keys), 0));
         RFuture<Map<String, V>> future = commandExecutor.readAsync(keys[0], new DelegateDecoderCodec(codec), command, keys);
         return commandExecutor.get(future);
     }

diff --git a/redisson/src/main/java/org/redisson/RedissonDelayedQueue.java b/redisson/src/main/java/org/redisson/RedissonDelayedQueue.java
@@ -28,11 +28,7 @@
 import org.redisson.api.RTopic;
 import org.redisson.client.codec.Codec;
 import org.redisson.client.codec.LongCodec;
-import org.redisson.client.protocol.RedisCommand;
-import org.redisson.client.protocol.RedisCommand.ValueType;
 import org.redisson.client.protocol.RedisCommands;
-import org.redisson.client.protocol.convertor.BooleanReplayConvertor;
-import org.redisson.client.protocol.convertor.VoidReplayConvertor;
 import org.redisson.command.CommandAsyncExecutor;
 
 import io.netty.util.internal.ThreadLocalRandom;
@@ -45,8 +41,6 @@
  */
 public class RedissonDelayedQueue<V> extends RedissonExpirable implements RDelayedQueue<V> {
 
-    private static final RedisCommand<Void> EVAL_OFFER = new RedisCommand<Void>("EVAL", new VoidReplayConvertor(), 9);
-
     private final QueueTransferService queueTransferService;
 
     protected RedissonDelayedQueue(QueueTransferService queueTransferService, Codec codec, final CommandAsyncExecutor commandExecutor, String name) {
@@ -108,7 +102,7 @@ public RFuture<Void> offerAsync(V e, long delay, TimeUnit timeUnit) {
         long timeout = System.currentTimeMillis() + delayInMs;
 
         long randomId = ThreadLocalRandom.current().nextLong();
-        return commandExecutor.evalWriteAsync(getName(), codec, EVAL_OFFER,
+        return commandExecutor.evalWriteAsync(getName(), codec, RedisCommands.EVAL_VOID,
                 "local value = struct.pack('dLc0', tonumber(ARGV[2]), string.len(ARGV[3]), ARGV[3]);" 
               + "redis.call('zadd', KEYS[2], ARGV[1], value);"
               + "redis.call('rpush', KEYS[3], value);"
@@ -120,7 +114,7 @@ public RFuture<Void> offerAsync(V e, long delay, TimeUnit timeUnit) {
               + "end;"
                  ,
               Arrays.<Object>asList(getName(), getTimeoutSetName(), getQueueName(), getChannelName()), 
-              timeout, randomId, e);
+              timeout, randomId, encode(e));
     }
 
     @Override
@@ -286,7 +280,7 @@ public RFuture<Boolean> removeAsync(Object o) {
     }
 
     protected RFuture<Boolean> removeAsync(Object o, int count) {
-        return commandExecutor.evalWriteAsync(getName(), codec, new RedisCommand<Boolean>("EVAL", new BooleanReplayConvertor(), 5),
+        return commandExecutor.evalWriteAsync(getName(), codec, RedisCommands.EVAL_BOOLEAN,
                 "local s = redis.call('llen', KEYS[1]);" +
                 "for i = 0, s-1, 1 do "
                     + "local v = redis.call('lindex', KEYS[1], i);"
@@ -298,7 +292,7 @@ protected RFuture<Boolean> removeAsync(Object o, int count) {
                     + "end; "
                + "end;" +
                "return 0;",
-        Arrays.<Object>asList(getQueueName(), getTimeoutSetName()), o);
+        Arrays.<Object>asList(getQueueName(), getTimeoutSetName()), encode(o));
     }
 
     @Override
@@ -307,7 +301,7 @@ public RFuture<Boolean> containsAllAsync(Collection<?> c) {
             return newSucceededFuture(true);
         }
 
-        return commandExecutor.evalReadAsync(getName(), codec, RedisCommands.EVAL_BOOLEAN_WITH_VALUES,
+        return commandExecutor.evalReadAsync(getName(), codec, RedisCommands.EVAL_BOOLEAN,
                 "local s = redis.call('llen', KEYS[1]);" +
                 "for i = 0, s-1, 1 do "
                     + "local v = redis.call('lindex', KEYS[1], i);"
@@ -320,7 +314,7 @@ public RFuture<Boolean> containsAllAsync(Collection<?> c) {
                     + "end; "
                + "end;" +
                "return #ARGV == 0 and 1 or 0;",
-                Collections.<Object>singletonList(getQueueName()), c.toArray());
+                Collections.<Object>singletonList(getQueueName()), encode(c).toArray());
     }
 
     @Override
@@ -339,7 +333,7 @@ public RFuture<Boolean> removeAllAsync(Collection<?> c) {
             return newSucceededFuture(false);
         }
 
-        return commandExecutor.evalWriteAsync(getName(), codec, new RedisCommand<Boolean>("EVAL", new BooleanReplayConvertor(), 5, ValueType.OBJECTS),
+        return commandExecutor.evalWriteAsync(getName(), codec, RedisCommands.EVAL_BOOLEAN,
                 "local result = 0;" + 
                 "local s = redis.call('llen', KEYS[1]);" + 
                 "local i = 0;" +
@@ -360,7 +354,7 @@ public RFuture<Boolean> removeAllAsync(Collection<?> c) {
                     + "i = i + 1;"
                + "end; " 
                + "return result;",
-               Arrays.<Object>asList(getQueueName(), getTimeoutSetName()), c.toArray());
+               Arrays.<Object>asList(getQueueName(), getTimeoutSetName()), encode(c).toArray());
     }
 
     @Override
@@ -379,7 +373,7 @@ public RFuture<Boolean> retainAllAsync(Collection<?> c) {
             return deleteAsync();
         }
 
-        return commandExecutor.evalWriteAsync(getName(), codec, RedisCommands.EVAL_BOOLEAN_WITH_VALUES,
+        return commandExecutor.evalWriteAsync(getName(), codec, RedisCommands.EVAL_BOOLEAN,
                      "local changed = 0; " +
                      "local items = redis.call('lrange', KEYS[1], 0, -1); "
                    + "local i = 1; "
@@ -399,7 +393,7 @@ public RFuture<Boolean> retainAllAsync(Collection<?> c) {
                         + "i = i + 1; "
                    + "end; "
                    + "return changed; ",
-                Collections.<Object>singletonList(getQueueName()), c.toArray());
+                Collections.<Object>singletonList(getQueueName()), encode(c).toArray());
     }  
 
     @Override
@@ -458,7 +452,7 @@ public RFuture<V> pollLastAndOfferFirstToAsync(String queueName) {
 
     @Override
     public RFuture<Boolean> containsAsync(Object o) {
-        return commandExecutor.evalReadAsync(getName(), codec, new RedisCommand<Boolean>("EVAL", new BooleanReplayConvertor(), 4),
+        return commandExecutor.evalReadAsync(getName(), codec, RedisCommands.EVAL_BOOLEAN,
                         "local s = redis.call('llen', KEYS[1]);" +
                         "for i = 0, s-1, 1 do "
                             + "local v = redis.call('lindex', KEYS[1], i);"
@@ -468,7 +462,7 @@ public RFuture<Boolean> containsAsync(Object o) {
                             + "end; "
                        + "end;" +
                        "return 0;",
-                Collections.<Object>singletonList(getQueueName()), o);
+                Collections.<Object>singletonList(getQueueName()), encode(o));
     }
 
     @Override

diff --git a/redisson/src/main/java/org/redisson/RedissonDeque.java b/redisson/src/main/java/org/redisson/RedissonDeque.java
@@ -23,9 +23,7 @@
 import org.redisson.api.RedissonClient;
 import org.redisson.client.codec.Codec;
 import org.redisson.client.protocol.RedisCommand;
-import org.redisson.client.protocol.RedisCommand.ValueType;
 import org.redisson.client.protocol.RedisCommands;
-import org.redisson.client.protocol.convertor.VoidReplayConvertor;
 import org.redisson.client.protocol.decoder.ListFirstObjectDecoder;
 import org.redisson.command.CommandAsyncExecutor;
 
@@ -38,7 +36,6 @@
  */
 public class RedissonDeque<V> extends RedissonQueue<V> implements RDeque<V> {
 
-    private static final RedisCommand<Void> RPUSH_VOID = new RedisCommand<Void>("RPUSH", new VoidReplayConvertor(), 2, ValueType.OBJECTS);
     private static final RedisCommand<Object> LRANGE_SINGLE = new RedisCommand<Object>("LRANGE", new ListFirstObjectDecoder());
 
 
@@ -57,7 +54,7 @@ public void addFirst(V e) {
 
     @Override
     public RFuture<Void> addFirstAsync(V e) {
-        return commandExecutor.writeAsync(getName(), codec, RedisCommands.LPUSH_VOID, getName(), e);
+        return commandExecutor.writeAsync(getName(), codec, RedisCommands.LPUSH_VOID, getName(), encode(e));
     }
 
     @Override
@@ -67,7 +64,7 @@ public void addLast(V e) {
 
     @Override
     public RFuture<Void> addLastAsync(V e) {
-        return commandExecutor.writeAsync(getName(), codec, RPUSH_VOID, getName(), e);
+        return commandExecutor.writeAsync(getName(), codec, RedisCommands.RPUSH_VOID, getName(), encode(e));
     }
 
 
@@ -128,7 +125,7 @@ public boolean offerFirst(V e) {
 
     @Override
     public RFuture<Boolean> offerFirstAsync(V e) {
-        return commandExecutor.writeAsync(getName(), codec, RedisCommands.LPUSH_BOOLEAN, getName(), e);
+        return commandExecutor.writeAsync(getName(), codec, RedisCommands.LPUSH_BOOLEAN, getName(), encode(e));
     }
 
     @Override

diff --git a/redisson/src/main/java/org/redisson/RedissonHyperLogLog.java b/redisson/src/main/java/org/redisson/RedissonHyperLogLog.java
@@ -69,14 +69,14 @@ public void mergeWith(String... otherLogNames) {
 
     @Override
     public RFuture<Boolean> addAsync(V obj) {
-        return commandExecutor.writeAsync(getName(), codec, RedisCommands.PFADD, getName(), obj);
+        return commandExecutor.writeAsync(getName(), codec, RedisCommands.PFADD, getName(), encode(obj));
     }
 
     @Override
     public RFuture<Boolean> addAllAsync(Collection<V> objects) {
         List<Object> args = new ArrayList<Object>(objects.size() + 1);
         args.add(getName());
-        args.addAll(objects);
+        encode(args, objects);
         return commandExecutor.writeAsync(getName(), codec, RedisCommands.PFADD, getName(), args.toArray());
     }