private static DnsResponse newResponse(DatagramPacket packet, ByteBuf buf) {
    final int id = buf.readUnsignedShort();

    final int flags = buf.readUnsignedShort();
    if (flags >> 15 == 0) {
        throw new CorruptedFrameException("not a response");
    }

    final DnsResponse response = new DatagramDnsResponse(
        packet.sender(),
        packet.recipient(),
        id,
        DnsOpCode.valueOf((byte) (flags >> 11 & 0xf)), DnsResponseCode.valueOf((byte) (flags & 0xf)));

    response.setRecursionDesired((flags >> 8 & 1) == 1);
    response.setAuthoritativeAnswer((flags >> 10 & 1) == 1);
    response.setTruncated((flags >> 9 & 1) == 1);
    response.setRecursionAvailable((flags >> 7 & 1) == 1);
    response.setZ(flags >> 4 & 0x7);
    return response;
}
 

private static DnsQuery newQuery(DatagramPacket packet, ByteBuf buf) {
    final int id = buf.readUnsignedShort();

    final int flags = buf.readUnsignedShort();
    if (flags >> 15 == 1) {
        throw new CorruptedFrameException("not a query");
    }
    final DnsQuery query =
        new DatagramDnsQuery(
            packet.sender(),
            packet.recipient(),
            id,
            DnsOpCode.valueOf((byte) (flags >> 11 & 0xf)));
    query.setRecursionDesired((flags >> 8 & 1) == 1);
    query.setZ(flags >> 4 & 0x7);
    return query;
}
 

@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out)
        throws Exception {
    in.markReaderIndex();
    int preIndex = in.readerIndex();
    int length = readRawVarint32(in);
    if (preIndex == in.readerIndex()) {
        return;
    }
    if (length < 0) {
        throw new CorruptedFrameException("negative length: " + length);
    }

    if (in.readableBytes() < length) {
        in.resetReaderIndex();
    } else {
        out.add(in.readRetainedSlice(length));
    }
}
 

@Test(expected = CorruptedFrameException.class)
public void testNonJsonContent2() {
    EmbeddedChannel ch = new EmbeddedChannel(new JsonObjectDecoder());
    ch.writeInbound(Unpooled.copiedBuffer("  [1,2,3]  ", CharsetUtil.UTF_8));

    ByteBuf res = ch.readInbound();
    assertEquals("[1,2,3]", res.toString(CharsetUtil.UTF_8));
    res.release();

    try {
        ch.writeInbound(Unpooled.copiedBuffer(" a {\"key\" : 10}", CharsetUtil.UTF_8));
    } finally {
        assertFalse(ch.finish());
    }

    fail();
}
 

@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
    in.markReaderIndex();
    int preIndex = in.readerIndex();
    int length = readRawVarint32(in);
    if (length >= maxMsgLength) {
        logger.error("recv a big msg, host : {}, msg length is : {}", ctx.channel().remoteAddress(),
                length);
        in.clear();
        channel.close();
        return;
    }
    if (preIndex == in.readerIndex()) {
        return;
    }
    if (length < 0) {
        throw new CorruptedFrameException("negative length: " + length);
    }

    if (in.readableBytes() < length) {
        in.resetReaderIndex();
    } else {
        out.add(in.readRetainedSlice(length));
    }
}
 

static ObjectNode readTree(final ByteBuf in) {

        checkNotNull(in, "in");
        final JsonNode node;

        try (final InputStream istream = new ByteBufInputStream(in)) {
            node = objectMapper.readTree(istream);
        } catch (final IOException error) {
            throw new CorruptedFrameException(error);
        }

        if (node.isObject()) {
            return (ObjectNode)node;
        }

        final String cause = lenientFormat("Expected %s, not %s", JsonNodeType.OBJECT, node.getNodeType());
        throw new CorruptedFrameException(cause);
    }
 

