/*
    * 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.edge;
  
  import com.google.common.collect.UnmodifiableIterator;
  import org.apache.giraph.utils.ExtendedDataInput;
  import org.apache.giraph.utils.ExtendedDataOutput;
  import org.apache.giraph.utils.Trimmable;
  import org.apache.giraph.utils.WritableUtils;
  import org.apache.hadoop.io.Writable;
  import org.apache.hadoop.io.WritableComparable;
  
  import java.io.DataInput;
  import java.io.DataOutput;
  import java.io.IOException;
  import java.util.Arrays;
  import java.util.Iterator;
  import java.util.LinkedList;
  import java.util.List;
  import java.util.Collections;
  
  /**
   * {@link OutEdges} implementation backed by a byte array.
   * Parallel edges are allowed.
   * Note: this implementation is optimized for space usage,
   * but edge removals are expensive.
   *
   * @param <I> Vertex id
   * @param <E> Edge value
   */
  public class ByteArrayEdges<I extends WritableComparable, E extends Writable>
      extends ConfigurableOutEdges<I, E>
      implements ReuseObjectsOutEdges<I, E>, Trimmable {
    /** Serialized edges. */
    private byte[] serializedEdges;
    /** Number of bytes used in serializedEdges. */
    private int serializedEdgesBytesUsed;
    /** Number of edges. */
    private int edgeCount;
  
    @Override
    public void initialize(Iterable<Edge<I, E>> edges) {
      ExtendedDataOutput extendedOutputStream =
          getConf().createExtendedDataOutput();
      for (Edge<I, E> edge : edges) {
        try {
          WritableUtils.writeEdge(extendedOutputStream, edge);
        } catch (IOException e) {
          throw new IllegalStateException("initialize: Failed to serialize " +
              edge);
        }
        ++edgeCount;
      }
      serializedEdges = extendedOutputStream.getByteArray();
      serializedEdgesBytesUsed = extendedOutputStream.getPos();
    }
  
    @Override
    public void initialize(int capacity) {
      // We have no way to know the size in bytes used by a certain
      // number of edges.
      initialize();
    }
  
    @Override
    public void initialize() {
      // No-op: no need to initialize the byte-array if there are no edges,
      // since add() and iterator() work fine with a null buffer.
    }
  
    @Override
    public void add(Edge<I, E> edge) {
      ExtendedDataOutput extendedDataOutput =
          getConf().createExtendedDataOutput(
              serializedEdges, serializedEdgesBytesUsed);
      try {
        WritableUtils.writeEdge(extendedDataOutput, edge);
      } catch (IOException e) {
        throw new IllegalStateException("add: Failed to write to the new " +
            "byte array");
      } catch (NegativeArraySizeException negativeArraySizeException) {
        throw new IllegalStateException("add: Too many edges for a vertex, " +
          "hence failed to write to byte array");
     }
     serializedEdges = extendedDataOutput.getByteArray();
     serializedEdgesBytesUsed = extendedDataOutput.getPos();
     ++edgeCount;
   }
 
   @Override
   public void remove(I targetVertexId) {
     // Note that this is very expensive (deserializes all edges).
     ByteArrayEdgeIterator iterator = new ByteArrayEdgeIterator();
     List<Integer> foundStartOffsets = new LinkedList<Integer>();
     List<Integer> foundEndOffsets = new LinkedList<Integer>();
     int lastStartOffset = 0;
     while (iterator.hasNext()) {
       Edge<I, E> edge = iterator.next();
       if (edge.getTargetVertexId().equals(targetVertexId)) {
         foundStartOffsets.add(lastStartOffset);
         foundEndOffsets.add(iterator.extendedDataInput.getPos());
         --edgeCount;
       }
       lastStartOffset = iterator.extendedDataInput.getPos();
     }
     foundStartOffsets.add(serializedEdgesBytesUsed);
 
     Iterator<Integer> foundStartOffsetIter = foundStartOffsets.iterator();
     Integer foundStartOffset = foundStartOffsetIter.next();
     for (Integer foundEndOffset : foundEndOffsets) {
       Integer nextFoundStartOffset = foundStartOffsetIter.next();
       System.arraycopy(serializedEdges, foundEndOffset,
           serializedEdges, foundStartOffset,
           nextFoundStartOffset - foundEndOffset);
       serializedEdgesBytesUsed -= foundEndOffset - foundStartOffset;
       foundStartOffset = nextFoundStartOffset;
     }
   }
 
   @Override
   public int size() {
     return edgeCount;
   }
 
   @Override
   public void trim() {
     if (serializedEdges != null &&
         serializedEdges.length > serializedEdgesBytesUsed) {
       serializedEdges =
           Arrays.copyOf(serializedEdges, serializedEdgesBytesUsed);
     }
   }
 
   /**
    * Iterator that reuses the same Edge object.
    */
   private class ByteArrayEdgeIterator
       extends UnmodifiableIterator<Edge<I, E>> {
     /** Input for processing the bytes */
     private ExtendedDataInput extendedDataInput =
         getConf().createExtendedDataInput(
             serializedEdges, 0, serializedEdgesBytesUsed);
     /** Representative edge object. */
     private ReusableEdge<I, E> representativeEdge =
         getConf().createReusableEdge();
 
     @Override
     public boolean hasNext() {
       return serializedEdges != null && !extendedDataInput.endOfInput();
     }
 
     @Override
     public Edge<I, E> next() {
       try {
         WritableUtils.readEdge(extendedDataInput, representativeEdge);
       } catch (IOException e) {
         throw new IllegalStateException("next: Failed on pos " +
             extendedDataInput.getPos() + " edge " + representativeEdge);
       }
       return representativeEdge;
     }
   }
 
   @Override
   public Iterator<Edge<I, E>> iterator() {
     if (edgeCount == 0) {
       return Collections.emptyListIterator();
     } else {
       return new ByteArrayEdgeIterator();
     }
   }
 
   @Override
   public void readFields(DataInput in) throws IOException {
     serializedEdgesBytesUsed = in.readInt();
     if (serializedEdgesBytesUsed > 0) {
       // Only create a new buffer if the old one isn't big enough
       if (serializedEdges == null ||
           serializedEdgesBytesUsed > serializedEdges.length) {
         serializedEdges = new byte[serializedEdgesBytesUsed];
       }
       in.readFully(serializedEdges, 0, serializedEdgesBytesUsed);
     }
     edgeCount = in.readInt();
   }
 
   @Override
   public void write(DataOutput out) throws IOException {
     out.writeInt(serializedEdgesBytesUsed);
     if (serializedEdgesBytesUsed > 0) {
       out.write(serializedEdges, 0, serializedEdgesBytesUsed);
     }
     out.writeInt(edgeCount);
   }
 }