Redis client codec

src/main/java/org/jboss/netty/handler/codec/redis/BulkReply.java

@@ -0,0 +1,26 @@
+package org.jboss.netty.handler.codec.redis;
+
+import org.jboss.netty.buffer.ChannelBuffer;
+
+import java.io.IOException;
+
+public class BulkReply extends Reply {
+    public static final char MARKER = '$';
+    public final byte[] bytes;
+
+    public BulkReply(byte[] bytes) {
+        this.bytes = bytes;
+    }
+
+    public void write(ChannelBuffer os) throws IOException {
+        os.writeByte(MARKER);
+        if (bytes == null) {
+            os.writeBytes(Command.NEG_ONE);
+        } else {
+            os.writeBytes(Command.numToBytes(bytes.length));
+            os.writeBytes(Command.CRLF);
+            os.writeBytes(bytes);
+        }
+        os.writeBytes(Command.CRLF);
+    }
+}

src/main/java/org/jboss/netty/handler/codec/redis/Command.java

@@ -0,0 +1,120 @@
+package org.jboss.netty.handler.codec.redis;
+
+import org.jboss.netty.buffer.ChannelBuffer;
+
+import java.io.IOException;
+
+/**
+ * Command serialization.
+ * User: sam
+ * Date: 7/27/11
+ * Time: 3:04 PM
+ * To change this template use File | Settings | File Templates.
+ */
+public class Command {
+    public static final byte[] ARGS_PREFIX = "*".getBytes();
+    public static final byte[] CRLF = "\r\n".getBytes();
+    public static final byte[] BYTES_PREFIX = "$".getBytes();
+    public static final byte[] EMPTY_BYTES = new byte[0];
+    public static final byte[] NEG_ONE = Command.numToBytes(-1);
+
+    private byte[][] arguments;
+    private Object[] objects;
+
+    public String getName() {
+        if (arguments == null) {
+            Object o = objects[0];
+            if (o instanceof byte[]) {
+                return new String((byte[]) o);
+            } else {
+                return o.toString();
+            }
+        } else {
+            return new String(arguments[0]);
+        }
+    }
+
+    public Command(byte[]... arguments) {
+        this.arguments = arguments;
+    }
+
+    public Command(Object... objects) {
+        this.objects = objects;
+    }
+
+    public void write(ChannelBuffer os) throws IOException {
+        writeDirect(os, objects);
+    }
+
+    public static void writeDirect(ChannelBuffer os, Object... objects) throws IOException {
+        int length = objects.length;
+        byte[][] arguments = new byte[length][];
+        for (int i = 0; i < length; i++) {
+            Object object = objects[i];
+            if (object == null) {
+                arguments[i] = EMPTY_BYTES;
+            } else if (object instanceof byte[]) {
+                arguments[i] = (byte[]) object;
+            } else {
+                arguments[i] = object.toString().getBytes(Reply.UTF_8);
+            }
+        }
+        writeDirect(os, arguments);
+    }
+
+    private static void writeDirect(ChannelBuffer os, byte[][] arguments) throws IOException {
+        os.writeBytes(ARGS_PREFIX);
+        os.writeBytes(Command.numToBytes(arguments.length));
+        os.writeBytes(CRLF);
+        for (byte[] argument : arguments) {
+            os.writeBytes(BYTES_PREFIX);
+            os.writeBytes(Command.numToBytes(argument.length));
+            os.writeBytes(CRLF);
+            os.writeBytes(argument);
+            os.writeBytes(CRLF);
+        }
+    }
+
+    private static final int NUM_MAP_LENGTH = 256;
+    private static byte[][] numMap = new byte[NUM_MAP_LENGTH][];
+
+    static {
+        for (int i = 0; i < NUM_MAP_LENGTH; i++) {
+            numMap[i] = convert(i);
+        }
+    }
+
+    // Optimized for the direct to ASCII bytes case
+    // Could be even more optimized but it is already
+    // about twice as fast as using Long.toString().getBytes()
+    public static byte[] numToBytes(long value) {
+        if (value >= 0 && value < NUM_MAP_LENGTH) {
+            return numMap[((int) value)];
+        } else if (value == -1) {
+            return NEG_ONE;
+        }
+        return convert(value);
+    }
+
+    private static byte[] convert(long value) {
+        boolean negative = value < 0;
+        int index = negative ? 2 : 1;
+        long current = negative ? -value : value;
+        while ((current /= 10) > 0) {
+            index++;
+        }
+        byte[] bytes = new byte[index];
+        if (negative) {
+            bytes[0] = '-';
+        }
+        current = negative ? -value : value;
+        long tmp = current;
+        while ((tmp /= 10) > 0) {
+            bytes[--index] = (byte) ('0' + (current % 10));
+            current = tmp;
+        }
+        bytes[--index] = (byte) ('0' + current);
+        return bytes;
+    }
+
+}

