/*
 * 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.datacollection.adaptor;

import java.io.File;
import java.io.RandomAccessFile;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;

import org.apache.hadoop.chukwa.ChunkImpl;
import org.apache.hadoop.chukwa.datacollection.ChunkReceiver;
import org.apache.hadoop.chukwa.datacollection.agent.ChukwaAgent;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.chukwa.util.ExceptionUtil;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;


class FileAdaptorTailer extends Thread {
  static Logger log = Logger.getLogger(FileAdaptorTailer.class);
  private List<FileAdaptor> adaptors = null;
  
  /**
   * How often to call each adaptor.
   */
  int DEFAULT_SAMPLE_PERIOD_MS = 1000 * 10;
  int SAMPLE_PERIOD_MS = DEFAULT_SAMPLE_PERIOD_MS;

  
  public FileAdaptorTailer() {
    ChukwaAgent agent = ChukwaAgent.getAgent();
    if (agent != null) {
      Configuration conf = agent.getConfiguration();
      if (conf != null) {
        SAMPLE_PERIOD_MS = conf.getInt(
            "chukwaAgent.adaptor.context.switch.time",
            DEFAULT_SAMPLE_PERIOD_MS);
      }
    }
    
    // iterations are much more common than adding a new adaptor
    adaptors = new CopyOnWriteArrayList<FileAdaptor>();

    setDaemon(true);
    start();// start the FileAdaptorTailer thread
  }
  @Override
  public void run() {
    while(true) {
      try {

        long startTime = System.currentTimeMillis();
        for (FileAdaptor adaptor: adaptors) {
          log.info("calling sendFile for " + adaptor.toWatch.getCanonicalPath());
          adaptor.sendFile(); 
        }
        
        long timeToReadFiles = System.currentTimeMillis() - startTime;
        if (timeToReadFiles < SAMPLE_PERIOD_MS) {
          Thread.sleep(SAMPLE_PERIOD_MS);
        }
        
      }catch (Throwable e) {
        log.warn("Exception in FileAdaptorTailer:",e);
      }
    }
  }
  
  public void addFileAdaptor(FileAdaptor adaptor) {
    adaptors.add(adaptor);
  }
  
  public void removeFileAdaptor(FileAdaptor adaptor) {
    adaptors.remove(adaptor);
  }
}

/**
 * File Adaptor push small size file in one chunk to collector
 */
public class FileAdaptor extends AbstractAdaptor {

  static Logger log = Logger.getLogger(FileAdaptor.class);
  static FileAdaptorTailer tailer = null;
  
  static final int DEFAULT_TIMEOUT_PERIOD = 5*60*1000;
  int TIMEOUT_PERIOD = DEFAULT_TIMEOUT_PERIOD;
  
  static {
    tailer = new FileAdaptorTailer();
  }
  
  private long startTime = 0;
  private long timeOut = 0;
  
  protected volatile boolean finished = false;
  File toWatch;
  protected RandomAccessFile reader = null;
  protected long fileReadOffset;
  protected boolean deleteFileOnClose = false;
  protected boolean shutdownCalled = false;
  
  /**
   * The logical offset of the first byte of the file
   */
  private long offsetOfFirstByte = 0;

  public void start(long bytes) {
    // in this case params = filename
    log.info("adaptor id: " + adaptorID + " started file adaptor on file "
        + toWatch);
    this.startTime = System.currentTimeMillis();
    TIMEOUT_PERIOD = control.getConfiguration().getInt(
        "chukwaAgent.adaptor.fileadaptor.timeoutperiod",
        DEFAULT_TIMEOUT_PERIOD);
    this.timeOut = startTime + TIMEOUT_PERIOD;
    
    tailer.addFileAdaptor(this);
  }