/**
 * Decode a {@link UUID} as serialized by Microsoft APIs like {@code System.Guid.ToByteArray}
 *
 * @param in a {@link ByteBuf} containing the serialized {@link UUID} to be decoded
 * @return a new {@link UUID}
 */
public static UUID decode(final ByteBuf in) {

    checkNotNull(in, "in");

    if (in.readableBytes() < 2 * Long.BYTES) {
        final String reason = Strings.lenientFormat("invalid frame length: %s", in.readableBytes());
        throw new CorruptedFrameException(reason);
    }

    long mostSignificantBits = in.readUnsignedIntLE() << 32;

    mostSignificantBits |= (0x000000000000FFFFL & in.readShortLE()) << 16;
    mostSignificantBits |= (0x000000000000FFFFL & in.readShortLE());

    long leastSignificantBits = (0x000000000000FFFFL & in.readShortLE()) << (32 + 16);

    for (int shift = 32 + 8; shift >= 0; shift -= 8) {
        leastSignificantBits |= (0x00000000000000FFL & in.readByte()) << shift;
    }

    return new UUID(mostSignificantBits, leastSignificantBits);
}
 

@JsonProperty
public Object getValue() {

    final RntbdTokenType.Codec codec = this.header.type().codec();

    if (this.value == null) {
        return codec.defaultValue();
    }

    if (this.value instanceof ByteBuf) {
        final ByteBuf buffer = (ByteBuf) this.value;
        try {
            this.value = codec.defaultValue();
            this.value = codec.read(buffer);
        } catch (final CorruptedFrameException error) {
            String message = lenientFormat("failed to read %s value: %s", this.getName(), error.getMessage());
            throw new CorruptedFrameException(message);
        }
    } else {
        this.value = codec.convert(this.value);
    }

    return this.value;
}
 

/**
 * Encode the value in the format defined in specification as variable length
 * array.
 *
 * @throws IllegalArgumentException if the value is not in the specification bounds
 *  [0..268435455].
 */
static ByteBuf encodeRemainingLength(int value) throws CorruptedFrameException {
    if (value > MAX_LENGTH_LIMIT || value < 0) {
        throw new CorruptedFrameException("Value should in range 0.." + MAX_LENGTH_LIMIT + " found " + value);
    }

    ByteBuf encoded = Unpooled.buffer(4);
    byte digit;
    do {
        digit = (byte) (value % 128);
        value = value / 128;
        // if there are more digits to encode, set the top bit of this digit
        if (value > 0) {
            digit = (byte) (digit | 0x80);
        }
        encoded.writeByte(digit);
    } while (value > 0);
    return encoded;
}
 

@Test
public void testMaxFramePayloadLengthFailed() {
	httpServer = HttpServer.create()
	                       .port(0)
	                       .handle((in, out) -> out.sendWebsocket((i, o) -> o.sendString(Mono.just("12345678901"))))
	                       .wiretap(true)
	                       .bindNow();

	Mono<Void> response = HttpClient.create()
	                                .port(httpServer.port())
	                                .websocket(WebsocketClientSpec.builder().maxFramePayloadLength(10).build())
	                                .handle((in, out) -> in.receive()
	                                                       .asString()
	                                                       .map(srv -> srv))
	                                .log()
	                                .then();

	StepVerifier.create(response)
	            .expectError(CorruptedFrameException.class)
	            .verify(Duration.ofSeconds(30));
}
 

private DataFrame tryCreateDataFrame(ByteBuf in) {
    if (isFullDataFrame(in, in.readerIndex()))
    {
        int frameLength = peekLength(in, in.readerIndex());
        byte calculatedChecksum = calculateChecksum(in, in.readerIndex() + 1, in.readerIndex() + 1 + frameLength);
        byte frameChecksum = peekChecksum(in, in.readerIndex(), frameLength);
        if (calculatedChecksum != frameChecksum)
        {
            in.readBytes(frameLength + 2); // discard frame
            throw new CorruptedFrameException("Invalid frame checksum calc=" + ByteUtil.createString(calculatedChecksum) + " field=" + ByteUtil.createString(frameChecksum));
        }
        ByteBuf frameBuffer = in.readSlice(frameLength + 1);
        in.readByte(); // discard checksum
        return createDataFrame(frameBuffer);
    } else {
        return null;
    }
}
 

