/*
    * 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.hadoop.chukwa.util;
  
  import org.apache.hadoop.fs.Path;
  import org.apache.hadoop.fs.FileSystem;
  import org.apache.hadoop.fs.FSDataOutputStream;
  import org.apache.hadoop.chukwa.extraction.demux.processor.mapper.MapProcessor;
  import org.apache.hadoop.chukwa.extraction.demux.processor.mapper.TsProcessor;
  import org.apache.hadoop.chukwa.extraction.demux.Demux;
  import org.apache.hadoop.chukwa.extraction.engine.ChukwaRecordKey;
  import org.apache.hadoop.chukwa.extraction.engine.ChukwaRecord;
  import org.apache.hadoop.chukwa.ChunkImpl;
  import org.apache.hadoop.chukwa.ChukwaArchiveKey;
  import org.apache.hadoop.conf.Configuration;
  import org.apache.hadoop.io.SequenceFile;
  import org.apache.hadoop.mapred.OutputCollector;
  import org.apache.hadoop.mapred.JobConf;
  import org.apache.hadoop.mapred.Reporter;
  
  import java.io.FileInputStream;
  import java.io.IOException;
  import java.io.File;
  import java.io.BufferedReader;
  import java.io.InputStreamReader;
  import java.nio.charset.Charset;
  
  /**
   * Helper class used to create sequence files of Chukwa records
   */
  public class CreateRecordFile {
  
     public static void makeTestSequenceFile(File inputFile,
                                             Path outputFile,
                                             String clusterName,
                                             String dataType,
                                             String streamName,
                                             MapProcessor processor) throws IOException {
  
       //initialize the output collector and the default processor
       MockOutputCollector collector = new MockOutputCollector();
       if (processor == null) processor = new TsProcessor();
  
       //initialize the sequence file writer
       Configuration conf = new Configuration();
       FileSystem fs = outputFile.getFileSystem(conf);
       FSDataOutputStream out = fs.create(outputFile);
  
       SequenceFile.Writer seqFileWriter = SequenceFile.createWriter(conf, out,
                                      ChukwaRecordKey.class, ChukwaRecord.class,
                                      SequenceFile.CompressionType.NONE, null);
       long lastSeqID = 0;
       String line;
       FileInputStream fis = new FileInputStream(inputFile);
       BufferedReader reader = new BufferedReader(new InputStreamReader(fis, Charset.forName("UTF-8")));
  
       // for each line, create a chunk and an arckive key, pass it to the
       // processor, then write it to the sequence file.  
       while ((line = reader.readLine()) != null) {
  
         ChunkImpl chunk = new ChunkImpl(dataType, streamName,
           line.length()  + lastSeqID, line.getBytes(Charset.forName("UTF-8")), null);
         lastSeqID += line.length();
         chunk.addTag("cluster=\"" + clusterName + "\"");
  
         ChukwaArchiveKey archiveKey = new ChukwaArchiveKey();
         archiveKey.setTimePartition(System.currentTimeMillis());
         archiveKey.setDataType(chunk.getDataType());
         archiveKey.setStreamName(chunk.getStreamName());
         archiveKey.setSeqId(chunk.getSeqID());
  
         processor.process(archiveKey, chunk, collector, Reporter.NULL);
         seqFileWriter.append(collector.getChukwaRecordKey(),
                              collector.getChukwaRecord());
       }
  
       out.flush();
       out.close();
       seqFileWriter.close();
       reader.close();
     }
  
     private static class MockOutputCollector
            implements OutputCollector<ChukwaRecordKey, ChukwaRecord> {
      ChukwaRecordKey chukwaRecordKey;
      ChukwaRecord chukwaRecord;
 
      public void collect(ChukwaRecordKey chukwaRecordKey,
                          ChukwaRecord chukwaRecord) throws IOException {
        this.chukwaRecordKey = chukwaRecordKey;
        this.chukwaRecord = chukwaRecord;
      }
 
      public ChukwaRecordKey getChukwaRecordKey() { return chukwaRecordKey; }
      public ChukwaRecord getChukwaRecord() { return chukwaRecord; }
    }
 
    public static void main(String[] args) throws IOException,
                                                  ClassNotFoundException,
                                                  IllegalAccessException,
                                                  InstantiationException {
      if(args.length == 0 || (args.length==1 && args[0].contains("-h"))) {
        usage();
        return;
      }
 
      File inputFile = new File(args[0]);
      Path outputFile = new Path(args[1]);
      String clusterName = "testClusterName";
      String dataType = "testDataType";
      String streamName = "testStreamName";               
      MapProcessor processor = new TsProcessor();
      Path confFile = null;
 
      if (args.length > 2) clusterName = args[2];
      if (args.length > 3) dataType = args[3];
      if (args.length > 4) streamName = args[4];
 
      if (args.length > 5) {
        Class<?> clazz = null;
        try {
          clazz = Class.forName(args[5]);
        }
        catch (ClassNotFoundException e) {
          try {
            clazz = Class.forName(
                  "org.apache.hadoop.chukwa.extraction.demux.processor.mapper." + args[5]);
          }
          catch (Exception e2) {
            throw e;
          }
        }
        processor = (MapProcessor)clazz.newInstance();
      }
 
      if (args.length > 6) {
        confFile = new Path(args[6]);
        Demux.jobConf = new JobConf(confFile);
      }
 
      System.out.println("Creating sequence file using the following input:");
      System.out.println("inputFile  : " + inputFile);
      System.out.println("outputFile : " + outputFile);
      System.out.println("clusterName: " + clusterName);
      System.out.println("dataType   : " + dataType);
      System.out.println("streamName : " + streamName);
      System.out.println("processor  : " + processor.getClass().getName());
      System.out.println("confFile   : " + confFile);
 
      makeTestSequenceFile(inputFile, outputFile, clusterName, dataType, streamName, processor);
 
      System.out.println("Done");
    }
 
    public static void usage() {
      System.out.println("Usage: java " + CreateRecordFile.class.toString().split(" ")[1] + " <inputFile> <outputFile> [<clusterName> <dataType> <streamName> <processorClass> [confFile]]");
      System.out.println("Description: Takes a plain text input file and generates a Hadoop sequence file contaning ChukwaRecordKey,ChukwaRecord entries");
      System.out.println("Parameters: inputFile      - Text input file to read");
      System.out.println("            outputFile     - Sequence file to create");
      System.out.println("            clusterName    - Cluster name to use in the records");
      System.out.println("            dataType       - Data type to use in the records");
      System.out.println("            streamName     - Stream name to use in the records");
      System.out.println("            processorClass - Processor class to use. Defaults to TsProcessor");
      System.out.println("            confFile       - File to use to create the JobConf");
    }
 }