/*
    * Copyright 2015 The Netty Project
    *
    * The Netty Project licenses this file to you under the Apache License,
    * version 2.0 (the "License"); you may not use this file except in compliance
    * with the License. You may obtain a copy of the License at:
    *
    *   https://www.apache.org/licenses/LICENSE-2.0
    *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
   * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
   * License for the specific language governing permissions and limitations
   * under the License.
   */
  package io.netty.handler.codec.protobuf;
  
  import com.google.protobuf.CodedInputStream;
  import com.google.protobuf.nano.CodedInputByteBufferNano;
  import io.netty.buffer.ByteBuf;
  import io.netty.channel.ChannelHandlerContext;
  import io.netty.handler.codec.ByteToMessageDecoder;
  import io.netty.handler.codec.CorruptedFrameException;
  import io.netty.handler.codec.TooLongFrameException;
  
  import java.util.List;
  
  import static io.netty.util.internal.ObjectUtil.checkPositive;
  
  /**
   * A decoder that splits the received {@link ByteBuf}s dynamically by the
   * value of the Google Protocol Buffers
   * <a href="https://developers.google.com/protocol-buffers/docs/encoding#varints">Base
   * 128 Varints</a> integer length field in the message. For example:
   * <pre>
   * BEFORE DECODE (302 bytes)       AFTER DECODE (300 bytes)
   * +--------+---------------+      +---------------+
   * | Length | Protobuf Data |----->| Protobuf Data |
   * | 0xAC02 |  (300 bytes)  |      |  (300 bytes)  |
   * +--------+---------------+      +---------------+
   * </pre>
   *
   * @see CodedInputStream
   * @see CodedInputByteBufferNano
   */
  public class ProtobufVarint32FrameDecoder extends ByteToMessageDecoder {
  
      private final int maxFrameLength;
      private long bytesToDiscard;
  
      /**
       * Creates a new instance with no frame length limit.
       */
      public ProtobufVarint32FrameDecoder() {
          this(Integer.MAX_VALUE);
      }
  
      /**
       * Creates a new instance with the specified maximum frame length.
       *
       * @param maxFrameLength the maximum length of the frame.
       *                       If the length exceeds this value,
       *                       {@link TooLongFrameException} will be thrown.
       */
      public ProtobufVarint32FrameDecoder(int maxFrameLength) {
          this.maxFrameLength = checkPositive(maxFrameLength, "maxFrameLength");
      }
  
      @Override
      protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out)
              throws Exception {
          if (bytesToDiscard > 0) {
              int localBytesToDiscard = (int) Math.min(bytesToDiscard, in.readableBytes());
              in.skipBytes(localBytesToDiscard);
              bytesToDiscard -= localBytesToDiscard;
              return;
          }
  
          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 (length > maxFrameLength) {
              long discard = length - in.readableBytes();
              if (discard <= 0) {
                  in.skipBytes(length);
              } else {
                  bytesToDiscard = discard;
                  in.skipBytes(in.readableBytes());
              }
              throw new TooLongFrameException(
                      "Frame length exceeds " + maxFrameLength
                      + ": " + length);
         }
 
         if (in.readableBytes() < length) {
             in.resetReaderIndex();
         } else {
             out.add(in.readRetainedSlice(length));
         }
     }
 
     /**
      * Reads variable length 32bit int from buffer
      *
      * @return decoded int if buffers readerIndex has been forwarded else nonsense value
      */
     static int readRawVarint32(ByteBuf buffer) {
         if (buffer.readableBytes() < 4) {
             return readRawVarint24(buffer);
         }
         int wholeOrMore = buffer.getIntLE(buffer.readerIndex());
         int firstOneOnStop = ~wholeOrMore & 0x80808080;
         if (firstOneOnStop == 0) {
             return readRawVarint40(buffer, wholeOrMore);
         }
         int bitsToKeep = Integer.numberOfTrailingZeros(firstOneOnStop) + 1;
         buffer.skipBytes(bitsToKeep >> 3);
         int thisVarintMask = firstOneOnStop ^ (firstOneOnStop - 1);
         int wholeWithContinuations = wholeOrMore & thisVarintMask;
         // mix them up as per varint spec while dropping the continuation bits:
         // 0x7F007F isolate the first byte and the third byte dropping the continuation bits
         // 0x7F007F00 isolate the second byte and the fourth byte dropping the continuation bits
         // the second and fourth byte are shifted to the right by 1, filling the gaps left by the first and third byte
         // it means that the first and second bytes now occupy the first 14 bits (7 bits each)
         // and the third and fourth bytes occupy the next 14 bits (7 bits each), with a gap between the 2s of 2 bytes
         // and another gap of 2 bytes after the forth and third.
         wholeWithContinuations = (wholeWithContinuations & 0x7F007F) | ((wholeWithContinuations & 0x7F007F00) >> 1);
         // 0x3FFF isolate the first 14 bits i.e. the first and second bytes
         // 0x3FFF0000 isolate the next 14 bits i.e. the third and forth bytes
         // the third and forth bytes are shifted to the right by 2, filling the gaps left by the first and second bytes
         return (wholeWithContinuations & 0x3FFF) | ((wholeWithContinuations & 0x3FFF0000) >> 2);
     }
 
     private static int readRawVarint40(ByteBuf buffer, int wholeOrMore) {
         if (buffer.readableBytes() == 4) {
             // The first 4 bytes all have the continuation bit set, but the 5th byte did not arrive yet.
             // Leave the readerIndex untouched so the caller waits for more data.
             return 0;
         }
         byte lastByte = buffer.getByte(buffer.readerIndex() + 4);
         if (lastByte < 0) {
             throw new CorruptedFrameException("malformed varint.");
         }
         buffer.skipBytes(5);
         // add it to wholeOrMore
         return wholeOrMore & 0x7F |
                (((wholeOrMore >> 8) & 0x7F) << 7) |
                (((wholeOrMore >> 16) & 0x7F) << 14) |
                (((wholeOrMore >> 24) & 0x7F) << 21) |
                (lastByte << 28);
     }
 
     private static int readRawVarint24(ByteBuf buffer) {
         if (!buffer.isReadable()) {
             return 0;
         }
         buffer.markReaderIndex();
 
         byte tmp = buffer.readByte();
         if (tmp >= 0) {
             return tmp;
         }
         int result = tmp & 127;
         if (!buffer.isReadable()) {
             buffer.resetReaderIndex();
             return 0;
         }
         if ((tmp = buffer.readByte()) >= 0) {
             return result | tmp << 7;
         }
         result |= (tmp & 127) << 7;
         if (!buffer.isReadable()) {
             buffer.resetReaderIndex();
             return 0;
         }
         if ((tmp = buffer.readByte()) >= 0) {
             return result | tmp << 14;
         }
         // All 3 readable bytes have the continuation bit set, which means at least one more byte is needed to
         // complete the varint. We only end up here if less than 4 bytes were readable, so wait for more data.
         buffer.resetReaderIndex();
         return 0;
     }
 }