/*
* Copyright 2019 Red Hat, Inc.
* <p>
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
* and Apache License v2.0 which accompanies this distribution.
* <p>
* The Eclipse Public License is available at
* http://www.eclipse.org/legal/epl-v10.html
* <p>
* The Apache License v2.0 is available at
* http://www.opensource.org/licenses/apache2.0.php
* <p>
* You may elect to redistribute this code under either of these licenses.
*/
package io.vertx.redis.client.impl;
import io.vertx.core.Handler;
import io.vertx.core.buffer.Buffer;
import io.vertx.redis.client.Response;
import io.vertx.redis.client.impl.types.*;
import java.nio.charset.StandardCharsets;
public final class RESPParser implements Handler<Buffer> {
// 512Mb
private static final long MAX_STRING_LENGTH = 536870912;
// the callback when a full response message has been decoded
private final ParserHandler handler;
// a composite buffer to allow buffer concatenation as if it was
// a long stream
private final ReadableBuffer buffer = new ReadableBuffer();
// arrays can have nested objects so we need to keep track of the
// nesting while parsing
private final ArrayStack stack;
RESPParser(ParserHandler handler, int maxStack) {
this.handler = handler;
this.stack = new ArrayStack(maxStack);
}
// parser state machine state
private boolean eol = true;
private int bytesNeeded = 0;
@Override
public void handle(Buffer chunk) {
// add the chunk to the buffer
buffer.append(chunk);
while (buffer.readableBytes() >= (eol ? 3 : bytesNeeded + 2)) {
// setup a rollback point
buffer.mark();
// we need to locate the eol
if (eol) {
// this is the begin of a message
final byte type = buffer.readByte();
// locate the eol and handle as a C string
final int start = buffer.offset();
final int eol = buffer.findLineEnd();
// not found at all
if (eol == -1) {
buffer.reset();
break;
}
// special case for sync messages or messages that report the wrong length
if (type == '\r' && (start == eol)) {
if (!buffer.skip(1)) {
buffer.reset();
break;
}
// skipped CRLF
continue;
}
switch (type) {
case '+':
final int length = eol - start;
// special case OK
if (length == 2 && buffer.getByte(start) == 'O' && buffer.getByte(start + 1) == 'K') {
handleResponse(SimpleStringType.OK);
} else {
handleResponse(SimpleStringType.create(buffer.readLine(eol, StandardCharsets.ISO_8859_1)));
}
break;
case '-':
handleResponse(ErrorType.create(buffer.readLine(eol, StandardCharsets.ISO_8859_1)));
break;
case ':':
case '$':
case '*':
final long integer;
try {
integer = buffer.readLong(eol);
} catch (RuntimeException e) {
handler.fatal(e);
return;
}
switch (type) {
case ':':
handleResponse(IntegerType.create(integer));
break;
case '$':
// redis strings cannot be longer than 512Mb
if (integer > MAX_STRING_LENGTH) {
handler.fatal(ErrorType.create("ILLEGAL_STATE Redis Bulk cannot be larger than 512MB"));
return;
}
// special cases
if (integer < 0) {
if (integer == -1L) {
// this is a NULL string
handleResponse(null);
break;
}
// other negative values are not valid
handler.fatal(ErrorType.create("ILLEGAL_STATE Redis Bulk cannot have negative length"));
return;
}
// safe cast
bytesNeeded = (int) integer;
// in this case we switch from eol parsing to fixed len parsing
this.eol = false;
break;
case '*':
// special cases
// redis multi cannot have more than 2GB elements
if (integer > Integer.MAX_VALUE) {
handler.fatal(ErrorType.create("ILLEGAL_STATE Redis Multi cannot be larger 2GB elements"));
return;
}
if (integer < 0) {
if (integer == -1L) {
// this is a NULL array
handleResponse(null);
break;
}
// other negative values are not valid
handler.fatal(ErrorType.create("ILLEGAL_STATE Redis Multi cannot have negative length"));
}
// empty arrays can be cached and require no further processing
if (integer == 0L) {
handleResponse(MultiType.EMPTY);
} else {
// safe cast
handleResponse(MultiType.create((int) integer), true);
}
break;
}
break;
default:
// notify
handler.fatal(ErrorType.create("ILLEGAL_STATE Unknown RESP type " + (char) type));
return;
}
} else {
// empty string
if (bytesNeeded == 0) {
// special case as we don't need to allocate objects for this
handleResponse(BulkType.EMPTY);
} else {
// fixed length parsing && read the required bytes
handleResponse(BulkType.create(buffer.readBytes(bytesNeeded)));
}
// clean up the buffer, skip the last \r\n
if (buffer.skip(2)) {
// switch back to eol parsing
eol = true;
} else {
// operation failed
buffer.reset();
}
}
}
}
private void handleResponse(Response response) {
handleResponse(response, false);
}
private void handleResponse(Response response, boolean push) {
final MultiType multi = stack.peek();
// verify if there are multi's on the stack
if (multi != null) {
// add the parsed response to the multi
multi.add(response);
// push the given response to the stack
if (push) {
stack.push(response);
} else {
// break the chain and verify end condition
MultiType m = multi;
// clean up complete messages
while (m.complete()) {
stack.pop();
// in case of chaining we need to take into account
// if the stack is empty or not
if (stack.empty()) {
// handle the multi to the listener
handler.handle(m);
return;
}
// peek into the next entry
m = stack.peek();
if (m == null) {
handler.fatal(ErrorType.create("ILLEGAL_STATE Multi can't be null"));
return;
}
}
}
} else {
if (push) {
stack.push(response);
} else {
// there's nothing on the stack
// so we can handle the response directly
// to the listener
handler.handle(response);
}
}
}
}