/**
 * 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.sqoop.testutil;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapreduce.InputFormat;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.OutputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.sqoop.SqoopOptions;
import org.apache.sqoop.manager.HsqldbManager;
import org.apache.sqoop.manager.ImportJobContext;
import org.apache.sqoop.mapreduce.ImportJobBase;
import org.apache.sqoop.util.ImportException;
import org.apache.hadoop.util.ReflectionUtils;

/**
 * A ConnManager that uses dependency injection to control "import jobs"
 * that are mocked for testing.
 */
public class InjectableConnManager extends HsqldbManager {

  // The Configuration is used to control the injected classes.
  public static final String MAPPER_KEY = "sqoop.inject.mapper.class";
  public static final String INPUT_FORMAT_KEY =
      "sqoop.inject.input.format.class";
  public static final String OUTPUT_FORMAT_KEY =
      "sqoop.inject.output.format.class";
  public static final String IMPORT_JOB_KEY =
      "sqoop.inject.import.job.class";

  public InjectableConnManager(final SqoopOptions options) {
    super(options);
  }

  /**
   * Allow the user to inject custom mapper, input, and output formats
   * into the importTable() process.
   */
  @Override
  @SuppressWarnings("unchecked")
  public void importTable(ImportJobContext context)
      throws IOException, ImportException {

    SqoopOptions options = context.getOptions();
    Configuration conf = options.getConf();

    Class<? extends Mapper> mapperClass = (Class<? extends Mapper>)
        conf.getClass(MAPPER_KEY, Mapper.class);
    Class<? extends InputFormat> ifClass = (Class<? extends InputFormat>)
        conf.getClass(INPUT_FORMAT_KEY, TextInputFormat.class);
    Class<? extends OutputFormat> ofClass = (Class<? extends OutputFormat>)
        conf.getClass(OUTPUT_FORMAT_KEY, TextOutputFormat.class);

    Class<? extends ImportJobBase> jobClass = (Class<? extends ImportJobBase>)
        conf.getClass(IMPORT_JOB_KEY, ImportJobBase.class);

    String tableName = context.getTableName();

    // Instantiate the user's chosen ImportJobBase instance.
    ImportJobBase importJob = ReflectionUtils.newInstance(jobClass, conf);

    // And configure the dependencies to inject
    importJob.setOptions(options);
    importJob.setMapperClass(mapperClass);
    importJob.setInputFormatClass(ifClass);
    importJob.setOutputFormatClass(ofClass);

    importJob.runImport(tableName, context.getJarFile(),
        getSplitColumn(options, tableName), conf);
  }
}