src/main/java/org/jboss/netty/handler/codec/redis/ErrorReply.java

@@ -0,0 +1,29 @@
+package org.jboss.netty.handler.codec.redis;
+
+import org.jboss.netty.buffer.ChannelBuffer;
+
+import java.io.IOException;
+
+/**
+ * Created by IntelliJ IDEA.
+ * User: sam
+ * Date: 7/29/11
+ * Time: 10:23 AM
+ * To change this template use File | Settings | File Templates.
+ */
+public class ErrorReply extends Reply {
+    public static final char MARKER = '-';
+    private static final byte[] ERR = "ERR ".getBytes(UTF_8);
+    public final String error;
+
+    public ErrorReply(String error) {
+        this.error = error;
+    }
+
+    public void write(ChannelBuffer os) throws IOException {
+        os.writeByte(MARKER);
+        os.writeBytes(ERR);
+        os.writeBytes(error.getBytes(UTF_8));
+        os.writeBytes(Command.CRLF);
+    }
+}

src/main/java/org/jboss/netty/handler/codec/redis/IntegerReply.java

@@ -0,0 +1,20 @@
+package org.jboss.netty.handler.codec.redis;
+
+import org.jboss.netty.buffer.ChannelBuffer;
+
+import java.io.IOException;
+
+public class IntegerReply extends Reply {
+    public static final char MARKER = ':';
+    public final long integer;
+
+    public IntegerReply(long integer) {
+        this.integer = integer;
+    }
+
+    public void write(ChannelBuffer os) throws IOException {
+        os.writeByte(MARKER);
+        os.writeBytes(Command.numToBytes(integer));
+        os.writeBytes(Command.CRLF);
+    }
+}

src/main/java/org/jboss/netty/handler/codec/redis/MultiBulkReply.java

@@ -0,0 +1,69 @@
+package org.jboss.netty.handler.codec.redis;
+
+import org.jboss.netty.buffer.ChannelBuffer;
+
+import java.io.IOException;
+
+public class MultiBulkReply extends Reply {
+    public static final char MARKER = '*';
+
+    // State
+    public Object[] byteArrays;
+    private int size;
+    private int num;
+
+    public MultiBulkReply() {
+    }
+
+    public void read(RedisDecoder rd, ChannelBuffer is) throws IOException {
+        if (num == 0) {
+            size = RedisDecoder.readInteger(is);
+            byteArrays = new Object[size];
+            rd.checkpoint();
+        }
+        for (int i = num; i < size; i++) {
+            int read = is.readByte();
+            if (read == BulkReply.MARKER) {
+                byteArrays[i] = rd.readBytes(is);
+            } else if (read == IntegerReply.MARKER) {
+                byteArrays[i] = RedisDecoder.readInteger(is);
+            } else {
+                throw new IOException("Unexpected character in stream: " + read);
+            }
+            num = i;
+            rd.checkpoint();
+        }
+    }
+
+    public MultiBulkReply(Object... values) {
+        this.byteArrays = values;
+    }
+
+    public void write(ChannelBuffer os) throws IOException {
+        os.writeByte(MARKER);
+        if (byteArrays == null) {
+            os.writeBytes(Command.NEG_ONE);
+            os.writeBytes(Command.CRLF);
+        } else {
+            os.writeBytes(Command.numToBytes(byteArrays.length));
+            os.writeBytes(Command.CRLF);
+            for (Object value : byteArrays) {
+                if (value == null) {
+                    os.writeByte(BulkReply.MARKER);
+                    os.writeBytes(Command.NEG_ONE);
+                } else if (value instanceof byte[]) {
+                    byte[] bytes = (byte[]) value;
+                    os.writeByte(BulkReply.MARKER);
+                    int length = bytes.length;
+                    os.writeBytes(Command.numToBytes(length));
+                    os.writeBytes(Command.CRLF);
+                    os.writeBytes(bytes);
+                } else if (value instanceof Number) {
+                    os.writeByte(IntegerReply.MARKER);
+                    os.writeBytes(Command.numToBytes(((Number) value).longValue()));
+                }
+                os.writeBytes(Command.CRLF);
+            }
+        }
+    }
+}

