Issue #242: Add the ability to send many commands with a single call

Also:
* Code cleanup
* Hide internal constants from a user
This commit is contained in:
Trustin Lee 2012-03-30 17:03:02 +09:00
parent fd0b0a4e2b
commit a065b1cee2
3 changed files with 24 additions and 58 deletions

View File

@ -24,11 +24,11 @@ import java.io.IOException;
* Command serialization. * Command serialization.
*/ */
public class Command { public class Command {
public static final byte[] ARGS_PREFIX = "*".getBytes(); static final byte[] ARGS_PREFIX = "*".getBytes();
public static final byte[] CRLF = "\r\n".getBytes(); static final byte[] CRLF = "\r\n".getBytes();
public static final byte[] BYTES_PREFIX = "$".getBytes(); static final byte[] BYTES_PREFIX = "$".getBytes();
public static final byte[] EMPTY_BYTES = new byte[0]; static final byte[] EMPTY_BYTES = new byte[0];
public static final byte[] NEG_ONE_AND_CRLF = convertWithCRLF(-1); static final byte[] NEG_ONE_AND_CRLF = convertWithCRLF(-1);
private byte[][] arguments; private byte[][] arguments;
private Object[] objects; private Object[] objects;

View File

@ -44,7 +44,6 @@ public class RedisDecoder extends ReplayingDecoder<State> {
* via the {@link #readInteger(ChannelBuffer)} method * via the {@link #readInteger(ChannelBuffer)} method
* *
* @param is the {@link ChannelBuffer} to read from * @param is the {@link ChannelBuffer} to read from
* @return content
* @throws IOException is thrown if the line-ending is not CRLF * @throws IOException is thrown if the line-ending is not CRLF
*/ */
public static byte[] readBytes(ChannelBuffer is) throws IOException { public static byte[] readBytes(ChannelBuffer is) throws IOException {
@ -64,10 +63,6 @@ public class RedisDecoder extends ReplayingDecoder<State> {
/** /**
* Read an {@link Integer} from the {@link ChannelBuffer} * Read an {@link Integer} from the {@link ChannelBuffer}
*
* @param is
* @return integer
* @throws IOException
*/ */
public static int readInteger(ChannelBuffer is) throws IOException { public static int readInteger(ChannelBuffer is) throws IOException {
int size = 0; int size = 0;

View File

@ -18,71 +18,42 @@ package io.netty.handler.codec.redis;
import io.netty.buffer.ChannelBuffer; import io.netty.buffer.ChannelBuffer;
import io.netty.buffer.ChannelBuffers; import io.netty.buffer.ChannelBuffers;
import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener; import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.Channels; import io.netty.channel.Channels;
import io.netty.channel.MessageEvent; import io.netty.channel.MessageEvent;
import io.netty.channel.SimpleChannelDownstreamHandler; import io.netty.channel.SimpleChannelDownstreamHandler;
import io.netty.channel.ChannelHandler.Sharable;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
/** /**
* {@link SimpleChannelDownstreamHandler} which encodes {@link Command}'s to {@link ChannelBuffer}'s * {@link SimpleChannelDownstreamHandler} which encodes {@link Command}'s to {@link ChannelBuffer}'s
*
*
*/ */
@Sharable @Sharable
public class RedisEncoder extends SimpleChannelDownstreamHandler { public class RedisEncoder extends SimpleChannelDownstreamHandler {
private final Queue<ChannelBuffer> pool;
/**
* Calls {@link #RedisEncoder(boolean)} with <code>false</code>
*/
public RedisEncoder() {
this(false);
}
/**
* Create a new {@link RedisEncoder} instance
*
* @param poolBuffers <code>true</code> if the {@link ChannelBuffer}'s should be pooled. This should be used with caution as this
* can lead to unnecessary big memory consummation if one of the written values is very big and the rest is very small.
*/
public RedisEncoder(boolean poolBuffers) {
if (poolBuffers) {
pool = new ConcurrentLinkedQueue<ChannelBuffer>();
} else {
pool = null;
}
}
@Override @Override
public void writeRequested(ChannelHandlerContext ctx, MessageEvent e) throws Exception { public void writeRequested(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
Object o = e.getMessage(); Object o = e.getMessage();
if (o instanceof Command) { if (o instanceof Command) {
Command command = (Command) o; ChannelBuffer cb = ChannelBuffers.dynamicBuffer();
ChannelBuffer cb = null;
if (pool != null) {
cb = pool.poll();
}
if (cb == null) {
cb = ChannelBuffers.dynamicBuffer();
}
command.write(cb);
ChannelFuture future = e.getFuture(); ChannelFuture future = e.getFuture();
if (pool != null) { Command command = (Command) o;
final ChannelBuffer finalCb = cb; command.write(cb);
future.addListener(new ChannelFutureListener() { Channels.write(ctx, future, cb);
public void operationComplete(ChannelFuture channelFuture) throws Exception {
finalCb.clear(); } else if (o instanceof Iterable) {
pool.add(finalCb); ChannelBuffer cb = ChannelBuffers.dynamicBuffer();
ChannelFuture future = e.getFuture();
// Useful for transactions and database select
for (Object i : (Iterable<?>) o) {
if (i instanceof Command) {
Command command = (Command) i;
command.write(cb);
} else {
super.writeRequested(ctx, e);
return;
} }
});
} }
Channels.write(ctx, future, cb); Channels.write(ctx, future, cb);
} else { } else {