/*
    * 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.block_app.library;
  
  import static com.google.common.base.Preconditions.checkNotNull;
  
  import java.util.Iterator;
  
  import org.apache.giraph.comm.SendMessageCache.TargetVertexIdIterator;
  import org.apache.giraph.edge.Edge;
  import org.apache.giraph.function.PairPredicate;
  import org.apache.giraph.function.Predicate;
  import org.apache.giraph.function.primitive.PrimitiveRefs.IntRef;
  import org.apache.giraph.function.vertex.SupplierFromVertex;
  import org.apache.giraph.graph.Vertex;
  import org.apache.hadoop.io.Writable;
  import org.apache.hadoop.io.WritableComparable;
  
  import com.google.common.collect.AbstractIterator;
  import com.google.common.collect.Iterators;
  import com.google.common.collect.UnmodifiableIterator;
  
  /**
   * SupplierFromVertex that extract common information from
   * vertex itself.
   */
  @SuppressWarnings("rawtypes")
  public class VertexSuppliers {
    /** Hide constructor */
    private VertexSuppliers() { }
  
    /**
     * Supplier which extracts and returns vertex ID.
     * (note - do not modify the object, as it is not returning a copy)
     */
    public static
    <I extends WritableComparable, V extends Writable, E extends Writable>
    SupplierFromVertex<I, V, E, I> vertexIdSupplier() {
      return new SupplierFromVertex<I, V, E, I>() {
        @Override
        public I get(Vertex<I, V, E> vertex) {
          return vertex.getId();
        }
      };
    }
  
    /**
     * Supplier which extracts and returns vertex value.
     * (note - doesn't create a copy of vertex value)
     */
    public static
    <I extends WritableComparable, V extends Writable, E extends Writable>
    SupplierFromVertex<I, V, E, V> vertexValueSupplier() {
      return new SupplierFromVertex<I, V, E, V>() {
        @Override
        public V get(Vertex<I, V, E> vertex) {
          return vertex.getValue();
        }
      };
    }
  
    /**
     * Supplier which extracts and returns edges object.
     * (note - doesn't create a copy of vertex value)
     */
    public static
    <I extends WritableComparable, V extends Writable, E extends Writable>
    SupplierFromVertex<I, V, E, Iterable<Edge<I, E>>> vertexEdgesSupplier() {
      return new SupplierFromVertex<I, V, E, Iterable<Edge<I, E>>>() {
        @Override
        public Iterable<Edge<I, E>> get(Vertex<I, V, E> vertex) {
          return vertex.getEdges();
        }
      };
    }
  
    /**
     * Supplier which extracts and returns Iterator over all neighbor IDs.
     * Note - iterator returns reused object, so you need to "use" them,
     * before calling next() again.
     */
    public static
    <I extends WritableComparable, V extends Writable, E extends Writable>
    SupplierFromVertex<I, V, E, Iterator<I>> vertexNeighborsSupplier() {
     return new SupplierFromVertex<I, V, E, Iterator<I>>() {
       @Override
       public Iterator<I> get(final Vertex<I, V, E> vertex) {
         return new TargetVertexIdIterator<>(vertex);
       }
     };
   }
 
   /**
    * Supplier which extracts and returns Iterator over neighbor IDs
    * that return true for given predicate.
    * Note - iterator returns reused object, so you need to "use" them,
    * before calling next() again.
    */
   public static
   <I extends WritableComparable, V extends Writable, E extends Writable>
   SupplierFromVertex<I, V, E, Iterator<I>> vertexNeighborsSupplier(
       final Predicate<I> toSupply) {
     return new SupplierFromVertex<I, V, E, Iterator<I>>() {
       @Override
       public Iterator<I> get(final Vertex<I, V, E> vertex) {
         return Iterators.filter(
             new TargetVertexIdIterator<>(vertex),
             new com.google.common.base.Predicate<I>() {
               @Override
               public boolean apply(I input) {
                 return toSupply.apply(input);
               }
             });
       }
     };
   }
 
   /**
    * Supplier which gives Iterator over neighbor IDs that return true for given
    * predicate over (index, target)
    * Note - iterator returns reused object, so you need to "use" them,
    * before calling next() again.
    */
   public static
   <I extends WritableComparable, V extends Writable, E extends Writable>
   SupplierFromVertex<I, V, E, Iterator<I>> vertexNeighborsSupplierWithIndex(
       final PairPredicate<IntRef, I> toSupply) {
     return new SupplierFromVertex<I, V, E, Iterator<I>>() {
       @Override
       public Iterator<I> get(final Vertex<I, V, E> vertex) {
         // Every time we return an iterator, we return with a fresh (0) index.
         return filterWithIndex(
             new TargetVertexIdIterator<>(vertex), toSupply);
       }
     };
   }
 
   /**
    * Returns the elements of {@code unfiltered} that satisfy a
    * predicate over (index, t).
    */
   private static <T> UnmodifiableIterator<T> filterWithIndex(
       final Iterator<T> unfiltered, final PairPredicate<IntRef, T> predicate) {
     checkNotNull(unfiltered);
     checkNotNull(predicate);
     return new AbstractIterator<T>() {
       private final IntRef index = new IntRef(0);
       @Override protected T computeNext() {
         while (unfiltered.hasNext()) {
           T element = unfiltered.next();
           boolean res = predicate.apply(index, element);
           index.value++;
           if (res) {
             return element;
           }
         }
         return endOfData();
       }
     };
   }
 }