/*
    * 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.io;
  
  import org.apache.giraph.conf.ImmutableClassesGiraphConfiguration;
  import org.apache.giraph.utils.ExtendedDataOutput;
  import org.apache.hadoop.io.Writable;
  
  import com.google.common.collect.Iterables;
  import com.google.common.collect.Lists;
  
  import java.io.DataInput;
  import java.io.DataOutput;
  import java.io.IOException;
  import java.util.ArrayList;
  import java.util.List;
  
  /**
   * Implementations of {@link ExtendedDataOutput} are limited because they can
   * only handle up to 1GB of data. This {@link DataOutput} overcomes that
   * limitation, with almost no additional cost when data is not huge.
   *
   * Goes in pair with {@link BigDataInput}
   */
  public class BigDataOutput implements DataOutput, Writable {
    /** Default initial size of the stream */
    private static final int DEFAULT_INITIAL_SIZE = 16;
    /** Max allowed size of the stream */
    private static final int MAX_SIZE = 1 << 25;
    /**
     * Create a new stream when we have less then this number of bytes left in
     * the stream. Should be larger than the largest serialized primitive.
     */
    private static final int SIZE_DELTA = 100;
  
    /** Data output which we are currently writing to */
    protected ExtendedDataOutput currentDataOutput;
    /** List of filled outputs, will be null until we get a lot of data */
    protected List<ExtendedDataOutput> dataOutputs;
    /** Configuration */
    protected final ImmutableClassesGiraphConfiguration conf;
  
    /**
     * Constructor
     *
     * @param conf Configuration
     */
    public BigDataOutput(ImmutableClassesGiraphConfiguration conf) {
      this(DEFAULT_INITIAL_SIZE, conf);
    }
  
    /**
     * Constructor
     *
     * @param initialSize Initial size of data output
     * @param conf        Configuration
     */
    public BigDataOutput(int initialSize,
        ImmutableClassesGiraphConfiguration conf) {
      this.conf = conf;
      dataOutputs = null;
      currentDataOutput = createOutput(initialSize);
    }
  
    /**
     * Get max size for single data output
     *
     * @return Max size for single data output
     */
    protected int getMaxSize() {
      return MAX_SIZE;
    }
  
    /**
     * Create next data output
     *
     * @param size Size of data output to create
     * @return Created data output
     */
    protected ExtendedDataOutput createOutput(int size) {
      return conf.createExtendedDataOutput(size);
    }
  
   /**
    * Get DataOutput which data should be written to. If current DataOutput is
    * full it will create a new one.
    *
    * @return DataOutput which data should be written to
    */
   private ExtendedDataOutput getDataOutputToWriteTo() {
     return getDataOutputToWriteTo(SIZE_DELTA);
   }
 
   /**
    * Get DataOutput which data should be written to. If current DataOutput is
    * full it will create a new one.
    *
    * @param additionalSize How many additional bytes we need space for
    * @return DataOutput which data should be written to
    */
   private ExtendedDataOutput getDataOutputToWriteTo(int additionalSize) {
     if (currentDataOutput.getPos() + additionalSize > getMaxSize()) {
       if (dataOutputs == null) {
         dataOutputs = new ArrayList<>(1);
       }
       dataOutputs.add(currentDataOutput);
       currentDataOutput = createOutput(getMaxSize());
     }
     return currentDataOutput;
   }
 
   /**
    * Get number of DataOutputs which contain written data.
    *
    * @return Number of DataOutputs which contain written data
    */
   public int getNumberOfDataOutputs() {
     return (dataOutputs == null) ? 1 : dataOutputs.size() + 1;
   }
 
   /**
    * Get DataOutputs which contain written data.
    *
    * @return DataOutputs which contain written data
    */
   public Iterable<ExtendedDataOutput> getDataOutputs() {
     ArrayList<ExtendedDataOutput> currentList =
         Lists.newArrayList(currentDataOutput);
     if (dataOutputs == null) {
       return currentList;
     } else {
       return Iterables.concat(dataOutputs, currentList);
     }
   }
 
   public ImmutableClassesGiraphConfiguration getConf() {
     return conf;
   }
 
   /**
    * Get number of bytes written to this data output
    *
    * @return Size in bytes
    */
   public long getSize() {
     long size = currentDataOutput.getPos();
     if (dataOutputs != null) {
       for (ExtendedDataOutput dataOutput : dataOutputs) {
         size += dataOutput.getPos();
       }
     }
     return size;
   }
 
   @Override
   public void write(int b) throws IOException {
     getDataOutputToWriteTo().write(b);
   }
 
   @Override
   public void write(byte[] b) throws IOException {
     write(b, 0, b.length);
   }
 
   @Override
   public void write(byte[] b, int off, int len) throws IOException {
     if (len <= getMaxSize()) {
       getDataOutputToWriteTo(len).write(b, off, len);
     } else {
       // When we try to write more bytes than the biggest size of single data
       // output, we need to split up the byte array into multiple chunks
       while (len > 0) {
         int toWrite = Math.min(getMaxSize(), len);
         write(b, off, toWrite);
         len -= toWrite;
         off += toWrite;
       }
     }
   }
 
   @Override
   public void writeBoolean(boolean v) throws IOException {
     getDataOutputToWriteTo().writeBoolean(v);
   }
 
   @Override
   public void writeByte(int v) throws IOException {
     getDataOutputToWriteTo().writeByte(v);
   }
 
   @Override
   public void writeShort(int v) throws IOException {
     getDataOutputToWriteTo().writeShort(v);
   }
 
   @Override
   public void writeChar(int v) throws IOException {
     getDataOutputToWriteTo().writeChar(v);
   }
 
   @Override
   public void writeInt(int v) throws IOException {
     getDataOutputToWriteTo().writeInt(v);
   }
 
   @Override
   public void writeLong(long v) throws IOException {
     getDataOutputToWriteTo().writeLong(v);
   }
 
   @Override
   public void writeFloat(float v) throws IOException {
     getDataOutputToWriteTo().writeFloat(v);
   }
 
   @Override
   public void writeDouble(double v) throws IOException {
     getDataOutputToWriteTo().writeDouble(v);
   }
 
   @Override
   public void writeBytes(String s) throws IOException {
     getDataOutputToWriteTo().writeBytes(s);
   }
 
   @Override
   public void writeChars(String s) throws IOException {
     getDataOutputToWriteTo().writeChars(s);
   }
 
   @Override
   public void writeUTF(String s) throws IOException {
     getDataOutputToWriteTo().writeUTF(s);
   }
 
   /**
    * Write one of data outputs to another data output
    *
    * @param dataOutput Data output to write
    * @param out        Data output to write to
    */
   private void writeExtendedDataOutput(ExtendedDataOutput dataOutput,
       DataOutput out) throws IOException {
     out.writeInt(dataOutput.getPos());
     out.write(dataOutput.getByteArray(), 0, dataOutput.getPos());
   }
 
   /**
    * Read data output from data input
    *
    * @param in Data input to read from
    * @return Data output read
    */
   private ExtendedDataOutput readExtendedDataOutput(
       DataInput in) throws IOException {
     int length = in.readInt();
     byte[] data = new byte[length];
     in.readFully(data);
     return conf.createExtendedDataOutput(data, data.length);
   }
 
   @Override
   public void write(DataOutput out) throws IOException {
     if (dataOutputs == null) {
       out.writeInt(0);
     } else {
       out.writeInt(dataOutputs.size());
       for (ExtendedDataOutput stream : dataOutputs) {
         writeExtendedDataOutput(stream, out);
       }
     }
     writeExtendedDataOutput(currentDataOutput, out);
   }
 
   @Override
   public void readFields(DataInput in) throws IOException {
     int size = in.readInt();
     if (size == 0) {
       dataOutputs = null;
     } else {
       dataOutputs = new ArrayList<ExtendedDataOutput>(size);
       while (size-- > 0) {
         dataOutputs.add(readExtendedDataOutput(in));
       }
     }
     currentDataOutput = readExtendedDataOutput(in);
   }
 }