private void decode(ByteBuf in, List<Object> out) throws Exception {
    in.markReaderIndex();
    if (!Utils.checkHeaderAvailability(in)) {
        in.resetReaderIndex();
        return;
    }
    in.resetReaderIndex();

    byte messageType = Utils.readMessageType(in);

    DemuxDecoder decoder = m_decoderMap.get(messageType);
    if (decoder == null) {
        throw new CorruptedFrameException("Can't find any suitable decoder for message type: " + messageType);
    }
    decoder.decode(in, out);
}
 

protected void decode(ChannelHandlerContext ctx, ByteBuf buffer, List<Object> objects) {
	buffer.markReaderIndex();
	byte[] var4 = new byte[3];

	for (int var5 = 0; var5 < var4.length; ++var5) {
		if (!buffer.isReadable()) {
			buffer.resetReaderIndex();
			return;
		}

		var4[var5] = buffer.readByte();

		if (var4[var5] >= 0) {
			ByteBuf var6 = Unpooled.wrappedBuffer(var4);

			try {
				int var7 = PacketBuffer.readVarIntFromBuffer(var6);

				if (buffer.readableBytes() < var7) {
					buffer.resetReaderIndex();
					return;
				}

				objects.add(buffer.readBytes(var7));
			} finally {
				var6.release();
			}

			return;
		}
	}

	throw new CorruptedFrameException("length wider than 21-bit");
}
 

public static int readRawVarint32(ByteBufInputStream is) throws IOException {
  byte tmp = is.readByte();
  if (tmp >= 0) {
    return tmp;
  }
  int result = tmp & 0x7f;
  if ((tmp = is.readByte()) >= 0) {
    result |= tmp << 7;
  } else {
    result |= (tmp & 0x7f) << 7;
    if ((tmp = is.readByte()) >= 0) {
      result |= tmp << 14;
    } else {
      result |= (tmp & 0x7f) << 14;
      if ((tmp = is.readByte()) >= 0) {
        result |= tmp << 21;
      } else {
        result |= (tmp & 0x7f) << 21;
        result |= (tmp = is.readByte()) << 28;
        if (tmp < 0) {
          // Discard upper 32 bits.
          for (int i = 0; i < 5; i++) {
            if (is.readByte() >= 0) {
              return result;
            }
          }
          throw new CorruptedFrameException("Encountered a malformed varint.");
        }
      }
    }
  }
  return result;
}
 

protected void decode(ChannelHandlerContext ctx, ByteBuf buffer, List<Object> objects) {
	buffer.markReaderIndex();
	byte[] var4 = new byte[3];

	for (int var5 = 0; var5 < var4.length; ++var5) {
		if (!buffer.isReadable()) {
			buffer.resetReaderIndex();
			return;
		}

		var4[var5] = buffer.readByte();

		if (var4[var5] >= 0) {
			ByteBuf var6 = Unpooled.wrappedBuffer(var4);

			try {
				int var7 = PacketBuffer.readVarIntFromBuffer(var6);

				if (buffer.readableBytes() < var7) {
					buffer.resetReaderIndex();
					return;
				}

				objects.add(buffer.readBytes(var7));
			} finally {
				var6.release();
			}

			return;
		}
	}

	throw new CorruptedFrameException("length wider than 21-bit");
}
 

public static int readLength(ByteBuf byteBuf, int maxLength) {
    int size = byteBuf.readableBytes();
    if (size < 4) {
        return BUFFER_UNDER_RUN;
    }

    int len = byteBuf.readInt();
    if (len > maxLength) {
        throw new CorruptedFrameException("Message size exceed limit: " + len);
    }
    if (len > byteBuf.readableBytes()) {
        return BUFFER_UNDER_RUN;
    }
    return len;
}
 

