RedisCommandDecoder.java

output

Proxying *:8000 to localhost:6379 ...
21:25:58.917 [main] DEBUG i.n.u.i.l.InternalLoggerFactory - Using SLF4J as the default logging framework
21:25:58.927 [main] DEBUG i.n.c.MultithreadEventLoopGroup - -Dio.netty.eventLoopThreads: 16
21:25:58.950 [main] DEBUG i.n.util.internal.PlatformDependent - UID: 501
21:25:58.950 [main] DEBUG i.n.util.internal.PlatformDependent - Java version: 7
21:25:58.951 [main] DEBUG i.n.util.internal.PlatformDependent - -Dio.netty.noUnsafe: false
21:25:58.953 [main] DEBUG i.n.util.internal.PlatformDependent0 - java.nio.ByteBuffer.cleaner: available
21:25:58.953 [main] DEBUG i.n.util.internal.PlatformDependent0 - java.nio.Buffer.address: available
21:25:58.954 [main] DEBUG i.n.util.internal.PlatformDependent0 - sun.misc.Unsafe.theUnsafe: available
21:25:58.954 [main] DEBUG i.n.util.internal.PlatformDependent0 - sun.misc.Unsafe.copyMemory: available
21:25:58.955 [main] DEBUG i.n.util.internal.PlatformDependent0 - java.nio.Bits.unaligned: true
21:25:58.955 [main] DEBUG i.n.util.internal.PlatformDependent - sun.misc.Unsafe: available
21:25:58.955 [main] DEBUG i.n.util.internal.PlatformDependent - -Dio.netty.noJavassist: false
21:25:59.068 [main] DEBUG i.n.util.internal.PlatformDependent - Javassist: available
21:25:59.069 [main] DEBUG i.n.util.internal.PlatformDependent - -Dio.netty.noPreferDirect: false
21:25:59.104 [main] DEBUG io.netty.channel.nio.NioEventLoop - -Dio.netty.noKeySetOptimization: false
21:25:59.104 [main] DEBUG io.netty.channel.nio.NioEventLoop - -Dio.netty.selectorAutoRebuildThreshold: 512
21:25:59.156 [main] DEBUG i.n.util.internal.ThreadLocalRandom - -Dio.netty.initialSeedUniquifier: 0xf4306faa2b033eb9
21:25:59.208 [main] DEBUG io.netty.util.NetUtil - Loopback interface: lo0
21:25:59.209 [main] DEBUG io.netty.util.NetUtil - Loopback address: /0:0:0:0:0:0:0:1 (primary)
21:25:59.209 [main] DEBUG io.netty.util.NetUtil - Loopback address: /fe80:0:0:0:0:0:0:1%1
21:25:59.209 [main] DEBUG io.netty.util.NetUtil - Loopback address: /127.0.0.1
21:26:02.839 [nioEventLoopGroup-3-1] DEBUG io.netty.util.ResourceLeakDetector - -Dio.netty.noResourceLeakDetection: false
21:26:03.296 [nioEventLoopGroup-3-12] WARN  io.netty.util.ResourceLeakDetector - LEAK: ByteBuf was GC'd before being released correctly.  The following stack trace shows where the leaked object was created, rather than where you failed to release it.
io.netty.util.ResourceLeakException: io.netty.buffer.UnpooledUnsafeDirectByteBuf@720e6c33
	at io.netty.util.ResourceLeakDetector$DefaultResourceLeak.<init>(ResourceLeakDetector.java:175) ~[netty-all-4.0.12.Final.jar:na]
	at io.netty.util.ResourceLeakDetector.open(ResourceLeakDetector.java:117) ~[netty-all-4.0.12.Final.jar:na]
	at io.netty.buffer.UnpooledUnsafeDirectByteBuf.<init>(UnpooledUnsafeDirectByteBuf.java:72) [netty-all-4.0.12.Final.jar:na]
	at io.netty.buffer.UnpooledByteBufAllocator.newDirectBuffer(UnpooledByteBufAllocator.java:49) [netty-all-4.0.12.Final.jar:na]
	at io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:132) [netty-all-4.0.12.Final.jar:na]
	at io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:123) [netty-all-4.0.12.Final.jar:na]
	at io.netty.buffer.AbstractByteBufAllocator.ioBuffer(AbstractByteBufAllocator.java:84) [netty-all-4.0.12.Final.jar:na]
	at io.netty.channel.AdaptiveRecvByteBufAllocator$HandleImpl.allocate(AdaptiveRecvByteBufAllocator.java:104) [netty-all-4.0.12.Final.jar:na]
	at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:114) [netty-all-4.0.12.Final.jar:na]
	at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:485) [netty-all-4.0.12.Final.jar:na]
	at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:452) [netty-all-4.0.12.Final.jar:na]
	at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:346) [netty-all-4.0.12.Final.jar:na]
	at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:101) [netty-all-4.0.12.Final.jar:na]
	at java.lang.Thread.run(Thread.java:722) [na:1.7.0_09]

RedisCommandDecoder.java

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.util.CharsetUtil;

import java.util.List;

import redgate.protocol.RedisCommand;

public class RedisCommandDecoder extends ByteToMessageDecoder{
	
	private final static int MAX_LONGBUFFER_SIZE	= 10 * 1024;
	final static private ByteBuf REJECTED = Unpooled.copiedBuffer("-ERR Rejected", CharsetUtil.UTF_8);
	final static private ByteBuf RN = Unpooled.copiedBuffer("\r\n", CharsetUtil.UTF_8);
	
	@Override
	protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
		if(super.actualReadableBytes() > MAX_LONGBUFFER_SIZE ){
			reject(ctx, in);
		}
		if(isRedisCommand(in)){
			out.add(in.readBytes(super.actualReadableBytes()));
		}
	}
	
	void reject(ChannelHandlerContext ctx, ByteBuf in) {
		// Too long command, reject and close connection.
		System.out.println("too long command, rejecting. total len was:" + super.actualReadableBytes());
		ctx.write(REJECTED.duplicate().retain());
		ctx.writeAndFlush(RN.duplicate().retain());
		ctx.close();
	}
	
	boolean isRedisCommand(ByteBuf in) {
		// Legacy byte array based logic
		byte[] array = new byte[super.actualReadableBytes()];
		RedisCommand rc = new RedisCommand();
		in.readBytes(array);
		in.resetReaderIndex();
		return rc.parse(array);
	}
}