Encoder/Decoder added

redisson · Jul 5, 2015 · 93eb8fa · 93eb8fa
1 parent 394c953
commit 93eb8fa
Show file tree

Hide file tree

Showing 14 changed files with 197 additions and 98 deletions.
diff --git a/src/main/java/org/redisson/client/RedisClient.java b/src/main/java/org/redisson/client/RedisClient.java
@@ -21,8 +21,10 @@
 import org.redisson.client.handler.RedisData;
 import org.redisson.client.handler.RedisDecoder;
 import org.redisson.client.handler.RedisEncoder;
+import org.redisson.client.protocol.Codec;
 import org.redisson.client.protocol.RedisCommand;
 import org.redisson.client.protocol.RedisCommands;
+import org.redisson.client.protocol.StringCodec;
 
 import io.netty.bootstrap.Bootstrap;
 import io.netty.channel.Channel;
@@ -66,27 +68,38 @@ public ChannelFuture connect() {
         return future;
     }
 
-    public <R> Future<R> execute(RedisCommand<R> command, Object ... params) {
+//    public <R> Future<R> execute(Codec encoder, RedisCommand<R> command, Object ... params) {
+//        Promise<R> promise = bootstrap.group().next().<R>newPromise();
+//        channel.writeAndFlush(new RedisData<R, R>(promise, encoder, command, params));
+//        return promise;
+//    }
+
+    public <T, R> Future<R> execute(Codec encoder, RedisCommand<T> command, Object ... params) {
         Promise<R> promise = bootstrap.group().next().<R>newPromise();
-        channel.writeAndFlush(new RedisData<R>(promise, command, params));
+        channel.writeAndFlush(new RedisData<T, R>(promise, encoder, command, params));
         return promise;
     }
 
     public static void main(String[] args) throws InterruptedException {
         RedisClient rc = new RedisClient("127.0.0.1", 6379);
         rc.connect().sync();
-        for (int i = 0; i < 10000; i++) {
-            final int j = i;
-            Future<String> res = rc.execute(RedisCommands.SET, "test", "" + Math.random());
-            res.addListener(new FutureListener<String>() {
-
-                @Override
-                public void operationComplete(Future<String> future) throws Exception {
-                    System.out.println("res: " + future.getNow() + " " + j);
-                }
-
-            });
-        }
-//        rc.execute(RedisCommands.GET, "test");
+        Future<String> res = rc.execute(new StringCodec(), RedisCommands.SET, "test", "" + Math.random());
+        res.addListener(new FutureListener<String>() {
+
+            @Override
+            public void operationComplete(Future<String> future) throws Exception {
+                System.out.println("res 1: " + future.getNow());
+            }
+
+        });
+
+        Future<String> r = rc.execute(new StringCodec(), RedisCommands.GET, "test");
+        r.addListener(new FutureListener<Object>() {
+
+            @Override
+            public void operationComplete(Future<Object> future) throws Exception {
+                System.out.println("res 2: " + future.getNow());
+            }
+        });
     }
 }
