/*

  * 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 com.esotericsoftware.kryo.io.Output;

 import java.io.IOException;

 import java.io.UTFDataFormatException;

 import java.lang.reflect.Field;

 import java.util.Arrays;

 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 output stream that uses Unsafe methods to serialize/deserialize

  * much faster.

  *

  * This stream now extends com.esotericsoftware.kryo.io.Output so that kryo

  * serialization can directly write to this stream without using an

  * additional buffer, providing a faster serialization.

  *

  * Users of this class has to explicitly close the stream to avoid style check

  * errors even though close is no-op when the underlying stream is not set.

  */

 public class UnsafeByteArrayOutputStream extends Output

   implements ExtendedDataOutput {

   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("UnsafeByteArrayOutputStream: Failed to " +

           "get unsafe", e);

     }

   }

   /** Default number of bytes */

   private static final int DEFAULT_BYTES = 32;

   /** Access to the unsafe class */

   private static final sun.misc.Unsafe UNSAFE;

   /** Offset of a byte array */

   private static final long BYTE_ARRAY_OFFSET  =

       UNSAFE.arrayBaseOffset(byte[].class);

   /**

    * Constructor

    */

   public UnsafeByteArrayOutputStream() {

     this(DEFAULT_BYTES);

   }

   /**

    * Constructor

    *

    * @param size Initial size of the underlying byte array

    */

   public UnsafeByteArrayOutputStream(int size) {

     buffer = new byte[size];

     capacity = size;

   }

   /**

    * Constructor to take in a buffer

    *

    * @param buf Buffer to start with, or if null, create own buffer

    */

   public UnsafeByteArrayOutputStream(byte[] buf) {

     if (buf == null) {

       this.buffer = new byte[DEFAULT_BYTES];

     } else {

       this.buffer = buf;

     }

     capacity = this.buffer.length;

   }

   /**

    * Constructor to take in a buffer with a given position into that buffer

    *

    * @param buf Buffer to start with

    * @param pos Position to write at the buffer

    */

   public UnsafeByteArrayOutputStream(byte[] buf, int pos) {

     this(buf);

     this.position = pos;

   }

   /**

    * Ensure that this buffer has enough remaining space to add the size.

    * Creates and copies to a new buffer if necessary

    *

    * @param size Size to add

    */

   @Override

   protected boolean require(int size) {

     if (position + size > buffer.length) {

       byte[] newBuf = new byte[(buffer.length + size) << 1];

       System.arraycopy(buffer, 0, newBuf, 0, position);

       buffer = newBuf;

       capacity = buffer.length;

       return true;

     }

     return false;

   }

   @Override

   public byte[] getByteArray() {

     return buffer;

   }

   @Override

   public byte[] toByteArray() {

     return Arrays.copyOf(buffer, position);

   }

   @Override

   public byte[] toByteArray(int offset, int length) {

     if (offset + length > position) {

       throw new IndexOutOfBoundsException(String.format("Offset: %d + " +

         "Length: %d exceeds the size of buffer : %d",

             offset, length, position));

     }

     return Arrays.copyOfRange(buffer, offset, length);

   }

   @Override

   public void reset() {

     position = 0;

   }

   @Override

   public int getPos() {

     return position;

   }

   @Override

   public void write(int b) {

     require(SIZE_OF_BYTE);

     buffer[position] = (byte) b;

     position += SIZE_OF_BYTE;

   }

   @Override

   public void write(byte[] b) {

     require(b.length);

     System.arraycopy(b, 0, buffer, position, b.length);

     position += b.length;

   }

   @Override

   public void write(byte[] b, int off, int len) {

     require(len);

     System.arraycopy(b, off, buffer, position, len);

     position += len;

   }

   @Override

   public void writeBoolean(boolean v) {

     require(SIZE_OF_BOOLEAN);

     UNSAFE.putBoolean(buffer, BYTE_ARRAY_OFFSET + position, v);

     position += SIZE_OF_BOOLEAN;

   }

   @Override

   public void writeByte(int v) {

     require(SIZE_OF_BYTE);

     UNSAFE.putByte(buffer, BYTE_ARRAY_OFFSET + position, (byte) v);

     position += SIZE_OF_BYTE;

   }

   @Override

   public void writeShort(int v) {

     require(SIZE_OF_SHORT);

     UNSAFE.putShort(buffer, BYTE_ARRAY_OFFSET + position, (short) v);

     position += SIZE_OF_SHORT;

   }

   @Override

   public void writeChar(int v) throws IOException {

     require(SIZE_OF_CHAR);

     UNSAFE.putChar(buffer, BYTE_ARRAY_OFFSET + position, (char) v);

     position += SIZE_OF_CHAR;

   }

   @Override

   public void writeChar(char v) {

     require(SIZE_OF_CHAR);

     UNSAFE.putChar(buffer, BYTE_ARRAY_OFFSET + position, v);

     position += SIZE_OF_CHAR;

   }

   @Override

   public void writeInt(int v) {

     require(SIZE_OF_INT);

     UNSAFE.putInt(buffer, BYTE_ARRAY_OFFSET + position, v);

     position += SIZE_OF_INT;

   }

   @Override

   public void ensureWritable(int minSize) {

     if ((position + minSize) > buffer.length) {

       buffer = Arrays.copyOf(buffer,

                 Math.max(buffer.length << 1, position + minSize));

     }

   }

   @Override

   public void skipBytes(int bytesToSkip) {

     ensureWritable(bytesToSkip);

     position += bytesToSkip;

   }

   @Override

   public void writeInt(int pos, int value) {

     if (pos + SIZE_OF_INT > this.position) {

       throw new IndexOutOfBoundsException(

           "writeInt: Tried to write int to position " + pos +

               " but current length is " + this.position);

     }

     UNSAFE.putInt(buffer, BYTE_ARRAY_OFFSET + pos, value);

   }

   @Override

   public void writeLong(long v) {

     require(SIZE_OF_LONG);

     UNSAFE.putLong(buffer, BYTE_ARRAY_OFFSET + position, v);

     position += SIZE_OF_LONG;

   }

   @Override

   public void writeFloat(float v) {

     require(SIZE_OF_FLOAT);

     UNSAFE.putFloat(buffer, BYTE_ARRAY_OFFSET + position, v);

     position += SIZE_OF_FLOAT;

   }

   @Override

   public void writeDouble(double v) {

     require(SIZE_OF_DOUBLE);

     UNSAFE.putDouble(buffer, BYTE_ARRAY_OFFSET + position, v);

     position += SIZE_OF_DOUBLE;

   }

   @Override

   public void writeBytes(String s) throws IOException {

     // Note that this code is mostly copied from DataOutputStream

     int len = s.length();

     require(len);

     for (int i = 0; i < len; i++) {

       int v = s.charAt(i);

       writeByte(v);

     }

   }

   @Override

   public void writeChars(String s) throws IOException {

     // Note that this code is mostly copied from DataOutputStream

     int len = s.length();

     require(len * SIZE_OF_CHAR);

     for (int i = 0; i < len; i++) {

       int v = s.charAt(i);

       writeChar(v);

     }

   }

   @Override

   public void writeUTF(String s) throws IOException {

     // Note that this code is mostly copied from DataOutputStream

     int strlen = s.length();

     int utflen = 0;

     int c;

     /* use charAt instead of copying String to char array */

     for (int i = 0; i < strlen; i++) {

       c = s.charAt(i);

       if ((c >= 0x0001) && (c <= 0x007F)) {

         utflen++;

       } else if (c > 0x07FF) {

         utflen += 3;

       } else {

         utflen += 2;

       }

     }

     if (utflen > 65535) {

       throw new UTFDataFormatException(

           "encoded string too long: " + utflen + " bytes");

     }

     require(utflen + SIZE_OF_SHORT);

     writeShort(utflen);

     int i = 0;

     for (i = 0; i < strlen; i++) {

       c = s.charAt(i);

       if (!((c >= 0x0001) && (c <= 0x007F))) {

         break;

       }

       buffer[position++] = (byte) c;

     }

     for (; i < strlen; i++) {

       c = s.charAt(i);

       if ((c >= 0x0001) && (c <= 0x007F)) {

         buffer[position++] = (byte) c;

       } else if (c > 0x07FF) {

         buffer[position++] = (byte) (0xE0 | ((c >> 12) & 0x0F));

         buffer[position++] = (byte) (0x80 | ((c >>  6) & 0x3F));

         buffer[position++] = (byte) (0x80 | ((c >>  0) & 0x3F));

       } else {

         buffer[position++] = (byte) (0xC0 | ((c >>  6) & 0x1F));

         buffer[position++] = (byte) (0x80 | ((c >>  0) & 0x3F));

       }

     }

   }

 }