src/main/java/org/jboss/netty/handler/codec/redis/PSubscribeReply.java

@@ -0,0 +1,9 @@
+package org.jboss.netty.handler.codec.redis;
+
+public class PSubscribeReply extends SubscribeReply {
+
+    public PSubscribeReply(byte[][] patterns) {
+        super(patterns);
+    }
+
+}

src/main/java/org/jboss/netty/handler/codec/redis/PUnsubscribeReply.java

@@ -0,0 +1,17 @@
+package org.jboss.netty.handler.codec.redis;
+
+import org.jboss.netty.buffer.ChannelBuffer;
+
+import java.io.IOException;
+
+public class PUnsubscribeReply extends UnsubscribeReply {
+
+    public PUnsubscribeReply(byte[][] patterns) {
+        super(patterns);
+    }
+
+    @Override
+    public void write(ChannelBuffer os) throws IOException {
+
+    }
+}

src/main/java/org/jboss/netty/handler/codec/redis/RedisDecoder.java

@@ -0,0 +1,127 @@
+package org.jboss.netty.handler.codec.redis;
+
+import org.jboss.netty.buffer.ChannelBuffer;
+import org.jboss.netty.channel.Channel;
+import org.jboss.netty.channel.ChannelHandlerContext;
+import org.jboss.netty.handler.codec.replay.ReplayingDecoder;
+
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+
+enum State {
+
+}
+
+public class RedisDecoder extends ReplayingDecoder<State> {
+
+    private static final char CR = '\r';
+    private static final char LF = '\n';
+    private static final char ZERO = '0';
+
+    // We track the current multibulk reply in the case
+    // where we do not get a complete reply in a single
+    // decode invocation.
+    private MultiBulkReply reply;
+
+    public byte[] readBytes(ChannelBuffer is) throws IOException {
+        int size = readInteger(is);
+        if (size == -1) {
+            return null;
+        }
+        if (super.actualReadableBytes() < size + 2) {
+            // Trigger error
+            is.skipBytes(size + 2);
+            throw new AssertionError("Trustin says this isn't possible");
+        }
+        byte[] bytes = new byte[size];
+        is.readBytes(bytes, 0, size);
+        int cr = is.readByte();
+        int lf = is.readByte();
+        if (cr != CR || lf != LF) {
+            throw new IOException("Improper line ending: " + cr + ", " + lf);
+        }
+        return bytes;
+    }
+
+    public static int readInteger(ChannelBuffer is) throws IOException {
+        int 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;
+    }
+
+    public Reply receive(final ChannelBuffer is) throws IOException {
+        int code = is.readByte();
+        switch (code) {
+            case StatusReply.MARKER: {
+                return new StatusReply(new DataInputStream(new ChannelBufferInputStream(is)).readLine());
+            }
+            case ErrorReply.MARKER: {
+                return new ErrorReply(new DataInputStream(new ChannelBufferInputStream(is)).readLine());
+            }
+            case IntegerReply.MARKER: {
+                return new IntegerReply(readInteger(is));
+            }
+            case BulkReply.MARKER: {
+                return new BulkReply(readBytes(is));
+            }
+            case MultiBulkReply.MARKER: {
+                return decodeMultiBulkReply(is);
+            }
+            default: {
+                throw new IOException("Unexpected character in stream: " + code);
+            }
+        }
+    }
+
+    @Override
+    public void checkpoint() {
+        super.checkpoint();
+    }
+
+    @Override
+    protected Object decode(ChannelHandlerContext channelHandlerContext, Channel channel, ChannelBuffer channelBuffer, State anEnum) throws Exception {
+        return receive(channelBuffer);
+    }
+
+    public MultiBulkReply decodeMultiBulkReply(ChannelBuffer is) throws IOException {
+        if (reply == null) {
+            reply = new MultiBulkReply();
+        }
+        reply.read(this, is);
+        return reply;
+    }
+
+    private static class ChannelBufferInputStream extends InputStream {
+        private final ChannelBuffer is;
+
+        public ChannelBufferInputStream(ChannelBuffer is) {
+            this.is = is;
+        }
+
+        @Override
+        public int read() throws IOException {
+            return is.readByte();
+        }
+    }
+}

