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;
  }
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.
Create a frame out of the { @link ByteBuf } and return it .