/*

  * 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);

   }

 }