src/main/java/org/jboss/netty/handler/codec/redis/RedisEncoder.java

@@ -0,0 +1,40 @@
+package org.jboss.netty.handler.codec.redis;
+
+import org.jboss.netty.buffer.ChannelBuffer;
+import org.jboss.netty.buffer.ChannelBuffers;
+import org.jboss.netty.channel.ChannelFuture;
+import org.jboss.netty.channel.ChannelFutureListener;
+import org.jboss.netty.channel.ChannelHandlerContext;
+import org.jboss.netty.channel.Channels;
+import org.jboss.netty.channel.MessageEvent;
+import org.jboss.netty.channel.SimpleChannelDownstreamHandler;
+
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+public class RedisEncoder extends SimpleChannelDownstreamHandler {
+
+    private Queue<ChannelBuffer> pool = new ConcurrentLinkedQueue<ChannelBuffer>();
+
+    @Override
+    public void writeRequested(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
+        Object o = e.getMessage();
+        if (o instanceof Command) {
+            Command command = (Command) o;
+            ChannelBuffer cb = pool.poll();
+            if (cb == null) {
+                cb = ChannelBuffers.dynamicBuffer();
+            }
+            command.write(cb);
+            ChannelFuture future = e.getFuture();
+            final ChannelBuffer finalCb = cb;
+            future.addListener(new ChannelFutureListener() {
+                public void operationComplete(ChannelFuture channelFuture) throws Exception {
+                    finalCb.clear();
+                    pool.add(finalCb);
+                }
+            });
+            Channels.write(ctx, future, cb);
+        }
+    }
+}

src/main/java/org/jboss/netty/handler/codec/redis/Reply.java

@@ -0,0 +1,32 @@
+package org.jboss.netty.handler.codec.redis;
+
+import org.jboss.netty.buffer.ChannelBuffer;
+import org.jboss.netty.buffer.ChannelBuffers;
+
+import java.io.IOException;
+import java.nio.charset.Charset;
+
+/**
+ * Replies.
+ * User: sam
+ * Date: 7/27/11
+ * Time: 3:04 PM
+ * To change this template use File | Settings | File Templates.
+ */
+public abstract class Reply {
+
+    public static final Charset UTF_8 = Charset.forName("UTF-8");
+
+    public abstract void write(ChannelBuffer os) throws IOException;
+
+    public String toString() {
+        ChannelBuffer channelBuffer = ChannelBuffers.dynamicBuffer();
+        try {
+            write(channelBuffer);
+        } catch (IOException e) {
+            throw new AssertionError("Trustin says this won't happen either");
+        }
+        return channelBuffer.toString(UTF_8);
+    }
+
+}

src/main/java/org/jboss/netty/handler/codec/redis/StatusReply.java

