content: /*
   * Licensed to the Apache Software Foundation (ASF) under one
   * or more contributor license agreements.  See the NOTICE file
   * distributed with this work for additional information
   * regarding copyright ownership.  The ASF 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
   *
   *     http://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 org.apache.cassandra.utils.bytecomparable;
  
  import java.nio.ByteBuffer;
  import java.nio.charset.StandardCharsets;
  
  import org.apache.cassandra.db.marshal.ValueAccessor;
  import org.apache.cassandra.utils.bytecomparable.ByteComparable.Version;
  import org.apache.cassandra.utils.memory.MemoryUtil;
  
  import static com.google.common.base.Preconditions.checkArgument;
  
  /**
   * A stream of bytes, used for byte-order-comparable representations of data, and utilities to convert various values
   * to their byte-ordered translation.
   * See ByteComparable.md for details about the encoding scheme.
   */
  public interface ByteSource
  {
      /** Consume the next byte, unsigned. Must be between 0 and 255, or END_OF_STREAM if there are no more bytes. */
      int next();
  
      /** Value returned if at the end of the stream. */
      int END_OF_STREAM = -1;
  
      ByteSource EMPTY = () -> END_OF_STREAM;
  
      /**
       * Escape value. Used, among other things, to mark the end of subcomponents (so that shorter compares before anything longer).
       * Actual zeros in input need to be escaped if this is in use (see {@link AbstractEscaper}).
       */
      int ESCAPE = 0x00;
  
      // Zeros are encoded as a sequence of ESCAPE, 0 or more of ESCAPED_0_CONT, ESCAPED_0_DONE so zeroed spaces only grow by 1 byte
      int ESCAPED_0_CONT = 0xFE;
      int ESCAPED_0_DONE = 0xFF;
  
      // All separators must be within these bounds
      int MIN_SEPARATOR = 0x10;
      int MAX_SEPARATOR = 0xEF;
  
      // Next component marker.
      int NEXT_COMPONENT = 0x40;
      // Marker used to present null values represented by empty buffers (e.g. by Int32Type)
      int NEXT_COMPONENT_EMPTY = 0x3F;
      int NEXT_COMPONENT_EMPTY_REVERSED = 0x41;
      // Marker for null components in tuples, maps, sets and clustering keys.
      int NEXT_COMPONENT_NULL = 0x3E;
  
      // Section for next component markers which is not allowed for use
      int MIN_NEXT_COMPONENT = 0x3C;
      int MAX_NEXT_COMPONENT = 0x44;
  
      // Default terminator byte in sequences. Smaller than NEXT_COMPONENT_NULL, but larger than LT_NEXT_COMPONENT to
      // ensure lexicographic compares go in the correct direction
      int TERMINATOR = 0x38;
      // These are special endings, for exclusive/inclusive bounds (i.e. smaller than anything with more components,
      // bigger than anything with more components)
      int LT_NEXT_COMPONENT = 0x20;
      int GT_NEXT_COMPONENT = 0x60;
  
      // Unsupported, for artificial bounds
      int LTLT_NEXT_COMPONENT = 0x1F; // LT_NEXT_COMPONENT - 1
      int GTGT_NEXT_COMPONENT = 0x61; // GT_NEXT_COMPONENT + 1
  
      // Special value for components that should be excluded from the normal min/max span. (static rows)
      int EXCLUDED = 0x18;
  
      /**
       * Encodes byte-accessible data as a byte-comparable source that has 0s escaped and finishes in an escaped
       * state.
       * This provides a weakly-prefix-free byte-comparable version of the content to use in sequences.
       * (See {@link AbstractEscaper} for a detailed explanation.)
       */
      static <V> ByteSource of(ValueAccessor<V> accessor, V data, Version version)
      {
          return new AccessorEscaper<>(accessor, data, version);
      }
  
      /**
       * Encodes a byte buffer as a byte-comparable source that has 0s escaped and finishes in an escape.
       * This provides a weakly-prefix-free byte-comparable version of the content to use in sequences.
       * (See ByteSource.BufferEscaper/Multi for explanation.)
       */
      static ByteSource of(ByteBuffer buf, Version version)
      {
          return new BufferEscaper(buf, version);
      }
  
      /**
       * Encodes a byte array as a byte-comparable source that has 0s escaped and finishes in an escape.
       * This provides a prefix-free byte-comparable version of the content to use in sequences.
       * (See ByteSource.BufferEscaper/Multi for explanation.)
       */
      static ByteSource of(byte[] buf, Version version)
      {
          return new ArrayEscaper(buf, version);
      }
  
      /**
       * Encodes a memory range as a byte-comparable source that has 0s escaped and finishes in an escape.
       * This provides a weakly-prefix-free byte-comparable version of the content to use in sequences.
       * (See ByteSource.BufferEscaper/Multi for explanation.)
       */
      static ByteSource ofMemory(long address, int length, ByteComparable.Version version)
      {
          return new MemoryEscaper(address, length, version);
      }
  
      /**
       * Combines a chain of sources, turning their weak-prefix-free byte-comparable representation into the combination's
       * prefix-free byte-comparable representation, with the included terminator character.
       * For correctness, the terminator must be within MIN-MAX_SEPARATOR and outside the range reserved for
       * NEXT_COMPONENT markers.
       * Typically TERMINATOR, or LT/GT_NEXT_COMPONENT if used for partially specified bounds.
       */
      static ByteSource withTerminator(int terminator, ByteSource... srcs)
      {
          assert terminator >= MIN_SEPARATOR && terminator <= MAX_SEPARATOR;
          assert terminator < MIN_NEXT_COMPONENT || terminator > MAX_NEXT_COMPONENT;
          return new Multi(srcs, terminator);
      }
  
      /**
       * As above, but permits any separator. The legacy format wasn't using weak prefix freedom and has some
       * non-reversible transformations.
       */
      static ByteSource withTerminatorLegacy(int terminator, ByteSource... srcs)
      {
          return new Multi(srcs, terminator);
      }
  
      static ByteSource withTerminatorMaybeLegacy(Version version, int legacyTerminator, ByteSource... srcs)
      {
          return version == Version.LEGACY ? withTerminatorLegacy(legacyTerminator, srcs)
                                           : withTerminator(TERMINATOR, srcs);
      }
  
      static ByteSource of(String s, Version version)
      {
          return new ArrayEscaper(s.getBytes(StandardCharsets.UTF_8), version);
      }
  
      static ByteSource of(long value)
      {
          return new Number(value ^ (1L<<63), 8);
      }
  
      static ByteSource of(int value)
      {
          return new Number(value ^ (1L<<31), 4);
      }
  
      /**
       * Produce a source for a signed fixed-length number, also translating empty to null.
       * The first byte has its sign bit inverted, and the rest are passed unchanged.
       * Presumes that the length of the buffer is always either 0 or constant for the type, which permits decoding and
       * ensures the representation is prefix-free.
       */
      static <V> ByteSource optionalSignedFixedLengthNumber(ValueAccessor<V> accessor, V data)
      {
          return !accessor.isEmpty(data) ? signedFixedLengthNumber(accessor, data) : null;
      }
  
      /**
       * Produce a source for a signed fixed-length number.
       * The first byte has its sign bit inverted, and the rest are passed unchanged.
       * Presumes that the length of the buffer is always constant for the type.
       */
      static <V> ByteSource signedFixedLengthNumber(ValueAccessor<V> accessor, V data)
      {
          return new SignedFixedLengthNumber<>(accessor, data);
      }
  
      /**
       * Produce a source for a signed fixed-length floating-point number, also translating empty to null.
       * If sign bit is on, returns negated bytes. If not, add the sign bit value.
       * (Sign of IEEE floats is the highest bit, the rest can be compared in magnitude by byte comparison.)
       * Presumes that the length of the buffer is always either 0 or constant for the type, which permits decoding and
       * ensures the representation is prefix-free.
       */
      static <V> ByteSource optionalSignedFixedLengthFloat(ValueAccessor<V> accessor, V data)
      {
          return !accessor.isEmpty(data) ? signedFixedLengthFloat(accessor, data) : null;
      }
  
      /**
       * Produce a source for a signed fixed-length floating-point number.
       * If sign bit is on, returns negated bytes. If not, add the sign bit value.
       * (Sign of IEEE floats is the highest bit, the rest can be compared in magnitude by byte comparison.)
       * Presumes that the length of the buffer is always constant for the type.
       */
      static <V> ByteSource signedFixedLengthFloat(ValueAccessor<V> accessor, V data)
      {
          return new SignedFixedLengthFloat<>(accessor, data);
      }
  
      /**
       * Produce a source for a signed integer, stored using variable length encoding.
       * The representation uses between 1 and 9 bytes, is prefix-free and compares
       * correctly.
       */
      static ByteSource variableLengthInteger(long value)
      {
          return new VariableLengthInteger(value);
      }
  
      /**
       * Returns a separator for two byte sources, i.e. something that is definitely > prevMax, and <= currMin, assuming
       * prevMax < currMin.
       * This returns the shortest prefix of currMin that is greater than prevMax.
       */
      public static ByteSource separatorPrefix(ByteSource prevMax, ByteSource currMin)
      {
          return new Separator(prevMax, currMin, true);
      }
  
      /**
       * Returns a separator for two byte sources, i.e. something that is definitely > prevMax, and <= currMin, assuming
       * prevMax < currMin.
       * This is a source of length 1 longer than the common prefix of the two sources, with last byte one higher than the
       * prevMax source.
       */
      public static ByteSource separatorGt(ByteSource prevMax, ByteSource currMin)
      {
          return new Separator(prevMax, currMin, false);
      }
  
      public static ByteSource oneByte(int i)
      {
          assert i >= 0 && i <= 0xFF : "Argument must be a valid unsigned byte.";
          return new ByteSource()
          {
              boolean consumed = false;
  
              @Override
              public int next()
              {
                  if (consumed)
                      return END_OF_STREAM;
                  consumed = true;
                  return i;
              }
          };
      }
  
      public static ByteSource cut(ByteSource src, int cutoff)
      {
          return new ByteSource()
          {
              int pos = 0;
  
              @Override
              public int next()
              {
                  return pos++ < cutoff ? src.next() : END_OF_STREAM;
              }
          };
      }
  
      /**
       * Wrap a ByteSource in a length-fixing facade.
       *
       * If the length of {@code src} is less than {@code cutoff}, then pad it on the right with {@code padding} until
       * the overall length equals {@code cutoff}.  If the length of {@code src} is greater than {@code cutoff}, then
       * truncate {@code src} to that size.  Effectively a noop if {@code src} happens to have length {@code cutoff}.
       *
       * @param src the input source to wrap
       * @param cutoff the size of the source returned
       * @param padding a padding byte (an int subject to a 0xFF mask)
       */
      public static ByteSource cutOrRightPad(ByteSource src, int cutoff, int padding)
      {
          return new ByteSource()
          {
              int pos = 0;
  
              @Override
              public int next()
              {
                  if (pos++ >= cutoff)
                  {
                      return END_OF_STREAM;
                  }
                  int next = src.next();
                  return next == END_OF_STREAM ? padding : next;
              }
          };
      }
  
  
      /**
       * Variable-length encoding. Escapes 0s as ESCAPE + zero or more ESCAPED_0_CONT + ESCAPED_0_DONE.
       * If the source ends in 0, we use ESCAPED_0_CONT to make sure that the encoding remains smaller than that source
       * with a further 0 at the end.
       * Finishes in an escaped state (either with ESCAPE or ESCAPED_0_CONT), which in {@link Multi} is followed by
       * a component separator between 0x10 and 0xFE.
       *
       * E.g. "A\0\0B" translates to 4100FEFF4200
       *      "A\0B\0"               4100FF4200FE (+00 for {@link Version#LEGACY})
       *      "A\0"                  4100FE       (+00 for {@link Version#LEGACY})
       *      "AB"                   414200
       *
       * If in a single byte source, the bytes could be simply passed unchanged, but this would not allow us to
       * combine components. This translation preserves order, and since the encoding for 0 is higher than the separator
       * also makes sure shorter components are treated as smaller.
       *
       * The encoding is not prefix-free, since e.g. the encoding of "A" (4100) is a prefix of the encoding of "A\0"
       * (4100FE), but the byte following the prefix is guaranteed to be FE or FF, which makes the encoding weakly
       * prefix-free. Additionally, any such prefix sequence will compare smaller than the value to which it is a prefix,
       * because any permitted separator byte will be smaller than the byte following the prefix.
       */
      abstract static class AbstractEscaper implements ByteSource
      {
          private final Version version;
          private int bufpos;
          private boolean escaped;
  
          AbstractEscaper(int position, Version version)
          {
              this.bufpos = position;
              this.version = version;
          }
  
          @Override
          public final int next()
          {
              if (bufpos >= limit())
              {
                  if (bufpos > limit())
                      return END_OF_STREAM;
  
                  ++bufpos;
                  if (escaped)
                  {
                      escaped = false;
                      if (version == Version.LEGACY)
                          --bufpos; // place an ESCAPE at the end of sequence ending in ESCAPE
                      return ESCAPED_0_CONT;
                  }
                  return ESCAPE;
              }
  
              int index = bufpos++;
              int b = get(index) & 0xFF;
              if (!escaped)
              {
                  if (b == ESCAPE)
                      escaped = true;
                  return b;
              }
              else
              {
                  if (b == ESCAPE)
                      return ESCAPED_0_CONT;
                  --bufpos;
                  escaped = false;
                  return ESCAPED_0_DONE;
              }
          }
  
          protected abstract byte get(int index);
  
          protected abstract int limit();
      }
  
      static class AccessorEscaper<V> extends AbstractEscaper
      {
          private final V data;
          private final ValueAccessor<V> accessor;
  
          private AccessorEscaper(ValueAccessor<V> accessor, V data, Version version)
          {
              super(0, version);
              this.accessor = accessor;
              this.data = data;
          }
  
          protected int limit()
          {
              return accessor.size(data);
          }
  
          protected byte get(int index)
          {
              return accessor.getByte(data, index);
          }
      }
  
      static class BufferEscaper extends AbstractEscaper
      {
          private final ByteBuffer buf;
  
          private BufferEscaper(ByteBuffer buf, Version version)
          {
              super(buf.position(), version);
              this.buf = buf;
          }
  
          protected int limit()
          {
              return buf.limit();
          }
  
          protected byte get(int index)
          {
              return buf.get(index);
          }
      }
  
      static class ArrayEscaper extends AbstractEscaper
      {
          private final byte[] buf;
  
          private ArrayEscaper(byte[] buf, Version version)
          {
              super(0, version);
              this.buf = buf;
          }
  
          @Override
          protected byte get(int index)
          {
              return buf[index];
          }
  
          @Override
          protected int limit()
          {
              return buf.length;
          }
      }
  
      static class MemoryEscaper extends AbstractEscaper
      {
          private final long address;
          private final int length;
  
          MemoryEscaper(long address, int length, ByteComparable.Version version)
          {
              super(0, version);
              this.address = address;
              this.length = length;
          }
  
          protected byte get(int index)
          {
              return MemoryUtil.getByte(address + index);
          }
  
          protected int limit()
          {
              return length;
          }
      }
  
      /**
       * Fixed length signed number encoding. Inverts first bit (so that neg < pos), then just posts all bytes from the
       * buffer. Assumes buffer is of correct length.
       */
      static class SignedFixedLengthNumber<V> implements ByteSource
      {
          private final ValueAccessor<V> accessor;
          private final V data;
          private int bufpos;
  
          public SignedFixedLengthNumber(ValueAccessor<V> accessor, V data)
          {
              this.accessor = accessor;
              this.data = data;
              this.bufpos = 0;
          }
  
          @Override
          public int next()
          {
              if (bufpos >= accessor.size(data))
                  return END_OF_STREAM;
              int v = accessor.getByte(data, bufpos) & 0xFF;
              if (bufpos == 0)
                  v ^= 0x80;
              ++bufpos;
              return v;
          }
      }
  
      /**
       * Variable-length encoding for unsigned integers.
       * The encoding is similar to UTF-8 encoding.
       * Numbers between 0 and 127 are encoded in one byte, using 0 in the most significant bit.
       * Larger values have 1s in as many of the most significant bits as the number of additional bytes
       * in the representation, followed by a 0. This ensures that longer numbers compare larger than shorter
       * ones. Since we never use a longer representation than necessary, this implies numbers compare correctly.
       * As the number of bytes is specified in the bits of the first, no value is a prefix of another.
       */
      static class VariableLengthUnsignedInteger implements ByteSource
      {
          private final long value;
          private int pos = -1;
  
          public VariableLengthUnsignedInteger(long value)
          {
              this.value = value;
          }
  
          @Override
          public int next()
          {
              if (pos == -1)
              {
                  int bitsMinusOne = 63 - (Long.numberOfLeadingZeros(value | 1)); // 0 to 63 (the | 1 is to make sure 0 maps to 0 (1 bit))
                  int bytesMinusOne = bitsMinusOne / 7;
                  int mask = -256 >> bytesMinusOne;   // sequence of bytesMinusOne 1s in the most-significant bits
                  pos = bytesMinusOne * 8;
                  return (int) ((value >>> pos) | mask) & 0xFF;
              }
              pos -= 8;
              if (pos < 0)
                  return END_OF_STREAM;
              return (int) (value >>> pos) & 0xFF;
          }
      }
  
      /**
       * Variable-length encoding for signed integers.
       * The encoding is based on the unsigned encoding above, where the first bit stored is the inverted sign,
       * followed by as many matching bits as there are additional bytes in the encoding, followed by the two's
       * complement of the number.
       * Because of the inverted sign bit, negative numbers compare smaller than positives, and because the length
       * bits match the sign, longer positive numbers compare greater and longer negative ones compare smaller.
       *
       * Examples:
       *      0              encodes as           80
       *      1              encodes as           81
       *     -1              encodes as           7F
       *     63              encodes as           BF
       *     64              encodes as           C040
       *    -64              encodes as           40
       *    -65              encodes as           3FBF
       *   2^20-1            encodes as           EFFFFF
       *   2^20              encodes as           F0100000
       *  -2^20              encodes as           100000
       *   2^64-1            encodes as           FFFFFFFFFFFFFFFFFF
       *  -2^64              encodes as           000000000000000000
       *
       * As the number of bytes is specified in bits 2-9, no value is a prefix of another.
       */
      static class VariableLengthInteger implements ByteSource
      {
          private final long value;
          private int pos;
  
          public VariableLengthInteger(long value)
          {
              long negativeMask = value >> 63;    // -1 for negative, 0 for positive
              value ^= negativeMask;
  
              int bits = 64 - Long.numberOfLeadingZeros(value | 1); // 1 to 63 (can't be 64 because we flip negative numbers)
              int bytes = bits / 7 + 1;   // 0-6 bits 1 byte 7-13 2 bytes etc to 56-63 9 bytes
              if (bytes >= 9)
              {
                  value |= 0x8000000000000000L;   // 8th bit, which doesn't fit the first byte
                  pos = negativeMask < 0 ? 256 : -1; // out of 0-64 range integer such that & 0xFF is 0x00 for negative and 0xFF for positive
              }
              else
              {
                  long mask = (-0x100 >> bytes) & 0xFF; // one in sign bit and as many more as there are extra bytes
                  pos = bytes * 8;
                  value = value | (mask << (pos - 8));
              }
  
              value ^= negativeMask;
              this.value = value;
          }
  
          @Override
          public int next()
          {
              if (pos <= 0 || pos > 64)
              {
                  if (pos == 0)
                      return END_OF_STREAM;
                  else
                  {
                      // 8-byte value, returning first byte
                      int result = pos & 0xFF; // 0x00 for negative numbers, 0xFF for positive
                      pos = 64;
                      return result;
                  }
              }
              pos -= 8;
              return (int) (value >>> pos) & 0xFF;
          }
      }
  
      static class Number implements ByteSource
      {
          private final long value;
          private int pos;
  
          public Number(long value, int length)
          {
              this.value = value;
              this.pos = length;
          }
  
          @Override
          public int next()
          {
              if (pos == 0)
                  return END_OF_STREAM;
              return (int) ((value >> (--pos * 8)) & 0xFF);
          }
      }
  
      /**
       * Fixed length signed floating point number encoding. First bit is sign. If positive, add sign bit value to make
       * greater than all negatives. If not, invert all content to make negatives with bigger magnitude smaller.
       */
      static class SignedFixedLengthFloat<V> implements ByteSource
      {
          private final ValueAccessor<V> accessor;
          private final V data;
          private int bufpos;
          private boolean invert;
  
          public SignedFixedLengthFloat(ValueAccessor<V> accessor, V data)
          {
              this.accessor = accessor;
              this.data = data;
              this.bufpos = 0;
          }
  
          @Override
          public int next()
          {
              if (bufpos >= accessor.size(data))
                  return END_OF_STREAM;
              int v = accessor.getByte(data, bufpos) & 0xFF;
              if (bufpos == 0)
              {
                  invert = v >= 0x80;
                  v |= 0x80;
              }
              if (invert)
                  v = v ^ 0xFF;
              ++bufpos;
              return v;
          }
      }
  
      /**
       * Combination of multiple byte sources. Adds {@link NEXT_COMPONENT} before sources, or {@link NEXT_COMPONENT_NULL} if next is null.
       */
      static class Multi implements ByteSource
      {
          private final ByteSource[] srcs;
          private int srcnum = -1;
          private final int sequenceTerminator;
  
          Multi(ByteSource[] srcs, int sequenceTerminator)
          {
              this.srcs = srcs;
              this.sequenceTerminator = sequenceTerminator;
          }
  
          @Override
          public int next()
          {
              if (srcnum == srcs.length)
                  return END_OF_STREAM;
  
              int b = END_OF_STREAM;
              if (srcnum >= 0 && srcs[srcnum] != null)
                  b = srcs[srcnum].next();
              if (b > END_OF_STREAM)
                  return b;
  
              ++srcnum;
              if (srcnum == srcs.length)
                  return sequenceTerminator;
              if (srcs[srcnum] == null)
                  return NEXT_COMPONENT_NULL;
              return NEXT_COMPONENT;
          }
      }
  
      /**
       * Construct the shortest common prefix of prevMax and currMin that separates those two byte streams.
       * If {@code useCurr == true} the last byte of the returned stream comes from {@code currMin} and is the first
       * byte which is greater than byte on the corresponding position of {@code prevMax}.
       * Otherwise, the last byte of the returned stream comes from {@code prevMax} and is incremented by one, still
       * guaranteeing that it is <= than the byte on the corresponding position of {@code currMin}.
       */
      static class Separator implements ByteSource
      {
          private final ByteSource prev;
          private final ByteSource curr;
          private boolean done = false;
          private final boolean useCurr;
  
          Separator(ByteSource prevMax, ByteSource currMin, boolean useCurr)
          {
              this.prev = prevMax;
              this.curr = currMin;
              this.useCurr = useCurr;
          }
  
          @Override
          public int next()
          {
              if (done)
                  return END_OF_STREAM;
              int p = prev.next();
              int c = curr.next();
              assert p <= c : prev + " not less than " + curr;
              if (p == c)
                  return c;
              done = true;
              return useCurr ? c : p + 1;
          }
      }
  
      static <V> ByteSource optionalFixedLength(ValueAccessor<V> accessor, V data)
      {
          return !accessor.isEmpty(data) ? fixedLength(accessor, data) : null;
      }
  
      /**
       * A byte source of the given bytes without any encoding.
       * The resulting source is only guaranteed to give correct comparison results and be prefix-free if the
       * underlying type has a fixed length.
       * In tests, this method is also used to generate non-escaped test cases.
       */
      public static <V> ByteSource fixedLength(ValueAccessor<V> accessor, V data)
      {
          return new ByteSource()
          {
              int pos = -1;
  
              @Override
              public int next()
              {
                  return ++pos < accessor.size(data) ? accessor.getByte(data, pos) & 0xFF : END_OF_STREAM;
              }
          };
      }
  
      /**
       * A byte source of the given bytes without any encoding.
       * The resulting source is only guaranteed to give correct comparison results and be prefix-free if the
       * underlying type has a fixed length.
       * In tests, this method is also used to generate non-escaped test cases.
       */
      public static ByteSource fixedLength(ByteBuffer b)
      {
          return new ByteSource()
          {
              int pos = b.position() - 1;
  
              @Override
              public int next()
              {
                  return ++pos < b.limit() ? b.get(pos) & 0xFF : END_OF_STREAM;
              }
          };
      }
  
      /**
       * A byte source of the given bytes without any encoding.
       * If used in a sequence, the resulting source is only guaranteed to give correct comparison results if the
       * underlying type has a fixed length.
       * In tests, this method is also used to generate non-escaped test cases.
       */
      public static ByteSource fixedLength(byte[] b)
      {
          return fixedLength(b, 0, b.length);
      }
  
      public static ByteSource fixedLength(byte[] b, int offset, int length)
      {
          checkArgument(offset >= 0 && offset <= b.length);
          checkArgument(length >= 0 && offset + length <= b.length);
  
          return new ByteSource()
          {
              int pos = offset - 1;
  
              @Override
              public int next()
              {
                  return ++pos < offset + length ? b[pos] & 0xFF : END_OF_STREAM;
              }
          };
      }
  
      public class Peekable implements ByteSource
      {
          private static final int NONE = Integer.MIN_VALUE;
  
          private final ByteSource wrapped;
          private int peeked = NONE;
  
          public Peekable(ByteSource wrapped)
          {
              this.wrapped = wrapped;
          }
  
          @Override
          public int next()
          {
              if (peeked != NONE)
              {
                  int val = peeked;
                  peeked = NONE;
                  return val;
              }
              else
                  return wrapped.next();
          }
  
          public int peek()
          {
              if (peeked == NONE)
                  peeked = wrapped.next();
              return peeked;
          }
  
          @Override
          public String toString()
          {
              return "Peekable{" +
                     "wrapped=" + wrapped +
                     ", peeked=" + peeked +
                     '}';
          }
      }
  
      public static Peekable peekable(ByteSource p)
      {
          // When given a null source, we're better off not wrapping it and just returning null. This way existing
          // code that doesn't know about ByteSource.Peekable, but handles correctly null ByteSources won't be thrown
          // off by a non-null instance that semantically should have been null.
          if (p == null)
              return null;
          return (p instanceof Peekable)
                 ? (Peekable) p
                 : new Peekable(p);
      }
  }