// Copyright 2011 Google Inc. All Rights Reserved.

package com.google.appengine.tools.mapreduce;

import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;

import com.google.appengine.tools.mapreduce.impl.MapReduceConstants;
import com.google.common.base.Optional;

/**
 * Specification for a MapReduce job. The values here affect what computation is performed in the
 * MapReduce and its side-effects, but not how it is executed; see {@link MapReduceSettings} for
 * that.
 *
 * @author ohler@google.com (Christian Ohler)
 *
 * @param <I> type of input values
 * @param <K> type of intermediate keys
 * @param <V> type of intermediate values
 * @param <O> type of output values
 * @param <R> type of result returned by the {@link Output}
 */
public final class MapReduceSpecification<I, K, V, O, R> extends BaseSpecification<I, O, R> {

  private static final long serialVersionUID = 4773121986957455423L;

  public static class Builder<I, K, V, O, R> extends
      BaseSpecification.Builder<Builder<I, K, V, O, R>, I, O, R> {

    private Mapper<? extends I, K, V> mapper;
    private Marshaller<K> keyMarshaller;
    private Marshaller<V> valueMarshaller;
    private Reducer<K, V, ? extends O> reducer;
    private Integer numReducers;

    public Builder() {
      jobName = "MapReduceJob";
    }

    public Builder(Input<I> input, Mapper<I, K, V> mapper, Reducer<K, V, O> reducer,
        Output<O, R> output) {
      this();
      this.input = input;
      this.mapper = mapper;
      this.reducer = reducer;
      this.output = output;
    }

    @Override
    protected Builder<I, K, V, O, R> self() {
      return this;
    }

    /**
     * @param mapper processes the input and optionally generates intermediate key-value pairs to be
     *        processed by {@code reducer}
     */
    public Builder<I, K, V, O, R> setMapper(Mapper<? extends I, K, V> mapper) {
      this.mapper = mapper;
      return this;
    }

    /**
     * @param marshaller turns intermediate keys into bytes and back. see {@link Marshallers} for
     *        default implementations.
     */
    public Builder<I, K, V, O, R> setKeyMarshaller(Marshaller<K> marshaller) {
      keyMarshaller = marshaller;
      return this;
    }

    /**
     * @param marshaller turns intermediate values into bytes and back. see {@link Marshallers} for
     *        default implementations. defaults to {@link Marshallers#getSerializationMarshaller}.
     */
    public Builder<I, K, V, O, R> setValueMarshaller(Marshaller<V> marshaller) {
      valueMarshaller = marshaller;
      return this;
    }

    /**
     * @param reducer processes the intermediate key-value pairs generated by the {@code mapper} and
     *        optionally writes to {@code output}.
     */
    public Builder<I, K, V, O, R> setReducer(Reducer<K, V, ? extends O> reducer) {
      this.reducer = reducer;
      return this;
    }

    /**
     * @param numReducers specify the number of reducers
     */
    public Builder<I, K, V, O, R> setNumReducers(int numReducers) {
      this.numReducers = numReducers;
      return this;
    }

    public MapReduceSpecification<I, K, V, O, R> build() {
      return new MapReduceSpecification<>(this);
    }
  }

  private final Mapper<I, K, V> mapper;
  private final Marshaller<K> keyMarshaller;
  private final Marshaller<V> valueMarshaller;
  private final Reducer<K, V, O> reducer;
  private final int numReducers;

  @SuppressWarnings({"rawtypes", "unchecked"})
  private MapReduceSpecification(Builder builder) {
    super(builder);
    mapper = checkNotNull(builder.mapper, "Null mapper");
    // TODO(user): optimize on picking the best key/value marshaller based on the generic type
    keyMarshaller =
        Optional.fromNullable(builder.keyMarshaller).or(Marshallers.getSerializationMarshaller());
    valueMarshaller =
        Optional.fromNullable(builder.valueMarshaller).or(Marshallers.getSerializationMarshaller());
    reducer = checkNotNull(builder.reducer, "Null reducer");
    numReducers = checkNumShards("reduce", Optional.fromNullable(builder.numReducers).or(1),
        MapReduceConstants.MAX_REDUCE_SHARDS);
  }

  private static int checkNumShards(String stage, int shards, int maxShards) {
    checkArgument(shards > 0 && shards <= maxShards,
        "Invalid number of %s shards: %s must be between 1 and %s.", stage, shards, maxShards);
    return shards;
  }

  Mapper<I, K, V> getMapper() {
    return mapper;
  }

  Marshaller<K> getKeyMarshaller() {
    return keyMarshaller;
  }

  Marshaller<V> getValueMarshaller() {
    return valueMarshaller;
  }

  Reducer<K, V, O> getReducer() {
    return reducer;
  }

  int getNumReducers() {
    return numReducers;
  }

  @Override
  public String toString() {
    return getClass().getSimpleName() + "(" + getJobName() + ", " + getInput() + ", " + mapper
        + ", " + reducer + "[" + numReducers + "], " + getOutput() + ")";
  }
}