@@ -0,0 +1,20 @@
+package org.jboss.netty.handler.codec.redis;
+
+import org.jboss.netty.buffer.ChannelBuffer;
+
+import java.io.IOException;
+
+public class StatusReply extends Reply {
+    public static final char MARKER = '+';
+    public final String status;
+
+    public StatusReply(String status) {
+        this.status = status;
+    }
+
+    public void write(ChannelBuffer os) throws IOException {
+        os.writeByte(MARKER);
+        os.writeBytes(status.getBytes(UTF_8));
+        os.writeBytes(Command.CRLF);
+    }
+}

src/main/java/org/jboss/netty/handler/codec/redis/SubscribeReply.java

@@ -0,0 +1,22 @@
+package org.jboss.netty.handler.codec.redis;
+
+import org.jboss.netty.buffer.ChannelBuffer;
+
+import java.io.IOException;
+
+public class SubscribeReply extends Reply {
+
+    private final byte[][] patterns;
+
+    public SubscribeReply(byte[][] patterns) {
+        this.patterns = patterns;
+    }
+
+    public byte[][] getPatterns() {
+        return patterns;
+    }
+
+    @Override
+    public void write(ChannelBuffer os) throws IOException {
+    }
+}

src/main/java/org/jboss/netty/handler/codec/redis/UnsubscribeReply.java

@@ -0,0 +1,23 @@
+package org.jboss.netty.handler.codec.redis;
+
+import org.jboss.netty.buffer.ChannelBuffer;
+
+import java.io.IOException;
+
+public class UnsubscribeReply extends Reply {
+
+    private final byte[][] patterns;
+
+    public UnsubscribeReply(byte[][] patterns) {
+        this.patterns = patterns;
+    }
+
+    @Override
+    public void write(ChannelBuffer os) throws IOException {
+
+    }
+
+    public byte[][] getPatterns() {
+        return patterns;
+    }
+}

src/test/java/org/jboss/netty/handler/codec/redis/RedisClient.java

@@ -0,0 +1,53 @@
+package org.jboss.netty.handler.codec.redis;
+
+import org.jboss.netty.bootstrap.ClientBootstrap;
+import org.jboss.netty.channel.Channel;
+import org.jboss.netty.channel.ChannelFuture;
+import org.jboss.netty.channel.ChannelPipeline;
+import org.jboss.netty.channel.ChannelPipelineFactory;
+import org.jboss.netty.channel.Channels;
+import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;
+import org.jboss.netty.handler.queue.BlockingReadHandler;
+
+import java.net.InetSocketAddress;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+public class RedisClient {
+  private static final byte[] VALUE = "value".getBytes(Reply.UTF_8);
+
+    public static void main(String[] args) throws Exception {
+    ExecutorService executor = Executors.newCachedThreadPool();
+    final ClientBootstrap cb = new ClientBootstrap(new NioClientSocketChannelFactory(executor, executor));
+    final BlockingReadHandler<Reply> blockingReadHandler = new BlockingReadHandler<Reply>();
+    cb.setPipelineFactory(new ChannelPipelineFactory() {
+      public ChannelPipeline getPipeline() throws Exception {
+        ChannelPipeline pipeline = Channels.pipeline();
+        pipeline.addLast("redisEncoder", new RedisEncoder());
+        pipeline.addLast("redisDecoder", new RedisDecoder());
+        pipeline.addLast("result", blockingReadHandler);
+        return pipeline;
+      }
+    });
+    ChannelFuture redis = cb.connect(new InetSocketAddress("localhost", 6379));
+    redis.await().rethrowIfFailed();
+    Channel channel = redis.getChannel();
+
+    channel.write(new Command("set", "1", "value"));
+    System.out.print(blockingReadHandler.read());
+    channel.write(new Command("get", "1"));
+    System.out.print(blockingReadHandler.read());
+
+    int CALLS = 1000000;
+    long start = System.currentTimeMillis();
+    for (int i = 0; i < CALLS; i++) {
+      channel.write(new Command("SET".getBytes(), Command.numToBytes(i), VALUE));
+      blockingReadHandler.read();
+    }
+    long end = System.currentTimeMillis();
+    System.out.println(CALLS * 1000 / (end - start) + " calls per second");
+
+    channel.close();
+    cb.releaseExternalResources();
+  }
+}