/*
    * 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.master;
  
  import static org.apache.giraph.conf.GiraphConstants.COMPUTATION_LANGUAGE;
  
  import java.io.DataInput;
  import java.io.DataOutput;
  import java.io.IOException;
  
  import org.apache.giraph.combiner.MessageCombiner;
  import org.apache.giraph.conf.DefaultMessageClasses;
  import org.apache.giraph.conf.GiraphClasses;
  import org.apache.giraph.conf.ImmutableClassesGiraphConfiguration;
  import org.apache.giraph.conf.MessageClasses;
  import org.apache.giraph.conf.TypesHolder;
  import org.apache.giraph.graph.Computation;
  import org.apache.giraph.graph.Language;
  import org.apache.giraph.utils.ReflectionUtils;
  import org.apache.giraph.utils.WritableUtils;
  import org.apache.hadoop.io.Writable;
  import org.apache.hadoop.io.WritableComparable;
  import org.apache.log4j.Logger;
  import com.google.common.base.Preconditions;
  
  /**
   * Holds Computation and MessageCombiner class.
   */
  public class SuperstepClasses implements Writable {
    /** Class logger */
    private static final Logger LOG = Logger.getLogger(SuperstepClasses.class);
    /** Configuration */
    private final ImmutableClassesGiraphConfiguration conf;
  
    /** Computation class to be used in the following superstep */
    private Class<? extends Computation> computationClass;
    /** Incoming message classes, immutable, only here for cheecking */
    private MessageClasses<? extends WritableComparable, ? extends Writable>
    incomingMessageClasses;
    /** Outgoing message classes */
    private MessageClasses<? extends WritableComparable, ? extends Writable>
    outgoingMessageClasses;
  
    /**
     * Constructor
     * @param conf Configuration
     * @param computationClass computation class
     * @param incomingMessageClasses incoming message classes
     * @param outgoingMessageClasses outgoing message classes
     */
    public SuperstepClasses(
        ImmutableClassesGiraphConfiguration conf,
        Class<? extends Computation> computationClass,
        MessageClasses<? extends WritableComparable, ? extends Writable>
          incomingMessageClasses,
        MessageClasses<? extends WritableComparable, ? extends Writable>
          outgoingMessageClasses) {
      this.conf = conf;
      this.computationClass = computationClass;
      this.incomingMessageClasses = incomingMessageClasses;
      this.outgoingMessageClasses = outgoingMessageClasses;
    }
  
    /**
     * Create empty superstep classes, readFields needs to be called afterwards
     * @param conf Configuration
     * @return Superstep classes
     */
    public static SuperstepClasses createToRead(
        ImmutableClassesGiraphConfiguration conf) {
      return new SuperstepClasses(conf, null, null, null);
    }
  
    /**
     * Create superstep classes by initiazling from current state
     * in configuration
     * @param conf Configuration
     * @return Superstep classes
     */
    public static SuperstepClasses createAndExtractTypes(
        ImmutableClassesGiraphConfiguration conf) {
      return new SuperstepClasses(
          conf,
         conf.getComputationClass(),
         conf.getOutgoingMessageClasses(),
         conf.getOutgoingMessageClasses().createCopyForNewSuperstep());
   }
 
   public Class<? extends Computation> getComputationClass() {
     return computationClass;
   }
 
   public MessageClasses<? extends WritableComparable, ? extends Writable>
   getOutgoingMessageClasses() {
     return outgoingMessageClasses;
   }
 
   /**
    * Set's outgoing MessageClasses for next superstep.
    * Should not be used together with
    * setMessageCombinerClass/setOutgoingMessageClass methods.
    *
    * @param outgoingMessageClasses outgoing message classes
    */
   public void setOutgoingMessageClasses(
       MessageClasses<? extends WritableComparable, ? extends Writable>
         outgoingMessageClasses) {
     this.outgoingMessageClasses = outgoingMessageClasses;
   }
 
   /**
    * Set computation class
    * @param computationClass computation class
    */
   public void setComputationClass(
       Class<? extends Computation> computationClass) {
     this.computationClass = computationClass;
 
     if (computationClass != null) {
       Class[] computationTypes = ReflectionUtils.getTypeArguments(
           TypesHolder.class, computationClass);
       if (computationTypes[4] != null &&
           outgoingMessageClasses instanceof DefaultMessageClasses) {
         ((DefaultMessageClasses) outgoingMessageClasses)
           .setIfNotModifiedMessageClass(computationTypes[4]);
       }
     }
   }
 
   /**
    * Set message combiner class.
    * Should not be used together setOutgoingMessageClasses
    * (throws exception if called with it),
    * as it is unnecessary to do so.
    *
    * @param messageCombinerClass message combiner class
    */
   public void setMessageCombinerClass(
       Class<? extends MessageCombiner> messageCombinerClass) {
     Preconditions.checkState(
         outgoingMessageClasses instanceof DefaultMessageClasses);
     ((DefaultMessageClasses) outgoingMessageClasses).
         setMessageCombinerClass(messageCombinerClass);
   }
 
   /**
    * Set incoming message class
    * @param incomingMessageClass incoming message class
    */
   @Deprecated
   public void setIncomingMessageClass(
       Class<? extends Writable> incomingMessageClass) {
     if (!incomingMessageClasses.getMessageClass().
         equals(incomingMessageClass)) {
       throw new IllegalArgumentException(
           "Cannot change incoming message class from " +
           incomingMessageClasses.getMessageClass() +
           " previously, to " + incomingMessageClass);
     }
   }
 
   /**
    * Set outgoing message class.
    * Should not be used together setOutgoingMessageClasses
    * (throws exception if called with it),
    * as it is unnecessary to do so.
    *
    * @param outgoingMessageClass outgoing message class
    */
   public void setOutgoingMessageClass(
       Class<? extends Writable> outgoingMessageClass) {
     Preconditions.checkState(
         outgoingMessageClasses instanceof DefaultMessageClasses);
     ((DefaultMessageClasses) outgoingMessageClasses).
         setMessageClass(outgoingMessageClass);
   }
 
   /**
    * Get message combiner class
    * @return message combiner class
    */
   public Class<? extends MessageCombiner> getMessageCombinerClass() {
     MessageCombiner combiner =
         outgoingMessageClasses.createMessageCombiner(conf);
     return combiner != null ? combiner.getClass() : null;
   }
 
   /**
    * Verify that types of current Computation and MessageCombiner are valid.
    * If types don't match an {@link IllegalStateException} will be thrown.
    *
    * @param checkMatchingMesssageTypes Check that the incoming/outgoing
    *                                   message types match
    */
   public void verifyTypesMatch(boolean checkMatchingMesssageTypes) {
     // In some cases, for example when using Jython, the Computation class may
     // not be set. This is because it is created by a ComputationFactory
     // dynamically and not known ahead of time. In this case there is nothing to
     // verify here so we bail.
     if (COMPUTATION_LANGUAGE.get(conf) == Language.JYTHON) {
       return;
     }
 
     Class<?>[] computationTypes = ReflectionUtils.getTypeArguments(
         TypesHolder.class, computationClass);
     ReflectionUtils.verifyTypes(conf.getVertexIdClass(), computationTypes[0],
         "Vertex id", computationClass);
     ReflectionUtils.verifyTypes(conf.getVertexValueClass(), computationTypes[1],
         "Vertex value", computationClass);
     ReflectionUtils.verifyTypes(conf.getEdgeValueClass(), computationTypes[2],
         "Edge value", computationClass);
 
     if (checkMatchingMesssageTypes) {
       ReflectionUtils.verifyTypes(incomingMessageClasses.getMessageClass(),
           computationTypes[3], "Incoming message type", computationClass);
     }
 
     ReflectionUtils.verifyTypes(outgoingMessageClasses.getMessageClass(),
         computationTypes[4], "Outgoing message type", computationClass);
 
     outgoingMessageClasses.verifyConsistent(conf);
   }
 
   /**
    * Update GiraphClasses with updated types
    * @param classes Giraph classes
    */
   public void updateGiraphClasses(GiraphClasses classes) {
     classes.setComputationClass(computationClass);
     classes.setIncomingMessageClasses(incomingMessageClasses);
     classes.setOutgoingMessageClasses(outgoingMessageClasses);
   }
 
   @Override
   public void write(DataOutput output) throws IOException {
     WritableUtils.writeClass(computationClass, output);
     WritableUtils.writeWritableObject(incomingMessageClasses, output);
     WritableUtils.writeWritableObject(outgoingMessageClasses, output);
   }
 
   @Override
   public void readFields(DataInput input) throws IOException {
     computationClass = WritableUtils.readClass(input);
     incomingMessageClasses = WritableUtils.readWritableObject(input, conf);
     outgoingMessageClasses = WritableUtils.readWritableObject(input, conf);
   }
 
   @Override
   public String toString() {
     String computationName = computationClass == null ? "_not_set_" :
         computationClass.getName();
     return "(computation=" + computationName +
         ",incoming=" + incomingMessageClasses +
         ",outgoing=" + outgoingMessageClasses + ")";
   }
 
 }