/*
    * 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.conf;
  
  import org.apache.giraph.aggregators.AggregatorWriter;
  import org.apache.giraph.aggregators.TextAggregatorWriter;
  import org.apache.giraph.comm.messages.MessageEncodeAndStoreType;
  import org.apache.giraph.edge.ByteArrayEdges;
  import org.apache.giraph.edge.OutEdges;
  import org.apache.giraph.factories.ComputationFactory;
  import org.apache.giraph.factories.DefaultComputationFactory;
  import org.apache.giraph.factories.DefaultMessageValueFactory;
  import org.apache.giraph.graph.Computation;
  import org.apache.giraph.graph.DefaultVertexResolver;
  import org.apache.giraph.graph.DefaultVertexValueCombiner;
  import org.apache.giraph.graph.Vertex;
  import org.apache.giraph.graph.VertexResolver;
  import org.apache.giraph.graph.VertexValueCombiner;
  import org.apache.giraph.io.EdgeInputFormat;
  import org.apache.giraph.io.EdgeOutputFormat;
  import org.apache.giraph.io.MappingInputFormat;
  import org.apache.giraph.io.VertexInputFormat;
  import org.apache.giraph.io.VertexOutputFormat;
  import org.apache.giraph.io.filters.DefaultEdgeInputFilter;
  import org.apache.giraph.io.filters.DefaultVertexInputFilter;
  import org.apache.giraph.io.filters.EdgeInputFilter;
  import org.apache.giraph.io.filters.VertexInputFilter;
  import org.apache.giraph.master.DefaultMasterCompute;
  import org.apache.giraph.master.MasterCompute;
  import org.apache.giraph.partition.GraphPartitionerFactory;
  import org.apache.giraph.partition.HashPartitionerFactory;
  import org.apache.giraph.partition.Partition;
  import org.apache.giraph.partition.SimplePartition;
  import org.apache.giraph.types.NoMessage;
  import org.apache.giraph.worker.DefaultWorkerContext;
  import org.apache.giraph.worker.WorkerContext;
  import org.apache.hadoop.conf.Configuration;
  import org.apache.hadoop.io.Writable;
  import org.apache.hadoop.io.WritableComparable;
  
  /**
   * Holder for classes used by Giraph.
   *
   * @param <I> Vertex ID class
   * @param <V> Vertex Value class
   * @param <E> Edge class
   */
  @SuppressWarnings("unchecked")
  public class GiraphClasses<I extends WritableComparable,
      V extends Writable, E extends Writable>
      implements GiraphConstants {
    /** ComputationFactory class - cached for fast access */
    protected Class<? extends ComputationFactory<I, V, E,
        ? extends Writable, ? extends Writable>>
    computationFactoryClass;
    /** Computation class - cached for fast access */
    protected Class<? extends Computation<I, V, E,
        ? extends Writable, ? extends Writable>>
    computationClass;
    /** Generic types used to describe graph */
    protected GiraphTypes<I, V, E> giraphTypes;
    /** Vertex edges class - cached for fast access */
    protected Class<? extends OutEdges<I, E>> outEdgesClass;
    /** Input vertex edges class - cached for fast access */
    protected Class<? extends OutEdges<I, E>> inputOutEdgesClass;
  
    /** Graph partitioner factory class - cached for fast access */
    protected Class<? extends GraphPartitionerFactory<I, V, E>>
    graphPartitionerFactoryClass;
  
    /** Vertex input format class - cached for fast access */
    protected Class<? extends VertexInputFormat<I, V, E>>
    vertexInputFormatClass;
    /** Vertex output format class - cached for fast access */
    protected Class<? extends VertexOutputFormat<I, V, E>>
    vertexOutputFormatClass;
    /** Mapping input format - cached for fast access */
    protected Class<? extends MappingInputFormat<I, V, E, ? extends Writable>>
    mappingInputFormatClass;
    /** Edge input format class - cached for fast access */
    protected Class<? extends EdgeInputFormat<I, E>>
    edgeInputFormatClass;
    /** Edge output format class - cached for fast access */
    protected Class<? extends EdgeOutputFormat<I, V, E>>
   edgeOutputFormatClass;
 
   /** Aggregator writer class - cached for fast access */
   protected Class<? extends AggregatorWriter> aggregatorWriterClass;
 
   /** Incoming message classes */
   protected MessageClasses<I, ? extends Writable> incomingMessageClasses;
   /** Outgoing message classes */
   protected MessageClasses<I, ? extends Writable> outgoingMessageClasses;
 
   /** Vertex resolver class - cached for fast access */
   protected Class<? extends VertexResolver<I, V, E>> vertexResolverClass;
   /** Vertex value combiner class - cached for fast access */
   protected Class<? extends VertexValueCombiner<V>> vertexValueCombinerClass;
   /** Worker context class - cached for fast access */
   protected Class<? extends WorkerContext> workerContextClass;
   /** Master compute class - cached for fast access */
   protected Class<? extends MasterCompute> masterComputeClass;
 
   /** Partition class - cached for fast accesss */
   protected Class<? extends Partition<I, V, E>> partitionClass;
 
   /** Edge Input Filter class */
   protected Class<? extends EdgeInputFilter<I, E>> edgeInputFilterClass;
   /** Vertex Input Filter class */
   protected Class<? extends VertexInputFilter<I, V, E>> vertexInputFilterClass;
 
   /**
    * Empty constructor. Initialize with default classes or null.
    */
   public GiraphClasses() {
     computationFactoryClass = (Class) DefaultComputationFactory.class;
     giraphTypes = new GiraphTypes<I, V, E>();
     outEdgesClass = (Class) ByteArrayEdges.class;
     inputOutEdgesClass = (Class) ByteArrayEdges.class;
     graphPartitionerFactoryClass = (Class) HashPartitionerFactory.class;
     aggregatorWriterClass = TextAggregatorWriter.class;
     vertexResolverClass = (Class) DefaultVertexResolver.class;
     vertexValueCombinerClass = (Class) DefaultVertexValueCombiner.class;
     workerContextClass = DefaultWorkerContext.class;
     masterComputeClass = DefaultMasterCompute.class;
     partitionClass = (Class) SimplePartition.class;
     edgeInputFilterClass = (Class) DefaultEdgeInputFilter.class;
     vertexInputFilterClass = (Class) DefaultVertexInputFilter.class;
   }
 
   /**
    * Constructor that reads classes from a Configuration object.
    *
    * @param conf Configuration object to read from.
    */
   public GiraphClasses(Configuration conf) {
     giraphTypes = GiraphTypes.readFrom(conf);
     computationFactoryClass =
         (Class<? extends ComputationFactory<I, V, E,
             ? extends Writable, ? extends Writable>>)
             COMPUTATION_FACTORY_CLASS.get(conf);
     computationClass =
         (Class<? extends Computation<I, V, E,
             ? extends Writable, ? extends Writable>>)
             COMPUTATION_CLASS.get(conf);
 
     outEdgesClass = (Class<? extends OutEdges<I, E>>)
         VERTEX_EDGES_CLASS.get(conf);
     inputOutEdgesClass = (Class<? extends OutEdges<I, E>>)
         INPUT_VERTEX_EDGES_CLASS.getWithDefault(conf, outEdgesClass);
 
     graphPartitionerFactoryClass =
         (Class<? extends GraphPartitionerFactory<I, V, E>>)
             GRAPH_PARTITIONER_FACTORY_CLASS.get(conf);
 
     vertexInputFormatClass = (Class<? extends VertexInputFormat<I, V, E>>)
         VERTEX_INPUT_FORMAT_CLASS.get(conf);
     vertexOutputFormatClass = (Class<? extends VertexOutputFormat<I, V, E>>)
         VERTEX_OUTPUT_FORMAT_CLASS.get(conf);
     edgeInputFormatClass = (Class<? extends EdgeInputFormat<I, E>>)
         EDGE_INPUT_FORMAT_CLASS.get(conf);
     edgeOutputFormatClass = (Class<? extends EdgeOutputFormat<I, V, E>>)
         EDGE_OUTPUT_FORMAT_CLASS.get(conf);
     mappingInputFormatClass = (Class<? extends MappingInputFormat<I, V, E,
         ? extends Writable>>)
         MAPPING_INPUT_FORMAT_CLASS.get(conf);
 
     aggregatorWriterClass = AGGREGATOR_WRITER_CLASS.get(conf);
 
     // incoming messages shouldn't be used in first iteration at all
     // but empty message stores are created, etc, so using NoMessage
     // to enforce not a single message is read/written
     incomingMessageClasses = new DefaultMessageClasses(
         NoMessage.class,
         DefaultMessageValueFactory.class,
         null,
         MessageEncodeAndStoreType.BYTEARRAY_PER_PARTITION);
     outgoingMessageClasses = new DefaultMessageClasses(
         giraphTypes.getInitialOutgoingMessageValueClass(),
         OUTGOING_MESSAGE_VALUE_FACTORY_CLASS.get(conf),
         MESSAGE_COMBINER_CLASS.get(conf),
         MESSAGE_ENCODE_AND_STORE_TYPE.get(conf));
 
     vertexResolverClass = (Class<? extends VertexResolver<I, V, E>>)
         VERTEX_RESOLVER_CLASS.get(conf);
     vertexValueCombinerClass = (Class<? extends VertexValueCombiner<V>>)
         VERTEX_VALUE_COMBINER_CLASS.get(conf);
     workerContextClass = WORKER_CONTEXT_CLASS.get(conf);
     masterComputeClass =  MASTER_COMPUTE_CLASS.get(conf);
     partitionClass = (Class<? extends Partition<I, V, E>>)
         PARTITION_CLASS.get(conf);
 
     edgeInputFilterClass = (Class<? extends EdgeInputFilter<I, E>>)
         EDGE_INPUT_FILTER_CLASS.get(conf);
     vertexInputFilterClass = (Class<? extends VertexInputFilter<I, V, E>>)
         VERTEX_INPUT_FILTER_CLASS.get(conf);
   }
 
   public Class<? extends ComputationFactory<I, V, E,
       ? extends Writable, ? extends Writable>> getComputationFactoryClass() {
     return computationFactoryClass;
   }
 
   /**
    * Get Computation class
    *
    * @return Computation class.
    */
   public Class<? extends Computation<I, V, E,
       ? extends Writable, ? extends Writable>>
   getComputationClass() {
     return computationClass;
   }
 
   public GiraphTypes<I, V, E> getGiraphTypes() {
     return giraphTypes;
   }
 
   /**
    * Get Vertex ID class
    *
    * @return Vertex ID class
    */
   public Class<I> getVertexIdClass() {
     return giraphTypes.getVertexIdClass();
   }
 
 
   /**
    * Get Vertex implementation class
    *
    * @return Vertex implementation class
    */
   public Class<? extends Vertex> getVertexClass() {
     return giraphTypes.getVertexClass();
   }
 
 
   /**
    * Get Vertex Value class
    *
    * @return Vertex Value class
    */
   public Class<V> getVertexValueClass() {
     return giraphTypes.getVertexValueClass();
   }
 
   /**
    * Get Edge Value class
    *
    * @return Edge Value class
    */
   public Class<E> getEdgeValueClass() {
     return giraphTypes.getEdgeValueClass();
   }
 
 
   public MessageClasses<? extends WritableComparable, ? extends Writable>
   getIncomingMessageClasses() {
     return incomingMessageClasses;
   }
 
   public MessageClasses<? extends WritableComparable, ? extends Writable>
   getOutgoingMessageClasses() {
     return outgoingMessageClasses;
   }
 
   /**
    * Get Vertex edges class
    *
    * @return Vertex edges class.
    */
   public Class<? extends OutEdges<I, E>> getOutEdgesClass() {
     return outEdgesClass;
   }
 
   /**
    * Get Vertex edges class used during edge-based input
    *
    * @return Vertex edges class.
    */
   public Class<? extends OutEdges<I, E>> getInputOutEdgesClass() {
     return inputOutEdgesClass;
   }
 
   /**
    * Get the GraphPartitionerFactory
    *
    * @return GraphPartitionerFactory
    */
   public Class<? extends GraphPartitionerFactory<I, V, E>>
   getGraphPartitionerFactoryClass() {
     return graphPartitionerFactoryClass;
   }
 
   public Class<? extends EdgeInputFilter<I, E>>
   getEdgeInputFilterClass() {
     return edgeInputFilterClass;
   }
 
   public Class<? extends VertexInputFilter<I, V, E>>
   getVertexInputFilterClass() {
     return vertexInputFilterClass;
   }
 
   /**
    * Check if VertexInputFormat class is set
    *
    * @return true if VertexInputFormat class is set
    */
   public boolean hasVertexInputFormat() {
     return vertexInputFormatClass != null;
   }
 
   /**
    * Get VertexInputFormat held
    *
    * @return VertexInputFormat
    */
   public Class<? extends VertexInputFormat<I, V, E>>
   getVertexInputFormatClass() {
     return vertexInputFormatClass;
   }
 
   /**
    * Check if VertexOutputFormat is set
    *
    * @return true if VertexOutputFormat is set
    */
   public boolean hasVertexOutputFormat() {
     return vertexOutputFormatClass != null;
   }
 
   /**
    * Check if MappingInputFormat is set
    *
    * @return true if MappingInputFormat is set
    */
   public boolean hasMappingInputFormat() {
     return mappingInputFormatClass != null;
   }
 
   /**
    * Get VertexOutputFormat set
    *
    * @return VertexOutputFormat
    */
   public Class<? extends VertexOutputFormat<I, V, E>>
   getVertexOutputFormatClass() {
     return vertexOutputFormatClass;
   }
 
   public Class<? extends MappingInputFormat<I, V, E, ? extends Writable>>
   getMappingInputFormatClass() {
     return mappingInputFormatClass;
   }
 
   /**
    * Check if EdgeInputFormat is set
    *
    * @return true if EdgeInputFormat is set
    */
   public boolean hasEdgeInputFormat() {
     return edgeInputFormatClass != null;
   }
 
   /**
    * Get EdgeInputFormat used
    *
    * @return EdgeInputFormat
    */
   public Class<? extends EdgeInputFormat<I, E>> getEdgeInputFormatClass() {
     return edgeInputFormatClass;
   }
 
   /**
    * Check if EdgeOutputFormat is set
    *
    * @return true if EdgeOutputFormat is set
    */
   public boolean hasEdgeOutputFormat() {
     return edgeOutputFormatClass != null;
   }
 
   /**
    * Get VertexOutputFormat set
    *
    * @return VertexOutputFormat
    */
   public Class<? extends EdgeOutputFormat<I, V, E>>
   getEdgeOutputFormatClass() {
     return edgeOutputFormatClass;
   }
 
   /**
    * Check if AggregatorWriter is set
    *
    * @return true if AggregatorWriter is set
    */
   public boolean hasAggregatorWriterClass() {
     return aggregatorWriterClass != null;
   }
 
   /**
    * Get AggregatorWriter used
    *
    * @return AggregatorWriter
    */
   public Class<? extends AggregatorWriter> getAggregatorWriterClass() {
     return aggregatorWriterClass;
   }
 
   /**
    * Check if VertexResolver is set
    *
    * @return true if VertexResolver is set
    */
   public boolean hasVertexResolverClass() {
     return vertexResolverClass != null;
   }
 
   /**
    * Get VertexResolver used
    *
    * @return VertexResolver
    */
   public Class<? extends VertexResolver<I, V, E>> getVertexResolverClass() {
     return vertexResolverClass;
   }
 
   /**
    * Get VertexValueCombiner used
    *
    * @return VertexValueCombiner
    */
   public Class<? extends VertexValueCombiner<V>> getVertexValueCombinerClass() {
     return vertexValueCombinerClass;
   }
 
   /**
    * Check if WorkerContext is set
    *
    * @return true if WorkerContext is set
    */
   public boolean hasWorkerContextClass() {
     return workerContextClass != null;
   }
 
   /**
    * Get WorkerContext used
    *
    * @return WorkerContext
    */
   public Class<? extends WorkerContext> getWorkerContextClass() {
     return workerContextClass;
   }
 
   /**
    * Check if MasterCompute is set
    *
    * @return true MasterCompute is set
    */
   public boolean hasMasterComputeClass() {
     return masterComputeClass != null;
   }
 
   /**
    * Get MasterCompute used
    *
    * @return MasterCompute
    */
   public Class<? extends MasterCompute> getMasterComputeClass() {
     return masterComputeClass;
   }
 
   /**
    * Check if Partition is set
    *
    * @return true if Partition is set
    */
   public boolean hasPartitionClass() {
     return partitionClass != null;
   }
 
   /**
    * Get Partition
    *
    * @return Partition
    */
   public Class<? extends Partition<I, V, E>> getPartitionClass() {
     return partitionClass;
   }
 
   /**
    * Set Computation class held, and update message types
    *
    * @param computationClass Computation class to set
    * @return this
    */
   public GiraphClasses setComputationClass(Class<? extends
       Computation<I, V, E, ? extends Writable, ? extends Writable>>
       computationClass) {
     this.computationClass = computationClass;
     return this;
   }
 
   /**
    * Set Vertex ID class held
    *
    * @param vertexIdClass Vertex ID to set
    * @return this
    */
   public GiraphClasses setVertexIdClass(Class<I> vertexIdClass) {
     giraphTypes.setVertexIdClass(vertexIdClass);
     return this;
   }
 
   /**
    * Set Vertex Value class held
    *
    * @param vertexValueClass Vertex Value class to set
    * @return this
    */
   public GiraphClasses setVertexValueClass(Class<V> vertexValueClass) {
     giraphTypes.setVertexValueClass(vertexValueClass);
     return this;
   }
 
   /**
    * Set Edge Value class held
    *
    * @param edgeValueClass Edge Value class to set
    * @return this
    */
   public GiraphClasses setEdgeValueClass(Class<E> edgeValueClass) {
     giraphTypes.setEdgeValueClass(edgeValueClass);
     return this;
   }
 
   /**
    * Set incoming Message Value class held - messages which have been sent in
    * the previous superstep and are processed in the current one
    *
    * @param incomingMessageClasses Message classes value to set
    * @return this
    */
   public GiraphClasses setIncomingMessageClasses(
       MessageClasses<I, ? extends Writable> incomingMessageClasses) {
     this.incomingMessageClasses = incomingMessageClasses;
     return this;
   }
 
   /**
    * Set outgoing Message Value class held - messages which are going to be sent
    * during current superstep
    *
    * @param outgoingMessageClasses Message classes value to set
    * @return this
    */
   public GiraphClasses setOutgoingMessageClasses(
       MessageClasses<I, ? extends Writable> outgoingMessageClasses) {
     this.outgoingMessageClasses = outgoingMessageClasses;
     return this;
   }
 
   /**
    * Set OutEdges class held
    *
    * @param outEdgesClass Vertex edges class to set
    * @return this
    */
   public GiraphClasses setOutEdgesClass(
       Class<? extends OutEdges> outEdgesClass) {
     this.outEdgesClass =
         (Class<? extends OutEdges<I, E>>) outEdgesClass;
     return this;
   }
 
   /**
    * Set OutEdges class used during edge-input (if different from the one
    * used for computation)
    *
    * @param inputOutEdgesClass Input vertex edges class to set
    * @return this
    */
   public GiraphClasses setInputOutEdgesClass(
       Class<? extends OutEdges> inputOutEdgesClass) {
     this.inputOutEdgesClass =
         (Class<? extends OutEdges<I, E>>) inputOutEdgesClass;
     return this;
   }
 
   /**
    * Set GraphPartitionerFactory class held
    *
    * @param klass GraphPartitionerFactory to set
    * @return this
    */
   public GiraphClasses setGraphPartitionerFactoryClass(
       Class<? extends GraphPartitionerFactory<I, V, E>> klass) {
     this.graphPartitionerFactoryClass = klass;
     return this;
   }
 
   /**
    * Set MappingInputFormat held
    *
    * @param mappingInputFormatClass MappingInputFormat to set
    * @return this
    */
   public GiraphClasses setMappingInputFormatClass(
     Class<? extends MappingInputFormat<I, V, E, Writable>>
       mappingInputFormatClass) {
     this.mappingInputFormatClass = mappingInputFormatClass;
     return this;
   }
 
   /**
    * Set VertexInputFormat held
    *
    * @param vertexInputFormatClass VertexInputFormat to set
    * @return this
    */
   public GiraphClasses setVertexInputFormatClass(
       Class<? extends VertexInputFormat<I, V, E>> vertexInputFormatClass) {
     this.vertexInputFormatClass = vertexInputFormatClass;
     return this;
   }
 
   /**
    * Set VertexOutputFormat held
    *
    * @param vertexOutputFormatClass VertexOutputFormat to set
    * @return this
    */
   public GiraphClasses setVertexOutputFormatClass(
       Class<? extends VertexOutputFormat<I, V, E>> vertexOutputFormatClass) {
     this.vertexOutputFormatClass = vertexOutputFormatClass;
     return this;
   }
 
   /**
    * Set EdgeInputFormat class held
    *
    * @param edgeInputFormatClass EdgeInputFormat to set
    * @return this
    */
   public GiraphClasses setEdgeInputFormatClass(
       Class<? extends EdgeInputFormat<I, E>> edgeInputFormatClass) {
     this.edgeInputFormatClass = edgeInputFormatClass;
     return this;
   }
 
   /**
    * Set AggregatorWriter class used
    *
    * @param aggregatorWriterClass AggregatorWriter to set
    * @return this
    */
   public GiraphClasses setAggregatorWriterClass(
       Class<? extends AggregatorWriter> aggregatorWriterClass) {
     this.aggregatorWriterClass = aggregatorWriterClass;
     return this;
   }
 
   /**
    * Set VertexResolver used
    *
    * @param vertexResolverClass VertexResolver to set
    * @return this
    */
   public GiraphClasses setVertexResolverClass(
       Class<? extends VertexResolver<I, V, E>> vertexResolverClass) {
     this.vertexResolverClass = vertexResolverClass;
     return this;
   }
 
   /**
    * Set WorkerContext used
    *
    * @param workerContextClass WorkerContext class to set
    * @return this
    */
   public GiraphClasses setWorkerContextClass(
       Class<? extends WorkerContext> workerContextClass) {
     this.workerContextClass = workerContextClass;
     return this;
   }
 
   /**
    * Set MasterCompute class used
    *
    * @param masterComputeClass MasterCompute class to set
    * @return this
    */
   public GiraphClasses setMasterComputeClass(
       Class<? extends MasterCompute> masterComputeClass) {
     this.masterComputeClass = masterComputeClass;
     return this;
   }
 
   /**
    * Set Partition class to use
    *
    * @param partitionClass Partition class to set
    * @return this
    */
   public GiraphClasses setPartitionClass(
       Class<? extends Partition<I, V, E>> partitionClass) {
     this.partitionClass = partitionClass;
     return this;
   }
 }