Flume-TailFileSource

package org.apache.flume.source;

import java.io.File;
import java.io.RandomAccessFile;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.EventDrivenSource;
import org.apache.flume.conf.Configurable;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.instrumentation.SourceCounter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class TailFileSource extends AbstractSource implements EventDrivenSource,
  Configurable {
   private static final Logger logger = LoggerFactory.getLogger(TailFileSource.class);
 private SourceCounter sourceCounter;
 private String pointerFile;
 private String tailFile;
 private long collectInterval;
 private int batchLine;
 private boolean batch;
 private AtomicBoolean tailRun=new AtomicBoolean(true);
 private AtomicLong cursor=new AtomicLong(0);
 

 
 @Override
 public void configure(Context context) {
  if (sourceCounter == null) {
   sourceCounter = new SourceCounter(getName());
  }
  pointerFile = context.getString("tailfile.pointer","cursor.pt");
  tailFile = context.getString("tailfile.file","data.txt");
  collectInterval=context.getLong("tailfile.collectInterval",3000L);
  batchLine=context.getInteger("tailfile.batchLine",10);
  batch=context.getBoolean("tailfile.batch",false);
  
 }

 @Override
 public synchronized void start() {
  super.start();
  sourceCounter.start();
  TailFile tf=new TailFile();
  tf.addFileTailerListener(new FileTailerListener(){
   @Override
   public void newFileLine(List<Event> events) {
    if(!events.isEmpty()){
     getChannelProcessor().processEventBatch(events);
     sourceCounter.incrementAppendBatchAcceptedCount();
        sourceCounter.addToEventAcceptedCount(events.size());
    }
   }

   @Override
   public void newFileLine(Event event) {
     getChannelProcessor().processEvent(event);
     sourceCounter.incrementAppendAcceptedCount();
   }
  });
  Thread t=new java.lang.Thread(tf);
  t.setDaemon(true);
  t.start();
 }

 @Override
 public synchronized void stop() {
  tailRun.set(false);
  super.stop();
  sourceCounter.stop();
 }
 
 
 protected interface FileTailerListener{
     public void newFileLine(List<Event> events);
     public void newFileLine(Event event);
 }
 
 
 protected class TailFile implements java.lang.Runnable{
  private Set<FileTailerListener> listeners = new HashSet<FileTailerListener>();
  
  
     public void addFileTailerListener(FileTailerListener l) {
         this.listeners.add(l);
     }

     public void removeFileTailerListener(FileTailerListener l) {
         this.listeners.remove(l);
     }

  @Override
  public void run() {
   long[] st=this.readPointerFile();
   RandomAccessFile file=null;
   boolean flag=true;
   while(flag){
    try {
     File tf=new File(tailFile);
     file = new RandomAccessFile(tf, "r");
     if(st[0]==tf.lastModified()){
      cursor.set(st[1]);
     }else{
      st[0]=tf.lastModified();
      cursor.set(0);
     }
     flag=false;
    } catch (Exception e) {
     try {
      logger.error(e.getMessage()+",will retry file:"+tailFile);
      Thread.sleep(5000);
     } catch (Exception e1) {
      
     }
    }
   }
   
   while(tailRun.get()){
    try {
     if(!this.sameTailFile(st[0])) {
      logger.error("file change:"+tailFile);
      File tf=new File(tailFile);
      file = new RandomAccessFile(tf, "r");
      st[0]=tf.lastModified();
      cursor.set(0);
     }
     
     long fileLength =file.length();
     if (fileLength < cursor.get()) {
         cursor.set(fileLength);
     }
     
     if (fileLength > cursor.get()) {
         file.seek(cursor.get());
         String line = file.readLine();
         int i=1;
        
         if(batch){
          java.util.List<Event> batchAl=new java.util.ArrayList<Event>(batchLine);
          while (line != null) {
           batchAl.add(EventBuilder.withBody(line.getBytes()));
           if(i%batchLine==0) {
            fireNewFileLine(batchAl);
            batchAl.clear();
            cursor.set(file.getFilePointer());
            st[1]=cursor.get();
            writePointerFile(st);
           }
              line = file.readLine();
              i++;
          }
         
          if(!batchAl.isEmpty()){
           fireNewFileLine(batchAl);
           batchAl.clear();
           cursor.set(file.getFilePointer());
           st[1]=cursor.get();
           writePointerFile(st);
          }
         }else{
          while(line!=null){
           fireNewFileLine(EventBuilder.withBody(line.getBytes()));
           line = file.readLine();
          }
           cursor.set(file.getFilePointer());
           st[1]=cursor.get();
           writePointerFile(st);
         }

     }
     Thread.sleep(collectInterval);
    } catch (Exception e) {
     
    }
    
   }
   
   try {
    if(file!=null) file.close();
   } catch (Exception e) {
    
   }
  }
  
  
  private long[] readPointerFile(){
   logger.info("read pointerFile:"+pointerFile);
   java.io.ObjectInputStream ois=null;
   long[] temp={0L,0L};
   try {
    ois=new java.io.ObjectInputStream(new java.io.FileInputStream(new File(pointerFile)));
    temp[0]=ois.readLong();
    temp[1]=ois.readLong();
   } catch (Exception e) {
    logger.error("can't read pointerFile:"+pointerFile);
   } finally{
    try {
     if(ois!=null)ois.close();
    } catch (Exception e) {
     
    }
   }
   return temp;
  }
  
  private void writePointerFile(long[] temp){
   logger.debug("write pointerFile:"+pointerFile);
   java.io.ObjectOutputStream oos=null;
   try {
    oos=new java.io.ObjectOutputStream(new java.io.FileOutputStream(pointerFile));
    oos.writeLong(temp[0]);
    oos.writeLong(temp[1]);
   } catch (Exception e) {
    logger.error("can't write pointerFile:"+pointerFile);
   }finally{
    try {
     if(oos!=null)oos.close();
    } catch (Exception e) {
     
    }
   }
  
  }
  
  private boolean sameTailFile(long time){
   return new File(tailFile).lastModified()==time?true:false;
  }
  
     
     protected void fireNewFileLine(List<Event> events) {
         for (Iterator<FileTailerListener> i = this.listeners.iterator(); i.hasNext();) {
          FileTailerListener l =  i.next();
             l.newFileLine(events);
         }
     }
    
    
     protected void fireNewFileLine(Event event) {
         for (Iterator<FileTailerListener> i = this.listeners.iterator(); i.hasNext();) {
          FileTailerListener l =  i.next();
             l.newFileLine(event);
         }
     }
 }
}