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

import org.apache.giraph.conf.DefaultImmutableClassesGiraphConfigurable;
import org.apache.giraph.edge.Edge;
import org.apache.giraph.utils.Trimmable;
import org.apache.giraph.edge.MultiRandomAccessOutEdges;
import org.apache.giraph.edge.MutableEdge;
import org.apache.giraph.edge.MutableEdgesIterable;
import org.apache.giraph.edge.MutableEdgesWrapper;
import org.apache.giraph.edge.MutableOutEdges;
import org.apache.giraph.edge.OutEdges;
import org.apache.giraph.edge.StrictRandomAccessOutEdges;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableComparable;

import com.google.common.collect.UnmodifiableIterator;

import java.util.Iterator;

/**
 * Class which holds vertex id, data and edges.
 *
 * @param <I> Vertex id
 * @param <V> Vertex data
 * @param <E> Edge data
 */
public class DefaultVertex<I extends WritableComparable,
    V extends Writable, E extends Writable>
    extends DefaultImmutableClassesGiraphConfigurable<I, V, E>
    implements Vertex<I, V, E>, Trimmable {
  /** Vertex id. */
  private I id;
  /** Vertex value. */
  private V value;
  /** Outgoing edges. */
  private OutEdges<I, E> edges;
  /** If true, do not do anymore computation on this vertex. */
  private boolean halt;

  @Override
  public void initialize(I id, V value, Iterable<Edge<I, E>> edges) {
    this.id = id;
    this.value = value;
    setEdges(edges);
  }

  @Override
  public void initialize(I id, V value) {
    this.id = id;
    this.value = value;
    this.edges = getConf().createAndInitializeOutEdges(0);
  }

  @Override
  public void setEdges(Iterable<Edge<I, E>> edges) {
    // If the iterable is actually an instance of OutEdges,
    // we simply take the reference.
    // Otherwise, we initialize a new OutEdges.
    if (edges instanceof OutEdges) {
      this.edges = (OutEdges<I, E>) edges;
    } else {
      this.edges = getConf().createAndInitializeOutEdges(edges);
    }
  }

  @Override
  public I getId() {
    return id;
  }

  @Override
  public V getValue() {
    return value;
  }

  @Override
  public void setValue(V value) {
    this.value = value;
  }

  @Override
  public Iterable<Edge<I, E>> getEdges() {
    return edges;
  }

  @Override
  public Iterable<MutableEdge<I, E>> getMutableEdges() {
    // If the OutEdges implementation has a specialized mutable iterator,
    // we use that; otherwise, we build a new data structure as we iterate
    // over the current edges.
    if (edges instanceof MutableOutEdges) {
      return new Iterable<MutableEdge<I, E>>() {
        @Override
        public Iterator<MutableEdge<I, E>> iterator() {
          return ((MutableOutEdges<I, E>) edges).mutableIterator();
        }
      };
    } else {
      return new MutableEdgesIterable<I, E>(this);
    }
  }

  @Override
  public void unwrapMutableEdges() {
    if (edges instanceof MutableEdgesWrapper) {
      edges = ((MutableEdgesWrapper<I, E>) edges).unwrap();
    }
  }

  @Override
  public int getNumEdges() {
    return edges.size();
  }

  @Override
  public E getEdgeValue(I targetVertexId) {
    // If the OutEdges implementation has a specialized random-access
    // method, we use that; otherwise, we scan the edges.
    if (edges instanceof StrictRandomAccessOutEdges) {
      return ((StrictRandomAccessOutEdges<I, E>) edges)
          .getEdgeValue(targetVertexId);
    } else {
      for (Edge<I, E> edge : edges) {
        if (edge.getTargetVertexId().equals(targetVertexId)) {
          return edge.getValue();
        }
      }
      return null;
    }
  }

  @Override
  public void setEdgeValue(I targetVertexId, E edgeValue) {
    // If the OutEdges implementation has a specialized random-access
    // method, we use that; otherwise, we scan the edges.
    if (edges instanceof StrictRandomAccessOutEdges) {
      ((StrictRandomAccessOutEdges<I, E>) edges).setEdgeValue(
          targetVertexId, edgeValue);
    } else {
      for (MutableEdge<I, E> edge : getMutableEdges()) {
        if (edge.getTargetVertexId().equals(targetVertexId)) {
          edge.setValue(edgeValue);
        }
      }
    }
  }

  @Override
  public Iterable<E> getAllEdgeValues(final I targetVertexId) {
    // If the OutEdges implementation has a specialized random-access
    // method, we use that; otherwise, we scan the edges.
    if (edges instanceof MultiRandomAccessOutEdges) {
      return ((MultiRandomAccessOutEdges<I, E>) edges)
          .getAllEdgeValues(targetVertexId);
    } else {
      return new Iterable<E>() {
        @Override
        public Iterator<E> iterator() {
          return new UnmodifiableIterator<E>() {
            /** Iterator over all edges. */
            private Iterator<Edge<I, E>> edgeIterator = edges.iterator();
            /** Last matching edge found. */
            private Edge<I, E> currentEdge;

            @Override
            public boolean hasNext() {
              while (edgeIterator.hasNext()) {
                currentEdge = edgeIterator.next();
                if (currentEdge.getTargetVertexId().equals(targetVertexId)) {
                  return true;
                }
              }
              return false;
            }

            @Override
            public E next() {
              return currentEdge.getValue();
            }
          };
        }
      };
    }
  }

  @Override
  public void voteToHalt() {
    halt = true;
  }

  @Override
  public void wakeUp() {
    halt = false;
  }

  @Override
  public boolean isHalted() {
    return halt;
  }

  @Override
  public void trim() {
    if (edges instanceof Trimmable) {
      ((Trimmable) edges).trim();
    }
  }

  @Override
  public void addEdge(Edge<I, E> edge) {
    edges.add(edge);
  }

  @Override
  public void removeEdges(I targetVertexId) {
    edges.remove(targetVertexId);
  }

  @Override
  public String toString() {
    return "Vertex(id=" + getId() + ",value=" + getValue() +
        ",#edges=" + getNumEdges() + ")";
  }
}