/*
    * 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.giraph.utils;
  
  import java.io.IOException;
  import java.lang.reflect.Field;
  
  import static org.apache.giraph.utils.ByteUtils.SIZE_OF_BOOLEAN;
  import static org.apache.giraph.utils.ByteUtils.SIZE_OF_BYTE;
  import static org.apache.giraph.utils.ByteUtils.SIZE_OF_CHAR;
  import static org.apache.giraph.utils.ByteUtils.SIZE_OF_SHORT;
  import static org.apache.giraph.utils.ByteUtils.SIZE_OF_INT;
  import static org.apache.giraph.utils.ByteUtils.SIZE_OF_LONG;
  import static org.apache.giraph.utils.ByteUtils.SIZE_OF_FLOAT;
  import static org.apache.giraph.utils.ByteUtils.SIZE_OF_DOUBLE;
  
  /**
   * Byte array input stream that uses Unsafe methods to deserialize
   * much faster
   */
  @edu.umd.cs.findbugs.annotations.SuppressWarnings(
    "RV_RETURN_VALUE_IGNORED_NO_SIDE_EFFECT")
  public class UnsafeArrayReads extends UnsafeReads {
    /** Access to the unsafe class */
    private static final sun.misc.Unsafe UNSAFE;
    static {
      try {
        Field field = sun.misc.Unsafe.class.getDeclaredField("theUnsafe");
        field.setAccessible(true);
        UNSAFE = (sun.misc.Unsafe) field.get(null);
        // Checkstyle exception due to needing to check if unsafe is allowed
        // CHECKSTYLE: stop IllegalCatch
      } catch (Exception e) {
        // CHECKSTYLE: resume IllegalCatch
        throw new RuntimeException("UnsafeArrayReads: Failed to " +
            "get unsafe", e);
      }
    }
    /** Offset of a byte array */
    private static final long BYTE_ARRAY_OFFSET  =
        UNSAFE.arrayBaseOffset(byte[].class);
  
    /**
     * Constructor
     *
     * @param buf Buffer to read from
     */
    public UnsafeArrayReads(byte[] buf) {
      super(buf.length);
      this.buffer = buf;
    }
  
    /**
     * Constructor.
     *
     * @param buf Buffer to read from
     * @param offset Offsetin the buffer to start reading from
     * @param length Max length of the buffer to read
     */
    public UnsafeArrayReads(byte[] buf, int offset, int length) {
      super(offset, length);
      this.buffer = buf;
    }
  
    @Override
    public int available() {
      return (int) (limit - position);
    }
  
    @Override
    public boolean endOfInput() {
      return available() == 0;
    }
  
  
    @Override
    public int getPos() {
      return (int) position;
    }
  
    @Override
    public void readFully(byte[] b) throws IOException {
      require(b.length);
     System.arraycopy(buffer, (int) position, b, 0, b.length);
     position += b.length;
   }
 
   @Override
   public void readFully(byte[] b, int off, int len) throws IOException {
     require(len);
     System.arraycopy(buffer, (int) position, b, off, len);
     position += len;
   }
 
   @Override
   public boolean readBoolean() {
     require(SIZE_OF_BOOLEAN);
     boolean value = UNSAFE.getBoolean(buffer,
         BYTE_ARRAY_OFFSET + position);
     position += SIZE_OF_BOOLEAN;
     return value;
   }
 
   @Override
   public byte readByte() {
     require(SIZE_OF_BYTE);
     byte value = UNSAFE.getByte(buffer,
         BYTE_ARRAY_OFFSET + position);
     position += SIZE_OF_BYTE;
     return value;
   }
 
   @Override
   public int readUnsignedByte() throws IOException {
     return (short) (readByte() & 0xFF);
   }
 
   @Override
   public short readShort() {
     require(SIZE_OF_SHORT);
     short value = UNSAFE.getShort(buffer,
         BYTE_ARRAY_OFFSET + position);
     position += SIZE_OF_SHORT;
     return value;
   }
 
   @Override
   public int readUnsignedShort() throws IOException {
     return readShort() & 0xFFFF;
   }
 
   @Override
   public char readChar() {
     require(SIZE_OF_CHAR);
     char value = UNSAFE.getChar(buffer,
         BYTE_ARRAY_OFFSET + position);
     position += SIZE_OF_CHAR;
     return value;
   }
 
   @Override
   public int readInt() {
     require(SIZE_OF_INT);
     int value = UNSAFE.getInt(buffer,
         BYTE_ARRAY_OFFSET + position);
     position += SIZE_OF_INT;
     return value;
   }
 
   @Override
   public long readLong() {
     require(SIZE_OF_LONG);
     long value = UNSAFE.getLong(buffer,
         BYTE_ARRAY_OFFSET + position);
     position += SIZE_OF_LONG;
     return value;
   }
 
   @Override
   public float readFloat() {
     require(SIZE_OF_FLOAT);
     float value = UNSAFE.getFloat(buffer,
         BYTE_ARRAY_OFFSET + position);
     position += SIZE_OF_FLOAT;
     return value;
   }
 
   @Override
   public double readDouble() {
     require(SIZE_OF_DOUBLE);
     double value = UNSAFE.getDouble(buffer,
         BYTE_ARRAY_OFFSET + position);
     position += SIZE_OF_DOUBLE;
     return value;
   }
 
   /**
    * Get an int at an arbitrary position in a byte[]
    *
    * @param buf Buffer to get the int from
    * @param pos Position in the buffer to get the int from
    * @return Int at the buffer position
    */
   public static int getInt(byte[] buf, int pos) {
     return UNSAFE.getInt(buf,
         BYTE_ARRAY_OFFSET + pos);
   }
 }