/*
    * 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.filetailer;
  
  import java.io.File;
  import java.io.FileFilter;
  import java.io.IOException;
  import java.io.RandomAccessFile;
  import java.util.regex.Matcher;
  import java.util.regex.Pattern;
  import java.util.Collections;
  import java.util.LinkedList;
  
  import org.apache.commons.lang3.builder.HashCodeBuilder;
  
  /**
   * Checkpoint state:
   *   date modified of most-recently tailed file, offset of first byte of that file,
   *   then regular FTA arts 
   *
   */
  public class RCheckFTAdaptor extends LWFTAdaptor implements FileFilter {
    
    private static class FPair implements Comparable<FPair> {
      File f;
      long mod;
      FPair(File f) {
        this.f = f;
        mod = f.lastModified();
      }
      /**
       * -1   implies this is LESS THAN o
       */
      @Override
      public int compareTo(FPair o) {
        if(mod < o.mod)
          return -1;
        else if (mod > o.mod)
          return 1;
        //want toWatch to be last after a rotation; otherwise, this is basically 
        //just a heuristic that hasn't been tuned yet
        else return (o.f.getName().compareTo(f.getName()));
      }
      
      @Override
      public boolean equals(Object o) {
        if(o instanceof FPair) {
          return mod == ((FPair) o).mod;
        } else {
          return false;
        }
      }
      
      @Override
      public int hashCode() {
        return new HashCodeBuilder(643, 1321).
            append(this.mod).
            toHashCode();
      }
    }
    
    long prevFileLastModDate = 0;
    LinkedList<FPair> fileQ = new LinkedList<FPair>(); 
    String fBaseName;
    File cur; //this is the actual physical file being watched.  
              // in contrast, toWatch is the path specified by the user
    boolean caughtUp = false;
    /**
     * Check for date-modified and offset; if absent assume we just got a name.
     */
    @Override
    public String parseArgs(String params) { 
      Pattern cmd = Pattern.compile("d:(\\d+)\\s+(\\d+)\\s+(.+)\\s?");
      Matcher m = cmd.matcher(params);
      if (m.matches()) {
        prevFileLastModDate = Long.parseLong(m.group(1));
        offsetOfFirstByte = Long.parseLong(m.group(2));
        toWatch = new File(m.group(3)).getAbsoluteFile();
      } else {
        toWatch = new File(params.trim()).getAbsoluteFile();
      }
      fBaseName = toWatch.getName();
      return toWatch.getAbsolutePath();
    }
   
   public String getCurrentStatus() {
     return type.trim() + " d:" + prevFileLastModDate + " "  + offsetOfFirstByte + " " + toWatch.getPath();
   }
 
   @Override
   public boolean accept(File pathname) {
     return pathname.getName().startsWith(fBaseName) &&
      ( pathname.getName().equals(fBaseName) ||
        pathname.lastModified() > prevFileLastModDate);
   }
   
   
   protected void mkFileQ() {
     
     File toWatchDir = toWatch.getParentFile();
     File[] candidates = toWatchDir.listFiles(this);
     if(candidates == null) {
       log.error(toWatchDir + " is not a directory in "+adaptorID);
     } else {
       log.debug("saw " + candidates.length + " files matching pattern");
       fileQ = new LinkedList<FPair>();
       for(File f:candidates)
         fileQ.add(new FPair(f));
       Collections.sort(fileQ);
     } 
   }
   
   protected void advanceQ() {
     FPair next = fileQ.poll();
     if(next != null) {
       cur = next.f;
       caughtUp = toWatch.equals(cur);
       if(caughtUp && !fileQ.isEmpty()) 
         log.warn("expected rotation queue to be empty when caught up...");
     }
     else {
       cur = null;
       caughtUp = true;
     }
   }
   
   @Override
   public void start(long offset) {
     mkFileQ(); //figure out what to watch
     advanceQ();
     super.start(offset);
   }
   
   @Override
   public boolean tailFile()
   throws InterruptedException {
     boolean hasMoreData = false;
     try {
       
       if(caughtUp) {
         //we're caught up and watching an unrotated file
         mkFileQ(); //figure out what to watch
         advanceQ();
       }
       if(cur == null) //file we're watching doesn't exist
         return false;
       
       
       long len = cur.length();
       long tsPreTail = cur.exists() ? cur.lastModified() : prevFileLastModDate;
 
       if(log.isDebugEnabled())
         log.debug(adaptorID + " treating " + cur + " as " + toWatch + " len = " + len);
       
       if(len < fileReadOffset) {
         log.info("file "+ cur +" shrank from " + fileReadOffset + " to " + len);
         //no unseen changes to prev version, since mod date is older than last scan.
         offsetOfFirstByte += fileReadOffset;
         fileReadOffset = 0;
       } else if(len > fileReadOffset) {
         log.debug("slurping from " + cur+ " at offset " + fileReadOffset);
         RandomAccessFile reader = new RandomAccessFile(cur, "r");
         slurp(len, reader);
         reader.close();
       } else {
         //we're either caught up or at EOF
         if (!caughtUp) {
           prevFileLastModDate = cur.lastModified();
           //Hit EOF on an already-rotated file. Move on!
           offsetOfFirstByte += fileReadOffset;
           fileReadOffset = 0;
           advanceQ();
           log.debug("not caught up, and hit EOF.  Moving forward in queue to " + cur);
         } else 
           prevFileLastModDate = tsPreTail;
 
       }
         
     } catch(IOException e) {
       log.warn("IOException in "+adaptorID, e);
       deregisterAndStop();
     }
     
     return hasMoreData;
   }
   
 
   public String toString() {
     return "Rotation-aware Tailer on " + toWatch;
   }
 }