diff --git a/src/main/java/org/redisson/client/RedisException.java b/src/main/java/org/redisson/client/RedisException.java
@@ -0,0 +1,15 @@
+package org.redisson.client;
+
+public class RedisException extends RuntimeException {
+
+    private static final long serialVersionUID = 3389820652701696154L;
+
+    public RedisException(String message, Throwable cause) {
+        super(message, cause);
+    }
+
+    public RedisException(String message) {
+        super(message);
+    }
+
+}
diff --git a/src/main/java/org/redisson/client/handler/RedisCommandsQueue.java b/src/main/java/org/redisson/client/handler/RedisCommandsQueue.java
@@ -28,9 +28,9 @@ public class RedisCommandsQueue extends ChannelDuplexHandler {
 
     public enum QueueCommands {NEXT_COMMAND}
 
-    public static final AttributeKey<Promise<Object>> REPLAY_PROMISE = AttributeKey.valueOf("promise");
+    public static final AttributeKey<RedisData<Object, Object>> REPLAY_PROMISE = AttributeKey.valueOf("promise");
 
-    private final Queue<RedisData<Object>> queue = PlatformDependent.newMpscQueue();
+    private final Queue<RedisData<Object, Object>> queue = PlatformDependent.newMpscQueue();
 
     @Override
     public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
@@ -45,8 +45,8 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exc
     @Override
     public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
         if (msg instanceof RedisData) {
-            RedisData<Object> data = (RedisData<Object>) msg;
-            if (data.getSend().get()) {
+            RedisData<Object, Object> data = (RedisData<Object, Object>) msg;
+            if (data.getSended().get()) {
                 super.write(ctx, msg, promise);
             } else {
                 queue.add(data);
@@ -58,9 +58,9 @@ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
     }
 
     private void sendData(ChannelHandlerContext ctx) throws Exception {
-        RedisData<Object> data = queue.peek();
-        if (data != null && data.getSend().compareAndSet(false, true)) {
-            ctx.channel().attr(REPLAY_PROMISE).set(data.getPromise());
+        RedisData<Object, Object> data = queue.peek();
+        if (data != null && data.getSended().compareAndSet(false, true)) {
+            ctx.channel().attr(REPLAY_PROMISE).set(data);
             ctx.channel().writeAndFlush(data);
         }
     }

diff --git a/src/main/java/org/redisson/client/handler/RedisData.java b/src/main/java/org/redisson/client/handler/RedisData.java
@@ -17,24 +17,28 @@
 
 import java.util.concurrent.atomic.AtomicBoolean;
 
+import org.redisson.client.protocol.Codec;
+import org.redisson.client.protocol.Encoder;
 import org.redisson.client.protocol.RedisCommand;
 
 import io.netty.util.concurrent.Promise;
 
-public class RedisData<R> {
+public class RedisData<T, R> {
 
-    Promise<R> promise;
-    RedisCommand<R> command;
-    Object[] params;
-    AtomicBoolean send = new AtomicBoolean();
+    final Promise<R> promise;
+    final RedisCommand<T> command;
+    final Object[] params;
+    final Codec codec;
+    final AtomicBoolean sended = new AtomicBoolean();
 
-    public RedisData(Promise<R> promise, RedisCommand<R> command, Object[] params) {
+    public RedisData(Promise<R> promise, Codec encoder, RedisCommand<T> command, Object[] params) {
         this.promise = promise;
         this.command = command;
         this.params = params;
+        this.codec = encoder;
     }
 
-    public RedisCommand<R> getCommand() {
+    public RedisCommand<T> getCommand() {
         return command;
     }
 
@@ -46,8 +50,12 @@ public Promise<R> getPromise() {
         return promise;
     }
 
-    public AtomicBoolean getSend() {
-        return send;
+    public AtomicBoolean getSended() {
+        return sended;
+    }
+
+    public Codec getCodec() {
+        return codec;
     }
 
 }
diff --git a/src/main/java/org/redisson/client/handler/RedisDecoder.java b/src/main/java/org/redisson/client/handler/RedisDecoder.java
@@ -15,56 +15,91 @@
  */
 package org.redisson.client.handler;
 
+import java.io.IOException;
 import java.util.List;
 
+import org.redisson.client.RedisException;
 import org.redisson.client.handler.RedisCommandsQueue.QueueCommands;
+import org.redisson.client.protocol.Decoder;
 
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.codec.ReplayingDecoder;
-import io.netty.util.CharsetUtil;
 
 public class RedisDecoder extends ReplayingDecoder<Void> {
 
+    private static final char CR = '\r';
+    private static final char LF = '\n';
+    private static final char ZERO = '0';
+
     @Override
     protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
-        int code = in.readByte();
-        String status = in.readBytes(in.bytesBefore((byte) '\r')).toString(CharsetUtil.UTF_8);
-        in.skipBytes(2);
-        out.add(status);
-        System.out.println("status: " + status);
+        RedisData<Object, Object> data = ctx.channel().attr(RedisCommandsQueue.REPLAY_PROMISE).getAndRemove();
 
-        ctx.channel().attr(RedisCommandsQueue.REPLAY_PROMISE).getAndRemove().setSuccess(status);
+        int code = in.readByte();
+        if (code == '+') {
+            Object result = data.getCommand().getReponseDecoder().decode(in);
+            data.getPromise().setSuccess(result);
+        } else if (code == '-') {
+            Object result = data.getCommand().getReponseDecoder().decode(in);
+            data.getPromise().setFailure(new RedisException(result.toString()));
+        } else if (code == '$') {
+            Decoder<Object> decoder = data.getCommand().getReponseDecoder();
+            if (decoder == null) {
+                decoder = data.getCodec();
+            }
+            Object result = decoder.decode(readBytes(in));
+            data.getPromise().setSuccess(result);
+        } else {
+            throw new IllegalStateException("Can't decode replay");
+        }
 
         ctx.pipeline().fireUserEventTriggered(QueueCommands.NEXT_COMMAND);
-//        switch (code) {
-//          case StatusReply.MARKER: {
-//            String status = is.readBytes(is.bytesBefore((byte) '\r')).toString(Charsets.UTF_8);
-//            is.skipBytes(2);
-//            return new StatusReply(status);
-//          }
-//          case ErrorReply.MARKER: {
-//            String error = is.readBytes(is.bytesBefore((byte) '\r')).toString(Charsets.UTF_8);
-//            is.skipBytes(2);
-//            return new ErrorReply(error);
-//          }
-//          case IntegerReply.MARKER: {
-//            return new IntegerReply(readLong(is));
-//          }
-//          case BulkReply.MARKER: {
-//            return new BulkReply(readBytes(is));
-//          }
-//          case MultiBulkReply.MARKER: {
-//            if (reply == null) {
-//              return decodeMultiBulkReply(is);
-//            } else {
-//              return new RedisReplyDecoder(false).decodeMultiBulkReply(is);
-//            }
-//          }
-//          default: {
-//            throw new IOException("Unexpected character in stream: " + code);
-//          }
-//        }
+    }
+
+    public ByteBuf readBytes(ByteBuf is) throws IOException {
+        long l = readLong(is);
+        if (l > Integer.MAX_VALUE) {
+            throw new IllegalArgumentException(
+                    "Java only supports arrays up to " + Integer.MAX_VALUE + " in size");
+        }
+        int size = (int) l;
+        if (size == -1) {
+            return null;
+        }
+        ByteBuf buffer = is.readSlice(size);
+        int cr = is.readByte();
+        int lf = is.readByte();
+        if (cr != CR || lf != LF) {
+            throw new IOException("Improper line ending: " + cr + ", " + lf);
+        }
+        return buffer;
+    }
+
+    public static long readLong(ByteBuf is) throws IOException {
+        long size = 0;
+        int sign = 1;
+        int read = is.readByte();
+        if (read == '-') {
+            read = is.readByte();
+            sign = -1;
+        }
+        do {
+            if (read == CR) {
+                if (is.readByte() == LF) {
+                    break;
+                }
+            }
+            int value = read - ZERO;
+            if (value >= 0 && value < 10) {
+                size *= 10;
+                size += value;
+            } else {
+                throw new IOException("Invalid character in integer");
+            }
+            read = is.readByte();
+        } while (true);
+        return size * sign;
     }
 
 }
diff --git a/src/main/java/org/redisson/client/handler/RedisEncoder.java b/src/main/java/org/redisson/client/handler/RedisEncoder.java
@@ -22,14 +22,14 @@
 import io.netty.util.CharsetUtil;
 import io.netty.util.concurrent.Promise;
 
-public class RedisEncoder extends MessageToByteEncoder<RedisData<Object>> {
+public class RedisEncoder extends MessageToByteEncoder<RedisData<Object, Object>> {
 
     final char ARGS_PREFIX = '*';
     final char BYTES_PREFIX = '$';
     final byte[] CRLF = "\r\n".getBytes();
 
     @Override
-    protected void encode(ChannelHandlerContext ctx, RedisData<Object> msg, ByteBuf out) throws Exception {
+    protected void encode(ChannelHandlerContext ctx, RedisData<Object, Object> msg, ByteBuf out) throws Exception {
         out.writeByte(ARGS_PREFIX);
         out.writeBytes(toChars(1 + msg.getParams().length));
         out.writeBytes(CRLF);
@@ -39,8 +39,8 @@ protected void encode(ChannelHandlerContext ctx, RedisData<Object> msg, ByteBuf
             writeArgument(out, param.toString().getBytes("UTF-8"));
         }
 
-        String o = out.toString(CharsetUtil.UTF_8);
-        System.out.println(o);
+//        String o = out.toString(CharsetUtil.UTF_8);
+//        System.out.println(o);
     }
 
     private void writeArgument(ByteBuf out, byte[] arg) {

diff --git a/src/main/java/org/redisson/client/protocol/Codec.java b/src/main/java/org/redisson/client/protocol/Codec.java
@@ -0,0 +1,5 @@
+package org.redisson.client.protocol;
+
+public interface Codec extends Encoder, Decoder<Object> {
+
+}
diff --git a/...disson/client/protocol/ObjectDecoder.java → ...org/redisson/client/protocol/Decoder.java b/...disson/client/protocol/ObjectDecoder.java → ...org/redisson/client/protocol/Decoder.java
@@ -15,6 +15,10 @@
  */
 package org.redisson.client.protocol;
 
-public class ObjectDecoder implements ResponseDecoder<Object> {
+import io.netty.buffer.ByteBuf;
+
+public interface Decoder<R> {
+
+    R decode(ByteBuf buf);
 
 }
diff --git a/src/main/java/org/redisson/client/protocol/Encoder.java b/src/main/java/org/redisson/client/protocol/Encoder.java
@@ -0,0 +1,7 @@
+package org.redisson.client.protocol;
+
+public interface Encoder {
+
+    Object encode(Object in);
+
+}
diff --git a/src/main/java/org/redisson/client/protocol/RedisCommand.java b/src/main/java/org/redisson/client/protocol/RedisCommand.java
@@ -17,10 +17,14 @@
 
 public class RedisCommand<R> {
 
-    private String name;
-    private ResponseDecoder<R> reponseDecoder;
+    private final String name;
+    private Decoder<R> reponseDecoder;
 
-    public RedisCommand(String name, ResponseDecoder<R> reponseDecoder) {
+    public RedisCommand(String name) {
+        this(name, null);
+    }
+
+    public RedisCommand(String name, Decoder<R> reponseDecoder) {
         super();
         this.name = name;
         this.reponseDecoder = reponseDecoder;
@@ -30,7 +34,7 @@ public String getName() {
         return name;
     }
 
-    public ResponseDecoder<R> getReponseDecoder() {
+    public Decoder<R> getReponseDecoder() {
         return reponseDecoder;
     }
 

diff --git a/src/main/java/org/redisson/client/protocol/RedisCommands.java b/src/main/java/org/redisson/client/protocol/RedisCommands.java
@@ -17,7 +17,7 @@
 
 public interface RedisCommands {
 
-    RedisCommand<Object> GET = new RedisCommand<Object>("GET", new ObjectDecoder());
-    RedisCommand<String> SET = new RedisCommand<String>("SET", new StringDecoder());
+    RedisCommand<Object> GET = new RedisCommand<Object>("GET");
+    RedisCommand<String> SET = new RedisCommand<String>("SET", new StringReplayDecoder());
 
 }