public static byte[] read(ByteBuf in, int len) {
    if (len < 0) {
        throw new CorruptedFrameException("Invalid message size: " + len);
    }

    byte[] buf = new byte[len];
    in.readBytes(buf);
    return buf;
}
 

@Override
public void exceptionCaught(final ChannelHandlerContext ctx, final Throwable cause) throws Exception {

    final Channel channel = ctx.channel();

    if (cause instanceof SSLException) {
        //We can ignore SSL Exceptions, since the channel gets closed anyway.
        return;

    } else if (cause instanceof ClosedChannelException) {
        //We can ignore this because the channel is already closed
        return;

    } else if (cause instanceof IOException) {

        //We can ignore this because the channel is already closed because of an IO problem
        return;

    } else if (cause instanceof CorruptedFrameException) {

        //We can ignore this because the channel is already closed because of an IO problem
        eventLog.clientWasDisconnected(channel, "Illegal websocket data sent by client: " + cause.getMessage());
        channel.close();
        return;


    } else if (cause instanceof IllegalArgumentException) {

        //do not log IllegalArgumentException as error

    } else {
        log.error("An unexpected error occurred for client with IP {}: {}",
                ChannelUtils.getChannelIP(channel).or("UNKNOWN"), ExceptionUtils.getStackTrace(cause));
    }

    if (channel != null) {
        eventLog.clientWasDisconnected(channel, "Channel exception: " + cause.getMessage());
        channel.close();
    }
}
 

@Override
protected void decode(@Nullable ChannelHandlerContext ctx, @Nullable ByteBuf msg, @Nullable List<Object> out) throws Exception {
   Preconditions.checkNotNull(ctx);
   Preconditions.checkNotNull(msg);
   Preconditions.checkNotNull(out);

   if (msg.readableBytes() < 4) {
      return;
   }

   int mgc = msg.getUnsignedByte(msg.readerIndex());
   if (mgc != MAGIC) {
      throw new CorruptedFrameException("invalid rtsp interleaved packet: invalid magic byte: 0x" + Integer.toHexString(mgc));
   }

   int len = msg.getUnsignedShort(msg.readerIndex() + 2);
   if (msg.readableBytes() < (len + 4)) {
      return;
   }

   int ch = msg.getUnsignedByte(msg.readerIndex() + 1);
   if (ch != channel) {
      channel = ch;
      ctx.fireUserEventTriggered(new RtspChannel(channel));
   }

   msg.skipBytes(4);
   out.add(msg.readSlice(len).retain());
}
 

@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
    // Wait until the length prefix is available.
    if (in.readableBytes() < 5) {
        return;
    }

    in.markReaderIndex();

    // Check the magic number.
    int magicNumber = in.readUnsignedByte();
    if (magicNumber != 'F') {
        in.resetReaderIndex();
        throw new CorruptedFrameException("Invalid magic number: " + magicNumber);
    }

    // Wait until the whole data is available.
    int dataLength = in.readInt();
    if (in.readableBytes() < dataLength) {
        in.resetReaderIndex();
        return;
    }

    // Convert the received data into a new BigInteger.
    byte[] decoded = new byte[dataLength];
    in.readBytes(decoded);

    out.add(new BigInteger(decoded));
}
 

@Test
public void readMalformedResponseTest() throws Exception {
    EmbeddedChannel embedder = new EmbeddedChannel(new DatagramDnsResponseDecoder());
    ByteBuf packet = embedder.alloc().buffer(512).writeBytes(malformedLoopPacket);
    exception.expect(CorruptedFrameException.class);
    embedder.writeInbound(new DatagramPacket(packet, null, new InetSocketAddress(0)));
}
 

public void finish() {
    checking = false;
    codep = 0;
    if (state != UTF8_ACCEPT) {
        state = UTF8_ACCEPT;
        throw new CorruptedFrameException("bytes are not UTF-8");
    }
}
 

