/*
    * 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.IOException;
  
  import org.apache.log4j.Logger;
  import org.apache.commons.io.FileUtils;
  import org.apache.commons.io.filefilter.FileFilterUtils;
  import org.apache.commons.io.filefilter.IOFileFilter;
  import org.apache.commons.io.filefilter.WildcardFileFilter;
  
  /**
   *  Explore a whole directory hierarchy, looking for files to tail. 
   *   DirTailingAdaptor will not try to start tailing a file more than once,
   *   if the file hasn't been modified in the interim.  
   *   
   *  Offset param is used to track last finished scan.
   *  
   *  Mandatory first parameter is a directory with an optional unix style file
   *  filter. Mandatory second parameter
   *  is the name of an adaptor to start.
   * 
   *  If the specified directory does not exist, the DirTailer will continue
   *  running, and will start tailing if the directory is later created.
   *
   */
  public class DirTailingAdaptor extends AbstractAdaptor implements Runnable {
    
    static Logger log = Logger.getLogger(DirTailingAdaptor.class); 
    
    Thread scanThread = new Thread(this);
    long lastSweepStartTime;
    volatile boolean continueScanning=true;
    File baseDir;
    String baseDirName; 
    long scanInterval;
  
  	protected String adaptorName; // name of adaptors to start
  
    IOFileFilter fileFilter;
    String wildCharacter;
  
    @Override
    public void start(long offset) throws AdaptorException {
      scanInterval = control.getConfiguration().getInt("adaptor.dirscan.intervalMs", 10000);
        
      scanThread.start();
      lastSweepStartTime = offset;
      try {
        baseDirName = baseDir.getCanonicalPath();
      } catch(IOException e) {
        throw new AdaptorException(e);
      }
    }
    
    public void run() {
      try {
        log.debug("dir tailer starting to scan");
        while(continueScanning) {
          try {
            long sweepStartTime = System.currentTimeMillis();
            scanDirHierarchy(baseDir);
            lastSweepStartTime=sweepStartTime;
            control.reportCommit(this, lastSweepStartTime);
          } catch(IOException e) {
            log.warn(e);
          }
          Thread.sleep(scanInterval);
        }
      } catch(InterruptedException e) {
      }
    }
    
    /*
     * Coded recursively.  Base case is a single non-dir file.
     */
    private void scanDirHierarchy(File dir) throws IOException {
      if(!dir.exists())
        return;
      if(!dir.isDirectory()) {
        //Don't start tailing if we would have gotten it on the last pass
        if(dir.lastModified() >= lastSweepStartTime) {
 				String newAdaptorID = control.processAddCommand(getAdaptorAddCommand(dir));
 
             log.info("DirTailingAdaptor " + adaptorID +  "  started new adaptor " + newAdaptorID);
        }
       
       } else {
         log.info("Scanning directory: " + dir.getName());
         
         for(Object f: FileUtils.listFiles(dir, fileFilter, FileFilterUtils.trueFileFilter())) {
          scanDirHierarchy((File)f);
         }
       }
   }
   
 	protected String getAdaptorAddCommand(File dir) throws IOException {
 		return "add " + adaptorName + " " + type + " " + dir.getCanonicalPath() + " 0";
 	}
 
   @Override
   public String getCurrentStatus() {
     return this.wildCharacter == null ? (type + " " + baseDirName + " " + adaptorName)
     		:(type + " " + baseDirName + " " + this.wildCharacter + " " + adaptorName);
   }
 
   @Override
   public String parseArgs(String status) {
     
     String[] args = status.split(" ");
      
     if(args.length == 2){
      baseDir = new File(args[0]);
      fileFilter = FileFilterUtils.trueFileFilter();
      adaptorName = args[1];
     }else if(args.length == 3){
      baseDir = new File(args[0]);
      this.wildCharacter = args[ 1 ];
      fileFilter = getFileFilter();
      adaptorName = args[2]; 
     }else{
      log.warn("bad syntax in DirTailingAdaptor args");
      return null;
     }
     
     return (args.length == 2)? baseDir + " " + adaptorName : baseDir + " " + this.wildCharacter + " " + adaptorName;  //both params mandatory
   }
 
   @Override
   public long shutdown(AdaptorShutdownPolicy shutdownPolicy)
       throws AdaptorException {
     continueScanning = false;
     return lastSweepStartTime;
   }
 
   /**
    * Returns {@link IOFileFilter} constructed using the wild character. Subclasses can override this method 
    * return their own version of {@link IOFileFilter} instance.
    *  
    * @return {@link IOFileFilter} constructed using the wild character. Subclasses can override this method 
    * return their own version of {@link IOFileFilter} instance.
    */
   protected IOFileFilter getFileFilter() {
   	return new WildcardFileFilter( this.wildCharacter );
   }
 }