/*
    * 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.validationframework.util;
  
  
  import java.io.DataInputStream;
  import java.io.EOFException;
  import java.io.File;
  import java.io.FileInputStream;
  import java.io.FileOutputStream;
  import java.io.FileWriter;
  import java.io.IOException;
  import java.net.URI;
  import org.apache.hadoop.chukwa.ChukwaArchiveKey;
  import org.apache.hadoop.chukwa.Chunk;
  import org.apache.hadoop.chukwa.ChunkImpl;
  import org.apache.hadoop.chukwa.conf.ChukwaConfiguration;
  import org.apache.hadoop.chukwa.extraction.engine.ChukwaRecord;
  import org.apache.hadoop.chukwa.extraction.engine.ChukwaRecordKey;
  import org.apache.hadoop.conf.Configuration;
  import org.apache.hadoop.fs.FileSystem;
  import org.apache.hadoop.fs.Path;
  import org.apache.hadoop.io.SequenceFile;
  import org.apache.log4j.Logger;
  
  public class DataOperations {
    static Logger log = Logger.getLogger(DataOperations.class);
  
    public static void copyFile(String fromFileName, String toFileName)
        throws IOException {
      File fromFile = new File(fromFileName);
      File toFile = new File(toFileName);
  
      FileInputStream from = null;
      FileOutputStream to = null;
      try {
        from = new FileInputStream(fromFile);
        to = new FileOutputStream(toFile);
        byte[] buffer = new byte[4096];
        int bytesRead;
  
        while ((bytesRead = from.read(buffer)) != -1)
          to.write(buffer, 0, bytesRead); // write
      } finally {
        if (from != null)
          try {
            from.close();
          } catch (IOException e) {
            ;
          }
        if (to != null)
          try {
            to.close();
          } catch (IOException e) {
            // ;
          }
      }
    }
  
    public static boolean validateMD5(String inputFile, String testFile) {
      // System.out.println("validateMD5 [" + inputFile + "] [" + testFile
      // + "]");
      String md5_1 = MD5.checksum(new File(inputFile));
      String md5_2 = MD5.checksum(new File(testFile));
      // System.out.println("MD5 [" + md5_1 + "] [" + md5_2 + "]");
      return md5_1.intern() == md5_2.intern();
    }
  
    public static boolean validateMD5(FileSystem fs, Path inputFile, Path testFile) {
      // System.out.println("validateMD5 [" + inputFile + "] [" + testFile
      // + "]");
      String md5_1 = MD5.checksum(fs, inputFile);
      String md5_2 = MD5.checksum(fs, testFile);
      // System.out.println("MD5 [" + md5_1 + "] [" + md5_2 + "]");
      return md5_1.intern() == md5_2.intern();
    }
  
    public static boolean validateChukwaRecords(FileSystem fs,
        Configuration conf, Path inputFile, Path testFile) {
      SequenceFile.Reader goldReader = null;
      SequenceFile.Reader testReader = null;
      try {
        // log.info(">>>>>>>>>>>>>> Openning records [" + inputFile.getName()
        // +"][" + testFile.getName() +"]");
       goldReader = new SequenceFile.Reader(fs, inputFile, conf);
       testReader = new SequenceFile.Reader(fs, testFile, conf);
 
       ChukwaRecordKey goldKey = new ChukwaRecordKey();
       ChukwaRecord goldRecord = new ChukwaRecord();
 
       ChukwaRecordKey testKey = new ChukwaRecordKey();
       ChukwaRecord testRecord = new ChukwaRecord();
 
       // log.info(">>>>>>>>>>>>>> Start reading");
       while (goldReader.next(goldKey, goldRecord)) {
         testReader.next(testKey, testRecord);
 
         if (goldKey.compareTo(testKey) != 0) {
           log.info(">>>>>>>>>>>>>> Not the same Key");
           log.info(">>>>>>>>>>>>>> Record [" + goldKey.getKey() + "] ["
               + goldKey.getReduceType() + "]");
           log.info(">>>>>>>>>>>>>> Record [" + testKey.getKey() + "] ["
               + testKey.getReduceType() + "]");
           return false;
         }
 
         if (goldRecord.compareTo(testRecord) != 0) {
           log.info(">>>>>>>>>>>>>> Not the same Value");
           log.info(">>>>>>>>>>>>>> Record [" + goldKey.getKey() + "] ["
               + goldKey.getReduceType() + "]");
           log.info(">>>>>>>>>>>>>> Record [" + testKey.getKey() + "] ["
               + testKey.getReduceType() + "]");
           log.info(">>>>>>>>>>>>>> Gold Value [" + goldRecord.toString() + "]");
           log.info(">>>>>>>>>>>>>> Test value [" + testRecord.toString() + "]");
 
           return false;
         }
       }
       // log.info(">>>>>>>>>>>>>> Same File");
       return true;
     } catch (IOException e) {
       e.printStackTrace();
       return false;
     } finally {
       try {
         goldReader.close();
         testReader.close();
       } catch (IOException e) {
       }
 
     }
   }
 
   public static void extractRawLogFromdataSink(String directory, String fileName)
       throws Exception {
     ChukwaConfiguration conf = new ChukwaConfiguration();
     String fsName = conf.get("writer.hdfs.filesystem");
     FileSystem fs = FileSystem.get(new URI(fsName), conf);
 
     SequenceFile.Reader r = new SequenceFile.Reader(fs, new Path(directory
         + fileName + ".done"), conf);
 
     File outputFile = new File(directory + fileName + ".raw");
 
     ChukwaArchiveKey key = new ChukwaArchiveKey();
     ChunkImpl chunk = ChunkImpl.getBlankChunk();
     FileWriter out = new FileWriter(outputFile);
     try {
       while (r.next(key, chunk)) {
         out.write(new String(chunk.getData()));
       }
     } finally {
       out.close();
       r.close();
     }
 
   }
 
   public static void extractRawLogFromDump(String directory, String fileName)
       throws Exception {
     File inputFile = new File(directory + fileName + ".bin");
     File outputFile = new File(directory + fileName + ".raw");
     DataInputStream dis = new DataInputStream(new FileInputStream(inputFile));
     Chunk chunk = null;
     FileWriter out = new FileWriter(outputFile);
     boolean eof = false;
     do {
       try {
         chunk = ChunkImpl.read(dis);
         out.write(new String(chunk.getData()));
       } catch (EOFException e) {
         eof = true;
       }
 
     } while (!eof);
 
     dis.close();
     out.close();
   }
 }