@Override
public boolean process(byte b) throws Exception {
    byte type = TYPES[b & 0xFF];

    codep = state != UTF8_ACCEPT ? b & 0x3f | codep << 6 : 0xff >> type & b;

    state = STATES[state + type];

    if (state == UTF8_REJECT) {
        checking = false;
        throw new CorruptedFrameException("bytes are not UTF-8");
    }
    return true;
}
 

private void checkUTF8String(ChannelHandlerContext ctx, ByteBuf buffer) {
    try {
        if (utf8Validator == null) {
            utf8Validator = new Utf8Validator();
        }
        utf8Validator.check(buffer);
    } catch (CorruptedFrameException ex) {
        if (ctx.channel().isActive()) {
            ctx.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
        }
    }
}
 

private void protocolViolation(ChannelHandlerContext ctx, CorruptedFrameException ex) {
    state = State.CORRUPT;
    if (ctx.channel().isActive()) {
        Object closeMessage;
        if (receivedClosingHandshake) {
            closeMessage = Unpooled.EMPTY_BUFFER;
        } else {
            closeMessage = new CloseWebSocketFrame(1002, null);
        }
        ctx.writeAndFlush(closeMessage).addListener(ChannelFutureListener.CLOSE);
    }
    throw ex;
}
 

/** */
protected void checkCloseFrameBody(
        ChannelHandlerContext ctx, ByteBuf buffer) {
    if (buffer == null || !buffer.isReadable()) {
        return;
    }
    if (buffer.readableBytes() == 1) {
        protocolViolation(ctx, "Invalid close frame body");
    }

    // Save reader index
    int idx = buffer.readerIndex();
    buffer.readerIndex(0);

    // Must have 2 byte integer within the valid range
    int statusCode = buffer.readShort();
    if (statusCode >= 0 && statusCode <= 999 || statusCode >= 1004 && statusCode <= 1006
            || statusCode >= 1012 && statusCode <= 2999) {
        protocolViolation(ctx, "Invalid close frame getStatus code: " + statusCode);
    }

    // May have UTF-8 message
    if (buffer.isReadable()) {
        try {
            new Utf8Validator().check(buffer);
        } catch (CorruptedFrameException ex) {
            protocolViolation(ctx, ex);
        }
    }

    // Restore reader index
    buffer.readerIndex(idx);
}
 

static String decodeDomainName(ByteBuf in) {
    in.markReaderIndex();
    try {
        return DefaultDnsRecordDecoder.decodeName(in);
    } catch (CorruptedFrameException e) {
        // In this case we just return null.
        return null;
    } finally {
        in.resetReaderIndex();
    }
}
 

@Test(expected = CorruptedFrameException.class)
public void testNonJsonContent1() {
    EmbeddedChannel ch = new EmbeddedChannel(new JsonObjectDecoder());
    try {
        ch.writeInbound(Unpooled.copiedBuffer("  b [1,2,3]", CharsetUtil.UTF_8));
    } finally {
        assertFalse(ch.finish());
    }

    fail();
}
 

public static int toInt(ByteBuf in) {
    if (in.readableBytes() < 4) {
        throw new CorruptedFrameException("to int.");
    }
    return ((in.readByte() & 0xff) << 24) | ((in.readByte() & 0xff) << 16) | ((in.readByte() & 0xff) << 8)
            | ((in.readByte() & 0xff) << 0);
}
 

public static long toLong(ByteBuf in) {
    if (in.readableBytes() < 8) {
        throw new CorruptedFrameException("to long.");
    }
    return ((in.readByte() & 0xff) << 56) | ((in.readByte() & 0xff) << 48) | ((in.readByte() & 0xff) << 40)
            | ((in.readByte() & 0xff) << 32) | ((in.readByte() & 0xff) << 24) | ((in.readByte() & 0xff) << 16)
            | ((in.readByte() & 0xff) << 8) | ((in.readByte() & 0xff) << 0);
}