  void sendFile() {
    long now = System.currentTimeMillis() ;
    long oneMinAgo = now - (60*1000);
    if (toWatch.exists()) {
     if (toWatch.lastModified() > oneMinAgo && now < timeOut) {
       log.info("Last modified time less than one minute, keep waiting");
       return;
     } else {
       try {
         
         long bufSize = toWatch.length();
         byte[] buf = new byte[(int) bufSize];
         
         reader = new RandomAccessFile(toWatch, "r");
         reader.read(buf);
         reader.close();
         reader = null;
         
         long fileTime = toWatch.lastModified();
         int bytesUsed = extractRecords(dest, 0, buf, fileTime);
         this.fileReadOffset = bytesUsed;
         finished = true;
         deregisterAndStop();
         cleanUp();
       } catch(Exception e) {
         log.warn("Exception while trying to read: " + toWatch.getAbsolutePath(),e);
       }  finally {
         if (reader != null) {
           try {
             reader.close();
           } catch (Exception e) {
             log.debug(ExceptionUtil.getStackTrace(e));
          }
           reader = null;
         }
       }
     }
    } else {
      if (now > timeOut) {
        finished = true;
        log.warn("Couldn't read this file: " + toWatch.getAbsolutePath());
        deregisterAndStop();
        cleanUp() ;
      }
    }
  }
  
  private void cleanUp() {
    tailer.removeFileAdaptor(this);
    if (reader != null) {
      try {
        reader.close();
      } catch (Exception e) {
        log.debug(ExceptionUtil.getStackTrace(e));
     }
      reader = null;
    } 
  }
  

  @Override
  public long shutdown(AdaptorShutdownPolicy shutdownPolicy) {
    log.info("Enter Shutdown:" + shutdownPolicy.name()+ " - ObjectId:" + this);
    switch(shutdownPolicy) {
      case GRACEFULLY : {
        int retry = 0;
        while (!finished && retry < 60) {
          try {
            log.info("GRACEFULLY Retry:" + retry);
            Thread.sleep(1000);
            retry++;
          } catch (InterruptedException ex) {
          }
        } 
      }
      break;
      case WAIT_TILL_FINISHED : {
        int retry = 0;
        while (!finished) {
          try {
            if (retry%100 == 0) {
              log.info("WAIT_TILL_FINISHED Retry:" + retry);
            }

            Thread.sleep(1000);
            retry++;
          } catch (InterruptedException ex) {
          }
        } 
      }

      break;
      default :
        cleanUp();
        break;
    }

    if (deleteFileOnClose && toWatch != null) {
      if (log.isDebugEnabled()) {
        log.debug("About to delete " + toWatch.getAbsolutePath());
      }
      if (toWatch.delete()) {
        if (log.isInfoEnabled()) {
          log.debug("Successfully deleted " + toWatch.getAbsolutePath());
        }
      } else {
        if (log.isEnabledFor(Level.WARN)) {
          log.warn("Could not delete " + toWatch.getAbsolutePath() + " (for unknown reason)");
        }
      }
    }
    
    log.info("Exist Shutdown:" + shutdownPolicy.name()+ " - ObjectId:" + this);
    return fileReadOffset + offsetOfFirstByte;
  }
  
  public String parseArgs(String params) {

    String[] words = params.split(" ");
    if (words.length == 2) {
      if (words[1].equals("delete")) {
        deleteFileOnClose = true;
        toWatch = new File(words[0]);
      } else {
        offsetOfFirstByte = Long.parseLong(words[0]);
        toWatch = new File(words[1]);
      }
    } else if (words.length == 3) {
      offsetOfFirstByte = Long.parseLong(words[0]);
      toWatch = new File(words[1]);
      deleteFileOnClose = words[2].equals("delete");
    } else {
      toWatch = new File(params);
    }
    return toWatch.getAbsolutePath();
  }

  /**
   * Extract records from a byte sequence
   * 
   * @param eq
   *          the queue to stick the new chunk[s] in
   * @param buffOffsetInFile
   *          the byte offset in the stream at which buf[] begins
   * @param buf
   *          the byte buffer to extract records from
   * @return the number of bytes processed
   * @throws InterruptedException
   */
  protected int extractRecords(final ChunkReceiver eq, long buffOffsetInFile,
      byte[] buf, long fileTime) throws InterruptedException {
    final ChunkImpl chunk = new ChunkImpl(type, toWatch.getAbsolutePath(),
        buffOffsetInFile + buf.length, buf, this);
    chunk.addTag("time=\"" + fileTime + "\"");
    log.info("Adding " + toWatch.getAbsolutePath() + " to the queue");
    eq.add(chunk);
    log.info( toWatch.getAbsolutePath() + " added to the queue");
    return buf.length;
  }

  @Override
  public String getCurrentStatus() {
    return type.trim() + " " + offsetOfFirstByte + " " + toWatch.getPath();
  }

}