net-commonlib/src/main/java/com/streamsets/pipeline/lib/parser/net/DelimitedLengthFieldBasedFrameDecoder.java
protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
    if (discardingTooLongFrame) {
      long bytesToDiscard = this.bytesToDiscard;
      int localBytesToDiscard = (int) Math.min(bytesToDiscard, in.readableBytes());
      in.skipBytes(localBytesToDiscard);
      bytesToDiscard -= localBytesToDiscard;
      this.bytesToDiscard = bytesToDiscard;

      failIfNecessary(false);
      return null;
    }

    if (consumingLength) {
      int delimIndex = indexOf(in, delimiter);
      if (delimIndex < 0) {
        return null;
      }

      final String lengthStr = in.toString(in.readerIndex(), delimIndex, lengthFieldCharset);
      try {
        frameLength = Long.parseLong(trimLengthString ? lengthStr.trim() : lengthStr);
      } catch (NumberFormatException e) {
        throw new CorruptedFrameException(
            String.format(
                "Invalid length field decoded (in %s charset): %s",
                lengthFieldCharset.name(),
                lengthStr
            ),
            e
        );
      }

      if (frameLength < 0) {
        throw new CorruptedFrameException("negative pre-adjustment length field: " + frameLength);
      }

      frameLength += lengthAdjustment;

      //consume length field and delimiter bytes
      in.skipBytes(delimIndex + delimiter.capacity());

      //consume delimiter bytes
      consumingLength = false;
    }

    if (frameLength > maxFrameLength) {
      long discard = frameLength - in.readableBytes();
      tooLongFrameLength = frameLength;

      if (discard < 0) {
        // buffer contains more bytes then the frameLength so we can discard all now
        in.skipBytes((int) frameLength);
      } else {
        // Enter the discard mode and discard everything received so far.
        discardingTooLongFrame = true;
        consumingLength = true;
        bytesToDiscard = discard;
        in.skipBytes(in.readableBytes());
      }
      failIfNecessary(true);
      return null;
    }

    // never overflows because it's less than maxFrameLength
    int frameLengthInt = (int) frameLength;
    if (in.readableBytes() < frameLengthInt) {
      // need more bytes available to read actual frame
      return null;
    }

    // the frame is now entirely present, reset state vars
    consumingLength = true;
    frameLength = 0;

    // extract frame
    int readerIndex = in.readerIndex();
    int actualFrameLength = frameLengthInt;// - initialBytesToStrip;
    ByteBuf frame = extractFrame(ctx, in, readerIndex, actualFrameLength);
    in.readerIndex(readerIndex + actualFrameLength);
    return frame;
  }
protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
    if (discardingTooLongFrame) {
      long bytesToDiscard = this.bytesToDiscard;
      int localBytesToDiscard = (int) Math.min(bytesToDiscard, in.readableBytes());
      in.skipBytes(localBytesToDiscard);
      bytesToDiscard -= localBytesToDiscard;
      this.bytesToDiscard = bytesToDiscard;

      failIfNecessary(false);
      return null;
    }

    if (consumingLength) {
      int delimIndex = indexOf(in, delimiter);
      if (delimIndex < 0) {
        return null;
      }

      final String lengthStr = in.toString(in.readerIndex(), delimIndex, lengthFieldCharset);
      try {
        frameLength = Long.parseLong(trimLengthString ? lengthStr.trim() : lengthStr);
      } catch (NumberFormatException e) {
        throw new CorruptedFrameException(
            String.format(
                "Invalid length field decoded (in %s charset): %s",
                lengthFieldCharset.name(),
                lengthStr
            ),
            e
        );
      }

      if (frameLength < 0) {
        throw new CorruptedFrameException("negative pre-adjustment length field: " + frameLength);
      }

      frameLength += lengthAdjustment;

      //consume length field and delimiter bytes
      in.skipBytes(delimIndex + delimiter.capacity());

      //consume delimiter bytes
      consumingLength = false;
    }

    if (frameLength > maxFrameLength) {
      long discard = frameLength - in.readableBytes();
      tooLongFrameLength = frameLength;

      if (discard < 0) {
        // buffer contains more bytes then the frameLength so we can discard all now
        in.skipBytes((int) frameLength);
      } else {
        // Enter the discard mode and discard everything received so far.
        discardingTooLongFrame = true;
        consumingLength = true;
        bytesToDiscard = discard;
        in.skipBytes(in.readableBytes());
      }
      failIfNecessary(true);
      return null;
    }

    // never overflows because it's less than maxFrameLength
    int frameLengthInt = (int) frameLength;
    if (in.readableBytes() < frameLengthInt) {
      // need more bytes available to read actual frame
      return null;
    }

    // the frame is now entirely present, reset state vars
    consumingLength = true;
    frameLength = 0;

    // extract frame
    int readerIndex = in.readerIndex();
    int actualFrameLength = frameLengthInt;// - initialBytesToStrip;
    ByteBuf frame = extractFrame(ctx, in, readerIndex, actualFrameLength);
    in.readerIndex(readerIndex + actualFrameLength);
    return frame;
  }
Create a frame out of the {@link ByteBuf} and return it.

@param ctx the {@link ChannelHandlerContext} which this {@link ByteToMessageDecoder} belongs to
@param in the {@link ByteBuf} from which to read data
@return frame           the {@link ByteBuf} which represent the frame or {@code null} if no frame could
be created.