/*

  * 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 java.io.DataInput;

 import java.io.DataOutput;

 import java.io.IOException;

 import org.apache.giraph.combiner.MessageCombiner;

 import org.apache.giraph.comm.messages.MessageEncodeAndStoreType;

 import org.apache.giraph.factories.DefaultMessageValueFactory;

 import org.apache.giraph.factories.MessageValueFactory;

 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 com.google.common.base.Preconditions;

 /**

  * Default implementation of MessageClasses

  *

  * @param <I> Vertex id type

  * @param <M> Message type

  */

 public class DefaultMessageClasses

     <I extends WritableComparable, M extends Writable>

     implements MessageClasses<I, M> {

   /** message class */

   private Class<M> messageClass;

   /** message value factory class */

   private Class<? extends MessageValueFactory<M>>

   messageValueFactoryClass;

   /** message combiner class */

   private Class<? extends MessageCombiner<? super I, M>> messageCombinerClass;

   /** whether message class was manually modified in this superstep */

   private boolean messageClassModified;

   /** message encode and store type */

   private MessageEncodeAndStoreType messageEncodeAndStoreType;

   /** Constructor */

   public DefaultMessageClasses() {

   }

   /**

    * Constructor

    * @param messageClass message class

    * @param messageValueFactoryClass message value factory class

    * @param messageCombinerClass message combiner class

    * @param messageEncodeAndStoreType message encode and store type

    */

   public DefaultMessageClasses(

       Class<M> messageClass,

       Class<? extends MessageValueFactory<M>> messageValueFactoryClass,

       Class<? extends MessageCombiner<? super I, M>> messageCombinerClass,

         MessageEncodeAndStoreType messageEncodeAndStoreType) {

     this.messageClass = messageClass;

     this.messageValueFactoryClass = messageValueFactoryClass;

     this.messageCombinerClass = messageCombinerClass;

     this.messageClassModified = false;

     this.messageEncodeAndStoreType = messageEncodeAndStoreType;

   }

   @Override

   public Class<M> getMessageClass() {

     return messageClass;

   }

   @Override

   public MessageValueFactory<M> createMessageValueFactory(

       ImmutableClassesGiraphConfiguration conf) {

     if (messageValueFactoryClass.equals(DefaultMessageValueFactory.class)) {

       return new DefaultMessageValueFactory<>(messageClass, conf);

     }

     MessageValueFactory factory = ReflectionUtils.newInstance(

         messageValueFactoryClass, conf);

     if (!factory.newInstance().getClass().equals(messageClass)) {

       throw new IllegalStateException("Message factory " +

         messageValueFactoryClass + " creates " +

         factory.newInstance().getClass().getName() + ", but message type is " +

         messageClass.getName());

     }

     return factory;

   }

   @Override

   public MessageCombiner<? super I, M> createMessageCombiner(

       ImmutableClassesGiraphConfiguration conf) {

     if (messageCombinerClass == null) {

       return null;

     }

     MessageCombiner combiner =

         ReflectionUtils.newInstance(messageCombinerClass, conf);

     if (combiner != null) {

       Preconditions.checkState(

           combiner.createInitialMessage().getClass().equals(messageClass));

     }

     return combiner;

   }

   @Override

   public boolean useMessageCombiner() {

     return messageCombinerClass != null;

   }

   @Override

   public boolean ignoreExistingVertices() {

     return false;

   }

   @Override

   public MessageEncodeAndStoreType getMessageEncodeAndStoreType() {

     return messageEncodeAndStoreType;

   }

   @Override

   public MessageClasses<I, M> createCopyForNewSuperstep() {

     return new DefaultMessageClasses<>(messageClass, messageValueFactoryClass,

         messageCombinerClass, messageEncodeAndStoreType);

   }

   @Override

   public void verifyConsistent(

       ImmutableClassesGiraphConfiguration conf) {

     Class<?>[] factoryTypes = ReflectionUtils.getTypeArguments(

         MessageValueFactory.class, messageValueFactoryClass);

     ReflectionUtils.verifyTypes(messageClass, factoryTypes[0],

         "Message factory", messageValueFactoryClass);

     if (messageCombinerClass != null) {

       Class<?>[] combinerTypes = ReflectionUtils.getTypeArguments(

           MessageCombiner.class, messageCombinerClass);

       ReflectionUtils.verifyTypes(conf.getVertexIdClass(), combinerTypes[0],

           "Vertex id", messageCombinerClass);

       ReflectionUtils.verifyTypes(messageClass, combinerTypes[1],

           "Outgoing message", messageCombinerClass);

     }

   }

   /**

    * Set message class

    * @param messageClass message classs

    */

   public void setMessageClass(Class<M> messageClass) {

     this.messageClassModified = true;

     this.messageClass = messageClass;

   }

   /**

    * Set message class if not set already in this superstep

    * @param messageClass message class

    */

   public void setIfNotModifiedMessageClass(Class<M> messageClass) {

     if (!messageClassModified) {

       this.messageClass = messageClass;

     }

   }

   public void setMessageCombinerClass(

       Class<? extends MessageCombiner<? super I, M>> messageCombinerClass) {

     this.messageCombinerClass = messageCombinerClass;

   }

   public void setMessageValueFactoryClass(

       Class<? extends MessageValueFactory<M>> messageValueFactoryClass) {

     this.messageValueFactoryClass = messageValueFactoryClass;

   }

   public void setMessageEncodeAndStoreType(

       MessageEncodeAndStoreType messageEncodeAndStoreType) {

     this.messageEncodeAndStoreType = messageEncodeAndStoreType;

   }

   @Override

   public void write(DataOutput out) throws IOException {

     WritableUtils.writeClass(messageClass, out);

     WritableUtils.writeClass(messageValueFactoryClass, out);

     WritableUtils.writeClass(messageCombinerClass, out);

     out.writeBoolean(messageClassModified);

     out.writeByte(messageEncodeAndStoreType.ordinal());

   }

   @Override

   public void readFields(DataInput in) throws IOException {

     messageClass = WritableUtils.readClass(in);

     messageValueFactoryClass = WritableUtils.readClass(in);

     messageCombinerClass = WritableUtils.readClass(in);

     messageClassModified = in.readBoolean();

     messageEncodeAndStoreType =

         messageEncodeAndStoreType.values()[in.readByte()];

   }

 }