package io.netty.handler.codec.redis;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.util.ByteProcessor;
import io.netty.util.CharsetUtil;
import io.netty.util.internal.UnstableApi;
import java.util.List;
@UnstableApi
public final class RedisDecoder extends ByteToMessageDecoder {
private final ToPositiveLongProcessor toPositiveLongProcessor = new ToPositiveLongProcessor();
private final boolean decodeInlineCommands;
private final int maxInlineMessageLength;
private final RedisMessagePool messagePool;
private State state = State.DECODE_TYPE;
private RedisMessageType type;
private int remainingBulkLength;
private enum State {
DECODE_TYPE,
DECODE_INLINE,
DECODE_LENGTH,
DECODE_BULK_STRING_EOL,
DECODE_BULK_STRING_CONTENT,
}
public RedisDecoder() {
this(false);
}
public RedisDecoder(boolean decodeInlineCommands) {
this(RedisConstants.REDIS_INLINE_MESSAGE_MAX_LENGTH, FixedRedisMessagePool.INSTANCE, decodeInlineCommands);
}
public RedisDecoder(int maxInlineMessageLength, RedisMessagePool messagePool) {
this(maxInlineMessageLength, messagePool, false);
}
public RedisDecoder(int maxInlineMessageLength, RedisMessagePool messagePool, boolean decodeInlineCommands) {
if (maxInlineMessageLength <= 0 || maxInlineMessageLength > RedisConstants.REDIS_MESSAGE_MAX_LENGTH) {
throw new RedisCodecException("maxInlineMessageLength: " + maxInlineMessageLength +
" (expected: <= " + RedisConstants.REDIS_MESSAGE_MAX_LENGTH + ")");
}
this.maxInlineMessageLength = maxInlineMessageLength;
this.messagePool = messagePool;
this.decodeInlineCommands = decodeInlineCommands;
}
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
try {
for (;;) {
switch (state) {
case DECODE_TYPE:
if (!decodeType(in)) {
return;
}
break;
case DECODE_INLINE:
if (!decodeInline(in, out)) {
return;
}
break;
case DECODE_LENGTH:
if (!decodeLength(in, out)) {
return;
}
break;
case DECODE_BULK_STRING_EOL:
if (!decodeBulkStringEndOfLine(in, out)) {
return;
}
break;
case DECODE_BULK_STRING_CONTENT:
if (!decodeBulkStringContent(in, out)) {
return;
}
break;
default:
throw new RedisCodecException("Unknown state: " + state);
}
}
} catch (RedisCodecException e) {
resetDecoder();
throw e;
} catch (Exception e) {
resetDecoder();
throw new RedisCodecException(e);
}
}
private void resetDecoder() {
state = State.DECODE_TYPE;
remainingBulkLength = 0;
}
private boolean decodeType(ByteBuf in) throws Exception {
if (!in.isReadable()) {
return false;
}
type = RedisMessageType.readFrom(in, decodeInlineCommands);
state = type.isInline() ? State.DECODE_INLINE : State.DECODE_LENGTH;
return true;
}
private boolean decodeInline(ByteBuf in, List<Object> out) throws Exception {
ByteBuf lineBytes = readLine(in);
if (lineBytes == null) {
if (in.readableBytes() > maxInlineMessageLength) {
throw new RedisCodecException("length: " + in.readableBytes() +
" (expected: <= " + maxInlineMessageLength + ")");
}
return false;
}
out.add(newInlineRedisMessage(type, lineBytes));
resetDecoder();
return true;
}
private boolean decodeLength(ByteBuf in, List<Object> out) throws Exception {
ByteBuf lineByteBuf = readLine(in);
if (lineByteBuf == null) {
return false;
}
final long length = parseRedisNumber(lineByteBuf);
if (length < RedisConstants.NULL_VALUE) {
throw new RedisCodecException("length: " + length + " (expected: >= " + RedisConstants.NULL_VALUE + ")");
}
switch (type) {
case ARRAY_HEADER:
out.add(new ArrayHeaderRedisMessage(length));
resetDecoder();
return true;
case BULK_STRING:
if (length > RedisConstants.REDIS_MESSAGE_MAX_LENGTH) {
throw new RedisCodecException("length: " + length + " (expected: <= " +
RedisConstants.REDIS_MESSAGE_MAX_LENGTH + ")");
}
remainingBulkLength = (int) length;
return decodeBulkString(in, out);
default:
throw new RedisCodecException("bad type: " + type);
}
}
private boolean decodeBulkString(ByteBuf in, List<Object> out) throws Exception {
switch (remainingBulkLength) {
case RedisConstants.NULL_VALUE:
out.add(FullBulkStringRedisMessage.NULL_INSTANCE);
resetDecoder();
return true;
case 0:
state = State.DECODE_BULK_STRING_EOL;
return decodeBulkStringEndOfLine(in, out);
default:
out.add(new BulkStringHeaderRedisMessage(remainingBulkLength));
state = State.DECODE_BULK_STRING_CONTENT;
return decodeBulkStringContent(in, out);
}
}
private boolean decodeBulkStringEndOfLine(ByteBuf in, List<Object> out) throws Exception {
if (in.readableBytes() < RedisConstants.EOL_LENGTH) {
return false;
}
readEndOfLine(in);
out.add(FullBulkStringRedisMessage.EMPTY_INSTANCE);
resetDecoder();
return true;
}
private boolean decodeBulkStringContent(ByteBuf in, List<Object> out) throws Exception {
final int readableBytes = in.readableBytes();
if (readableBytes == 0 || remainingBulkLength == 0 && readableBytes < RedisConstants.EOL_LENGTH) {
return false;
}
if (readableBytes >= remainingBulkLength + RedisConstants.EOL_LENGTH) {
ByteBuf content = in.readSlice(remainingBulkLength);
readEndOfLine(in);
out.add(new DefaultLastBulkStringRedisContent(content.retain()));
resetDecoder();
return true;
}
int toRead = Math.min(remainingBulkLength, readableBytes);
remainingBulkLength -= toRead;
out.add(new DefaultBulkStringRedisContent(in.readSlice(toRead).retain()));
return true;
}
private static void readEndOfLine(final ByteBuf in) {
final short delim = in.readShort();
if (RedisConstants.EOL_SHORT == delim) {
return;
}
final byte[] bytes = RedisCodecUtil.shortToBytes(delim);
throw new RedisCodecException("delimiter: [" + bytes[0] + "," + bytes[1] + "] (expected: \\r\\n)");
}
private RedisMessage newInlineRedisMessage(RedisMessageType messageType, ByteBuf content) {
switch (messageType) {
case INLINE_COMMAND:
return new InlineCommandRedisMessage(content.toString(CharsetUtil.UTF_8));
case SIMPLE_STRING: {
SimpleStringRedisMessage cached = messagePool.getSimpleString(content);
return cached != null ? cached : new SimpleStringRedisMessage(content.toString(CharsetUtil.UTF_8));
}
case ERROR: {
ErrorRedisMessage cached = messagePool.getError(content);
return cached != null ? cached : new ErrorRedisMessage(content.toString(CharsetUtil.UTF_8));
}
case INTEGER: {
IntegerRedisMessage cached = messagePool.getInteger(content);
return cached != null ? cached : new IntegerRedisMessage(parseRedisNumber(content));
}
default:
throw new RedisCodecException("bad type: " + messageType);
}
}
private static ByteBuf readLine(ByteBuf in) {
if (!in.isReadable(RedisConstants.EOL_LENGTH)) {
return null;
}
final int lfIndex = in.forEachByte(ByteProcessor.FIND_LF);
if (lfIndex < 0) {
return null;
}
ByteBuf data = in.readSlice(lfIndex - in.readerIndex() - 1);
readEndOfLine(in);
return data;
}
private long parseRedisNumber(ByteBuf byteBuf) {
final int readableBytes = byteBuf.readableBytes();
final boolean negative = readableBytes > 0 && byteBuf.getByte(byteBuf.readerIndex()) == '-';
final int extraOneByteForNegative = negative ? 1 : 0;
if (readableBytes <= extraOneByteForNegative) {
throw new RedisCodecException("no number to parse: " + byteBuf.toString(CharsetUtil.US_ASCII));
}
if (readableBytes > RedisConstants.POSITIVE_LONG_MAX_LENGTH + extraOneByteForNegative) {
throw new RedisCodecException("too many characters to be a valid RESP Integer: " +
byteBuf.toString(CharsetUtil.US_ASCII));
}
if (negative) {
return -parsePositiveNumber(byteBuf.skipBytes(extraOneByteForNegative));
}
return parsePositiveNumber(byteBuf);
}
private long parsePositiveNumber(ByteBuf byteBuf) {
toPositiveLongProcessor.reset();
byteBuf.forEachByte(toPositiveLongProcessor);
return toPositiveLongProcessor.content();
}
private static final class ToPositiveLongProcessor implements ByteProcessor {
private long result;
@Override
public boolean process(byte value) throws Exception {
if (value < '0' || value > '9') {
throw new RedisCodecException("bad byte in number: " + value);
}
result = result * 10 + (value - '0');
return true;
}
public long content() {
return result;
}
public void reset() {
result = 0;
}
}
}