package jdk.incubator.http.internal.websocket;
import jdk.incubator.http.WebSocket.MessagePart;
import jdk.incubator.http.internal.common.Log;
import jdk.incubator.http.internal.websocket.Frame.Opcode;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.charset.CharacterCodingException;
import static java.lang.String.format;
import static java.nio.charset.StandardCharsets.UTF_8;
import static java.util.Objects.requireNonNull;
import static jdk.incubator.http.internal.common.Utils.dump;
import static jdk.incubator.http.internal.websocket.StatusCodes.NO_STATUS_CODE;
import static jdk.incubator.http.internal.websocket.StatusCodes.isLegalToReceiveFromServer;
class FrameConsumer implements Frame.Consumer {
private final MessageStreamConsumer output;
private final UTF8AccumulatingDecoder decoder = new UTF8AccumulatingDecoder();
private boolean fin;
private Opcode opcode, originatingOpcode;
private MessagePart part = MessagePart.WHOLE;
private long payloadLen;
private long unconsumedPayloadLen;
private ByteBuffer binaryData;
FrameConsumer(MessageStreamConsumer output) {
this.output = requireNonNull(output);
}
MessageStreamConsumer getOutput() {
return output;
}
@Override
public void fin(boolean value) {
Log.logTrace("Reading fin: {0}", value);
fin = value;
}
@Override
public void rsv1(boolean value) {
Log.logTrace("Reading rsv1: {0}", value);
if (value) {
throw new FailWebSocketException("Unexpected rsv1 bit");
}
}
@Override
public void rsv2(boolean value) {
Log.logTrace("Reading rsv2: {0}", value);
if (value) {
throw new FailWebSocketException("Unexpected rsv2 bit");
}
}
@Override
public void rsv3(boolean value) {
Log.logTrace("Reading rsv3: {0}", value);
if (value) {
throw new FailWebSocketException("Unexpected rsv3 bit");
}
}
@Override
public void opcode(Opcode v) {
Log.logTrace("Reading opcode: {0}", v);
if (v == Opcode.PING || v == Opcode.PONG || v == Opcode.CLOSE) {
if (!fin) {
throw new FailWebSocketException("Fragmented control frame " + v);
}
opcode = v;
} else if (v == Opcode.TEXT || v == Opcode.BINARY) {
if (originatingOpcode != null) {
throw new FailWebSocketException(
format("Unexpected frame %s (fin=%s)", v, fin));
}
opcode = v;
if (!fin) {
originatingOpcode = v;
}
} else if (v == Opcode.CONTINUATION) {
if (originatingOpcode == null) {
throw new FailWebSocketException(
format("Unexpected frame %s (fin=%s)", v, fin));
}
opcode = v;
} else {
throw new FailWebSocketException("Unknown opcode " + v);
}
}
@Override
public void mask(boolean value) {
Log.logTrace("Reading mask: {0}", value);
if (value) {
throw new FailWebSocketException("Masked frame received");
}
}
@Override
public void payloadLen(long value) {
Log.logTrace("Reading payloadLen: {0}", value);
if (opcode.isControl()) {
if (value > 125) {
throw new FailWebSocketException(
format("%s's payload length %s", opcode, value));
}
assert Opcode.CLOSE.isControl();
if (opcode == Opcode.CLOSE && value == 1) {
throw new FailWebSocketException("Incomplete status code");
}
}
payloadLen = value;
unconsumedPayloadLen = value;
}
@Override
public void maskingKey(int value) {
throw new InternalError();
}
@Override
public void payloadData(ByteBuffer data) {
Log.logTrace("Reading payloadData: data={0}", data);
unconsumedPayloadLen -= data.remaining();
boolean isLast = unconsumedPayloadLen == 0;
if (opcode.isControl()) {
if (binaryData != null) {
binaryData.put(data);
} else if (!isLast) {
int remaining = data.remaining();
assert remaining < 125 : dump(remaining);
binaryData = ByteBuffer.allocate(125).put(data);
} else {
binaryData = ByteBuffer.allocate(data.remaining()).put(data);
}
} else {
part = determinePart(isLast);
boolean text = opcode == Opcode.TEXT || originatingOpcode == Opcode.TEXT;
if (!text) {
output.onBinary(data.slice(), part);
data.position(data.limit());
} else {
boolean binaryNonEmpty = data.hasRemaining();
CharBuffer textData;
try {
textData = decoder.decode(data, part == MessagePart.WHOLE || part == MessagePart.LAST);
} catch (CharacterCodingException e) {
throw new FailWebSocketException(
"Invalid UTF-8 in frame " + opcode, StatusCodes.NOT_CONSISTENT)
.initCause(e);
}
if (!(binaryNonEmpty && !textData.hasRemaining())) {
output.onText(textData, part);
}
}
}
}
@Override
public void endFrame() {
if (opcode.isControl()) {
binaryData.flip();
}
switch (opcode) {
case CLOSE:
char statusCode = NO_STATUS_CODE;
String reason = "";
if (payloadLen != 0) {
int len = binaryData.remaining();
assert 2 <= len && len <= 125 : dump(len, payloadLen);
statusCode = binaryData.getChar();
if (!isLegalToReceiveFromServer(statusCode)) {
throw new FailWebSocketException(
"Illegal status code: " + statusCode);
}
try {
reason = UTF_8.newDecoder().decode(binaryData).toString();
} catch (CharacterCodingException e) {
throw new FailWebSocketException("Illegal close reason")
.initCause(e);
}
}
output.onClose(statusCode, reason);
break;
case PING:
output.onPing(binaryData);
binaryData = null;
break;
case PONG:
output.onPong(binaryData);
binaryData = null;
break;
default:
assert opcode == Opcode.TEXT || opcode == Opcode.BINARY
|| opcode == Opcode.CONTINUATION : dump(opcode);
if (fin) {
originatingOpcode = null;
}
break;
}
payloadLen = 0;
opcode = null;
}
private MessagePart determinePart(boolean isLast) {
boolean lastChunk = fin && isLast;
switch (part) {
case LAST:
case WHOLE:
return lastChunk ? MessagePart.WHOLE : MessagePart.FIRST;
case FIRST:
case PART:
return lastChunk ? MessagePart.LAST : MessagePart.PART;
default:
throw new InternalError(String.valueOf(part));
}
}
}