/**
 * 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.hadoop.io;

import java.io.IOException;
import java.util.Arrays;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import java.io.DataInput;
import java.io.DataOutput;

import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;

/** 
 * A byte sequence that is usable as a key or value.
 * It is resizable and distinguishes between the size of the sequence and
 * the current capacity. The hash function is the front of the md5 of the 
 * buffer. The sort order is the same as memcmp.
 */
@InterfaceAudience.Public
@InterfaceStability.Stable
public class BytesWritable extends BinaryComparable
    implements WritableComparable<BinaryComparable> {
  private static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;
  private static final int LENGTH_BYTES = 4;

  private static final byte[] EMPTY_BYTES = new byte[0];

  private int size;
  private byte[] bytes;

  /**
   * Create a zero-size sequence.
   */
  public BytesWritable() {
    this.bytes = EMPTY_BYTES;
    this.size = 0;
  }

  /**
   * Create a BytesWritable using the byte array as the initial value.
   * @param bytes This array becomes the backing storage for the object.
   */
  public BytesWritable(byte[] bytes) {
    this(bytes, bytes.length);
  }

  /**
   * Create a BytesWritable using the byte array as the initial value
   * and length as the length. Use this constructor if the array is larger
   * than the value it represents.
   * @param bytes This array becomes the backing storage for the object.
   * @param length The number of bytes to use from array.
   */
  public BytesWritable(byte[] bytes, int length) {
    this.bytes = bytes;
    this.size = length;
  }

  /**
   * Get a copy of the bytes that is exactly the length of the data.
   * See {@link #getBytes()} for faster access to the underlying array.
   *
   * @return copyBytes.
   */
  public byte[] copyBytes() {
    return Arrays.copyOf(bytes, size);
  }

  /**
   * Get the data backing the BytesWritable. Please use {@link #copyBytes()}
   * if you need the returned array to be precisely the length of the data.
   * @return The data is only valid between 0 and getLength() - 1.
   */
  @Override
  public byte[] getBytes() {
    return bytes;
  }

  /**
   * Get the data from the BytesWritable.
   * @deprecated Use {@link #getBytes()} instead.
   * @return data from the BytesWritable.
   */
  @Deprecated
  public byte[] get() {
    return getBytes();
  }

  /**
   * Get the current size of the buffer.
   */
  @Override
  public int getLength() {
    return size;
  }

  /**
   * Get the current size of the buffer.
   * @deprecated Use {@link #getLength()} instead.
   * @return current size of the buffer.
   */
  @Deprecated
  public int getSize() {
    return getLength();
  }

  /**
   * Change the size of the buffer. The values in the old range are preserved
   * and any new values are undefined. The capacity is changed if it is 
   * necessary.
   * @param size The new number of bytes
   */
  public void setSize(int size) {
    if (size > getCapacity()) {
      // Avoid overflowing the int too early by casting to a long.
      long newSize = Math.min(MAX_ARRAY_SIZE, (3L * size) / 2L);
      setCapacity((int) newSize);
    }
    this.size = size;
  }

  /**
   * Get the capacity, which is the maximum size that could handled without
   * resizing the backing storage.
   *
   * @return The number of bytes
   */
  public int getCapacity() {
    return bytes.length;
  }

  /**
   * Change the capacity of the backing storage. The data is preserved.
   *
   * @param capacity The new capacity in bytes.
   */
  public void setCapacity(final int capacity) {
    if (capacity != getCapacity()) {
      this.size = Math.min(size, capacity);
      this.bytes = Arrays.copyOf(this.bytes, capacity);
    }
  }

  /**
   * Set the BytesWritable to the contents of the given newData.
   *
   * @param newData the value to set this BytesWritable to.
   */
  public void set(BytesWritable newData) {
    set(newData.bytes, 0, newData.size);
  }

  /**
   * Set the value to a copy of the given byte range.
   *
   * @param newData the new values to copy in
   * @param offset the offset in newData to start at
   * @param length the number of bytes to copy
   */
  public void set(byte[] newData, int offset, int length) {
    setSize(0);
    setSize(length);
    System.arraycopy(newData, offset, bytes, 0, size);
  }

  @Override
  public void readFields(DataInput in) throws IOException {
    setSize(0); // clear the old data
    setSize(in.readInt());
    in.readFully(bytes, 0, size);
  }

  @Override
  public void write(DataOutput out) throws IOException {
    out.writeInt(size);
    out.write(bytes, 0, size);
  }

  /**
   * Are the two byte sequences equal?
   */
  @Override
  public boolean equals(Object right_obj) {
    if (right_obj instanceof BytesWritable)
      return super.equals(right_obj);
    return false;
  }

  @Override
  public int hashCode() {
    return super.hashCode();
  }

  /**
   * Generate the stream of bytes as hex pairs separated by ' '.
   */
  @Override
  public String toString() {
    return IntStream.range(0, size)
        .mapToObj(idx -> String.format("%02x", bytes[idx]))
        .collect(Collectors.joining(" "));
  }

  /** A Comparator optimized for BytesWritable. */ 
  public static class Comparator extends WritableComparator {
    public Comparator() {
      super(BytesWritable.class);
    }

    /**
     * Compare the buffers in serialized form.
     */
    @Override
    public int compare(byte[] b1, int s1, int l1,
                       byte[] b2, int s2, int l2) {
      return compareBytes(b1, s1 + LENGTH_BYTES, l1 - LENGTH_BYTES,
                          b2, s2 + LENGTH_BYTES, l2 - LENGTH_BYTES);
    }
  }

  static {                                        // register this comparator
    WritableComparator.define(BytesWritable.class, new Comparator());
  }

}
