/*
    * 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.migration;
  
  import java.io.DataInput;
  import java.io.DataOutput;
  import java.io.IOException;
  
  import org.apache.giraph.aggregators.Aggregator;
  import org.apache.giraph.block_app.framework.api.BlockMasterApi;
  import org.apache.giraph.block_app.framework.api.StatusReporter;
  import org.apache.giraph.block_app.migration.MigrationAbstractComputation.MigrationFullAbstractComputation;
  import org.apache.giraph.combiner.MessageCombiner;
  import org.apache.giraph.conf.DefaultImmutableClassesGiraphConfigurable;
  import org.apache.giraph.conf.TypesHolder;
  import org.apache.giraph.reducers.ReduceOperation;
  import org.apache.giraph.utils.ReflectionUtils;
  import org.apache.hadoop.io.Writable;
  
  /**
   * Replacement for MasterCompute when migrating to Blocks Framework,
   * disallowing functions that are tied to execution order.
   */
  @SuppressWarnings({"unchecked", "rawtypes"})
  public abstract class MigrationMasterCompute
      extends DefaultImmutableClassesGiraphConfigurable implements Writable {
    private BlockMasterApi api;
  
    final void init(BlockMasterApi masterApi) {
      this.api = masterApi;
      setConf(masterApi.getConf());
    }
  
    @Override
    public void readFields(DataInput in) throws IOException {
    }
  
    @Override
    public void write(DataOutput out) throws IOException {
    }
  
    public void compute() {
    }
  
    public void initialize() throws InstantiationException,
        IllegalAccessException {
    }
  
    @SuppressWarnings("deprecation")
    public long getTotalNumVertices() {
      return api.getTotalNumVertices();
    }
  
    @SuppressWarnings("deprecation")
    public long getTotalNumEdges() {
      return api.getTotalNumEdges();
    }
  
    public final <S, R extends Writable> void registerReducer(
        String name, ReduceOperation<S, R> reduceOp) {
      api.registerReducer(name, reduceOp);
    }
  
    public final <S, R extends Writable> void registerReducer(
        String name, ReduceOperation<S, R> reduceOp, R globalInitialValue) {
      api.registerReducer(
          name, reduceOp, globalInitialValue);
    }
  
    public final <T extends Writable> T getReduced(String name) {
      return api.getReduced(name);
    }
  
    public final void broadcast(String name, Writable object) {
      api.broadcast(name, object);
    }
  
    public final <A extends Writable> boolean registerAggregator(
      String name, Class<? extends Aggregator<A>> aggregatorClass)
      throws InstantiationException, IllegalAccessException {
      return api.registerAggregator(
          name, aggregatorClass);
    }
  
   @SuppressWarnings("deprecation")
   public final <A extends Writable> boolean registerPersistentAggregator(
       String name,
       Class<? extends Aggregator<A>> aggregatorClass) throws
       InstantiationException, IllegalAccessException {
     return api.registerPersistentAggregator(name, aggregatorClass);
   }
 
   public final <A extends Writable> A getAggregatedValue(String name) {
     return api.<A>getAggregatedValue(name);
   }
 
   public final <A extends Writable> void setAggregatedValue(
       String name, A value) {
     api.setAggregatedValue(name, value);
   }
 
   public final void logToCommandLine(String line) {
     api.logToCommandLine(line);
   }
 
   public final StatusReporter getContext() {
     return api;
   }
 
   /**
    * Drop-in replacement for MasterCompute when migrating
    * to Blocks Framework.
    */
   public static class MigrationFullMasterCompute
       extends MigrationMasterCompute {
     private long superstep;
     private boolean halt;
     private Class<? extends MigrationAbstractComputation> computationClass;
     private Class<? extends MigrationAbstractComputation> newComputationClass;
     private Class<? extends Writable> originalMessage;
     private Class<? extends Writable> newMessage;
     private Class<? extends MessageCombiner> originalMessageCombiner;
     private Class<? extends MessageCombiner> newMessageCombiner;
 
     final void init(
         long superstep,
         Class<? extends MigrationAbstractComputation> computationClass,
         Class<? extends Writable> message,
         Class<? extends MessageCombiner> messageCombiner) {
       this.superstep = superstep;
       this.halt = false;
       this.computationClass = computationClass;
       this.newComputationClass = null;
       this.originalMessage = message;
       this.newMessage = null;
       this.originalMessageCombiner = messageCombiner;
       this.newMessageCombiner = null;
     }
 
     public final long getSuperstep() {
       return superstep;
     }
 
     @Override
     public final long getTotalNumVertices() {
       if (superstep == 0) {
         throw new RuntimeException(
             "getTotalNumVertices not available in superstep=0");
       }
       return super.getTotalNumVertices();
     }
 
     @Override
     public final long getTotalNumEdges() {
       if (superstep == 0) {
         throw new RuntimeException(
             "getTotalNumEdges not available in superstep=0");
       }
       return super.getTotalNumEdges();
     }
 
 
     public final void haltComputation() {
       halt = true;
     }
 
     public final boolean isHalted() {
       return halt;
     }
 
     public final void setComputation(
         Class<? extends MigrationFullAbstractComputation> computation) {
       if (computation != null) {
         newComputationClass = computation;
       } else {
         // TODO
         this.computationClass = null;
       }
     }
 
     public final
     Class<? extends MigrationAbstractComputation> getComputation() {
       if (newComputationClass != null) {
         return newComputationClass;
       }
       if (computationClass != null) {
         return computationClass;
       }
       return null;
     }
 
     public final void setMessageCombiner(
         Class<? extends MessageCombiner> combinerClass) {
       this.newMessageCombiner = combinerClass;
     }
 
     public final Class<? extends MessageCombiner> getMessageCombiner() {
       return newMessageCombiner != null ?
         newMessageCombiner : originalMessageCombiner;
     }
 
     public final void setIncomingMessage(
         Class<? extends Writable> incomingMessageClass) {
       if (!originalMessage.equals(incomingMessageClass)) {
         throw new IllegalArgumentException(
             originalMessage + " and " + incomingMessageClass + " must be same");
       }
     }
 
     public final void setOutgoingMessage(
         Class<? extends Writable> outgoingMessageClass) {
       newMessage = outgoingMessageClass;
     }
 
     final Class<? extends Writable> getOutgoingMessage() {
       if (newMessage != null) {
         return newMessage;
       }
 
       if (newComputationClass == null) {
         return originalMessage;
       }
       Class[] computationTypes = ReflectionUtils.getTypeArguments(
           TypesHolder.class, newComputationClass);
       return computationTypes[4];
     }
 
     final Class<? extends MigrationAbstractComputation> getComputationClass() {
       return newComputationClass != null ?
         newComputationClass : computationClass;
     }
 
     final
     Class<? extends MigrationAbstractComputation> getNewComputationClass() {
       return newComputationClass;
     }
 
     final Class<? extends Writable> getNewMessage() {
       return newMessage;
     }
 
     final Class<? extends MessageCombiner> getNewMessageCombiner() {
       return newMessageCombiner;
     }
   }
 }