/*

  * 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.framework.piece.messages;

 import org.apache.giraph.combiner.MessageCombiner;

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

 import org.apache.giraph.conf.ImmutableClassesGiraphConfiguration;

 import org.apache.giraph.conf.MessageClasses;

 import org.apache.giraph.factories.MessageValueFactory;

 import org.apache.giraph.utils.ReflectionUtils;

 import org.apache.giraph.writable.kryo.KryoWritable;

 import org.apache.hadoop.io.Writable;

 import org.apache.hadoop.io.WritableComparable;

 import com.google.common.base.Preconditions;

 /**

  * MessageClasses implementation that provides factory and combiner instances

  * through a provided supplier.

  *

  * @param <I> Vertex id type

  * @param <M> Message type

  */

 public class ObjectMessageClasses<I extends WritableComparable,

     M extends Writable> extends KryoWritable implements MessageClasses<I, M> {

   private final Class<M> messageClass;

   private final SupplierFromConf<MessageValueFactory<M>>

   messageValueFactorySupplier;

   private final SupplierFromConf<? extends MessageCombiner<? super I, M>>

   messageCombinerSupplier;

   private final MessageEncodeAndStoreType messageEncodeAndStoreType;

   private final boolean ignoreExistingVertices;

   public ObjectMessageClasses() {

     this(null, null, null, null, false);

   }

   public ObjectMessageClasses(Class<M> messageClass,

       SupplierFromConf<MessageValueFactory<M>> messageValueFactorySupplier,

       SupplierFromConf<? extends MessageCombiner<? super I, M>>

         messageCombinerSupplier,

       MessageEncodeAndStoreType messageEncodeAndStoreType,

       boolean ignoreExistingVertices) {

     this.messageClass = messageClass;

     this.messageValueFactorySupplier = messageValueFactorySupplier;

     this.messageCombinerSupplier = messageCombinerSupplier;

     this.messageEncodeAndStoreType = messageEncodeAndStoreType;

     this.ignoreExistingVertices = ignoreExistingVertices;

   }

   @Override

   public Class<M> getMessageClass() {

     return messageClass;

   }

   @Override

   public MessageValueFactory<M> createMessageValueFactory(

       ImmutableClassesGiraphConfiguration conf) {

     return Preconditions.checkNotNull(messageValueFactorySupplier.apply(conf));

   }

   @Override

   public MessageCombiner<? super I, M> createMessageCombiner(

       ImmutableClassesGiraphConfiguration<I, ? extends Writable,

         ? extends Writable> conf) {

     return messageCombinerSupplier != null ?

       Preconditions.checkNotNull(messageCombinerSupplier.apply(conf)) : null;

   }

   @Override

   public boolean useMessageCombiner() {

     return messageCombinerSupplier != null;

   }

   @Override

   public boolean ignoreExistingVertices() {

     return ignoreExistingVertices;

   }

   @Override

   public MessageEncodeAndStoreType getMessageEncodeAndStoreType() {

     return messageEncodeAndStoreType;

   }

   @Override

   public MessageClasses<I, M> createCopyForNewSuperstep() {

     return new ObjectMessageClasses<>(

         messageClass, messageValueFactorySupplier,

         messageCombinerSupplier, messageEncodeAndStoreType,

         ignoreExistingVertices);

   }

   @Override

   public void verifyConsistent(ImmutableClassesGiraphConfiguration conf) {

     MessageValueFactory<M> messageValueFactory =

         messageValueFactorySupplier.apply(conf);

     Preconditions.checkState(

         messageValueFactory.newInstance().getClass().equals(messageClass));

     if (messageCombinerSupplier != null) {

       MessageCombiner<? super I, M> messageCombiner =

           messageCombinerSupplier.apply(conf);

       Preconditions.checkState(messageCombiner.createInitialMessage()

           .getClass().equals(messageClass));

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

           MessageCombiner.class, messageCombiner.getClass());

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

           "Vertex id", messageCombiner.getClass());

       ReflectionUtils.verifyTypes(messageClass, combinerTypes[1],

           "Outgoing message", messageCombiner.getClass());

     }

   }

 }