Java 类org.apache.commons.io.input.Tailer 实例源码

private void startTailingLog() {
  TailerListener listener = new TailerListenerAdapter() {
    @Override
    public void handle(String line) {
      System.out.println(testName + ": " + line);
    }
  };

  // Tail the log
  File file = new File(getBasedir() + File.separator + getLogFileName());
  try {
    if (file.exists()) {
      file.delete();
    }
    file.createNewFile();
  } catch (IOException e) {
    e.printStackTrace();
  }
  Tailer tailer = new Tailer(file, listener, TAIL_DELAY_MILLIS);
  Thread thread = new Thread(tailer);
  thread.setDaemon(true);
  thread.start();
}

private void registerLogFileTailThreads() throws IOException {
    final Collection<String> outputFileNames = new HashSet<>();
    final Collection<Appender> loggerAppenders = this.loggerContext.getConfiguration().getAppenders().values();
    loggerAppenders.forEach(appender -> {
        if (appender instanceof FileAppender) {
            outputFileNames.add(((FileAppender) appender).getFileName());
        } else if (appender instanceof RandomAccessFileAppender) {
            outputFileNames.add(((RandomAccessFileAppender) appender).getFileName());
        } else if (appender instanceof RollingFileAppender) {
            outputFileNames.add(((RollingFileAppender) appender).getFileName());
        } else if (appender instanceof MemoryMappedFileAppender) {
            outputFileNames.add(((MemoryMappedFileAppender) appender).getFileName());
        } else if (appender instanceof RollingRandomAccessFileAppender) {
            outputFileNames.add(((RollingRandomAccessFileAppender) appender).getFileName());
        }
    });

    outputFileNames.forEach(s -> {
        final Tailer t = new Tailer(new File(s), new LogTailerListener(), 100, false, true);
        final Thread thread = new Thread(t);
        thread.setPriority(Thread.MIN_PRIORITY);
        thread.setName(s);
        thread.start();
    });
}

/**
 * Starts tailing the given file, thus updating the log lines in the table.
 *
 * @param file
 *         the file to tail
 *
 * @throws FileNotFoundException
 *         if the file was not found
 */
public void startTailingFile(File file) throws FileNotFoundException {
    if (!file.exists()) {
        throw new FileNotFoundException(file.getAbsolutePath());
    }
    closeCurrentFile();
    config.getState().addToRecentFiles(file.getAbsolutePath());
    logTailListener = new BufferedLogTailListener(columnizer.getValue(), columnizedLogs,
            config.getPreferences().getLogBufferSize());
    logTailListener.skipEmptyLogsProperty().bind(config.getPreferences().skipEmptyLogsProperty());
    logTailListener.limitNumberOfLogsProperty().bind(config.getPreferences().limitNumberOfLogsProperty());
    logTailListener.maxNumberOfLogsProperty().bind(config.getPreferences().maxNumberOfLogsProperty());
    tailer = Tailer.create(file, logTailListener, config.getPreferences().getTailingDelayInMillis());
    tailingFile.set(true);
    tailedFileName.set(file.getAbsolutePath());
}

/**
 * Creates a CsvStreamReader with supplied separator and quote char.
 *
 * @param source The file to an underlying CSV source
 * @param separator The delimiter to use for separating entries
 * @param quoteChar The character to use for quoted elements
 * @param escape The character to use for escaping a separator or quote
 * @param line The line number to skip for start reading
 * @param strictQuotes Sets if characters outside the quotes are ignored
 * @param ignoreLeadingWhiteSpace If true, parser should ignore
 *  white space before a quote in a field
 */
private CsvStreamReader(Source source, char separator, char quoteChar,
    char escape, int line, boolean strictQuotes,
    boolean ignoreLeadingWhiteSpace) {
  super(new StringReader("")); // dummy call to base constructor
  contentQueue = new ArrayDeque<>();
  TailerListener listener = new CsvContentListener(contentQueue);
  tailer = Tailer.create(source.file(), listener, DEFAULT_MONITOR_DELAY,
      false, true, 4096);
  this.parser = new CSVParser(separator, quoteChar, escape, strictQuotes,
      ignoreLeadingWhiteSpace);
  this.skipLines = line;
  try {
    // wait for tailer to capture data
    Thread.sleep(DEFAULT_MONITOR_DELAY);
  } catch (InterruptedException e) {
    throw new RuntimeException(e);
  }
}

protected LogTailer(File toRead, long timeForFileToBeCreated) throws IOException {
    long requiredTime = System.currentTimeMillis() + timeForFileToBeCreated;
    while ((System.currentTimeMillis() < requiredTime) && !(toRead.exists() && toRead.canRead())) {
        try {
            Thread.sleep(10);
        } catch (InterruptedException e) {
            // ignore
        }
    }
    if (!toRead.exists() || !toRead.canRead()) {
        throw new IllegalStateException("Couldn't read "+toRead.getCanonicalPath()+" in the configured timeout");
    }
    logger.debug("Initialising Tailer for "+toRead.getCanonicalPath());

    tailer = new Tailer(toRead, this, DELAY, false);
}

@Test
public void testShouldBeCallListenerHandleByLogging() throws Exception {
    Tailer.create(TestFileLoader.load("data/tailer/test.log"), listener, 100, true);

    ConcurrentTestFileWriter.writeTestData("data/tailer/test.log", "logging", 3);

    Thread.sleep(300);
    verify(mockCollector, times(3)).collect(anyString());
}

public void open(Map stormConf) {
  fileToParseName = fileToParse.getAbsolutePath();
  parsedLineQueue = new LinkedBlockingQueue<Map>(maxQueueSize);
  currentLine = 0;

  if (lineParser == null)
    lineParser = new NoOpLogLineParser();

  tailer = Tailer.create(fileToParse, this, tailerDelayMs);
  log.info("Started tailing "+fileToParseName);
}

public void run() {
  if (!fileToParse.isFile()) {
    log.warn("Skipping " + fileToParse.getAbsolutePath() + " because it doesn't exist anymore!");
    return;
  }

  lineNum = 0;
  batchOfDocs.clear();
  startMs = System.currentTimeMillis();

  if (logIndexer.tail) {
    tailer = new Tailer(fileToParse, this, logIndexer.tailerDelayMs);
    log.info("Tailing "+fileToParse+" with delay "+logIndexer.tailerDelayMs+" ms");

    logIndexer.tailerReaperBgThread.trackTailer(this);

    tailer.run(); // we're already in a thread, so just delegate to run
    log.info("Tailer stopped ... LogParser for " + fileName + " is done running.");
    logIndexer.onFinishedParsingFile(fileName, lineNum, skippedLines, System.currentTimeMillis() - startMs);
  } else {
    try {
      doParseFile(fileToParse);
    } catch (Exception exc) {
      log.error("Failed to process file '" + fileName + "' due to: " + exc, exc);
    }
  }
}

public void tail(TailerListener listener) {

        File file = new File(
                             fileLocation);
        Tailer tailer = Tailer.create(file, listener, 10);

        tailer.run();
    }

public void tail(TailerListener listener, String alias) {
    File file = logLookup.map(l -> l.lookup(alias))
                         .orElse(new File(
                                          fileLocation));
    Tailer tailer = Tailer.create(file, listener, 10);
    tailer.run();
}

@Autowired
public TailService(@Value("${geoip2.cityfile}") String cityFile,
        @Value("${access.logs}") String accessLogs, EventMessenger eventMessenger) {
    this.eventMessenger = eventMessenger;

    String databaseFile = cityFile;
    if (databaseFile != null) {
        Path database = Paths.get(databaseFile);
        if (Files.exists(database)) {
            try {
                this.reader = new DatabaseReader.Builder(database.toFile()).build();
            }
            catch (IOException e) {
                LoggerFactory.getLogger(getClass()).error("GeoIPCityService init", e);
            }
        }
    }

    this.tailers = new ArrayList<>();

    for (String logFile : accessLogs.split(",")) {
        Path p = Paths.get(logFile.trim());
        this.tailers.add(new Tailer(p.toFile(), new ListenerAdapter()));
    }

    this.executor = Executors.newFixedThreadPool(this.tailers.size());
    for (Tailer tailer : this.tailers) {
        this.executor.execute(tailer);
    }
}

@PreDestroy
public void preDestroy() {
    if (this.tailers != null) {
        for (Tailer tailer : this.tailers) {
            tailer.stop();
        }
    }

    if (this.executor != null) {
        this.executor.shutdown();
    }
}

private void initNetlogTailer() {

    if (netLogTailerThread != null) {
        netLogTailerThread.interrupt();
        netLogTailerThread = null;
    }

    if (activeNetLogFile != null) {
        TailerListener listener = new NetLogTailerListener();
        Tailer tailer = new Tailer(activeNetLogFile, listener, 2000);
        this.netLogTailerThread = new Thread(tailer);
        netLogTailerThread.setDaemon(true);
        netLogTailerThread.start();
    }
}

public Observable<String> getStream(final long pollIntervalMs) {
    return Observable.create(new OnSubscribeFunc<String>() {

        @Override
        public Subscription onSubscribe(
                final Observer<? super String> observer) {
            TailerListener listener = createListener(observer);
            final Tailer tailer = new Tailer(file, listener, pollIntervalMs);
            Thread t = new Thread(createRunnable(observer, tailer));
            t.start();
            return createSubscription(tailer);
        }
    });
}

private Runnable createRunnable(final Observer<? super String> observer,
        final Tailer tailer) {
    return new Runnable() {
        @Override
        public void run() {
            try {
                tailer.run();
            } catch (Exception e) {
                observer.onError(e);
            }
        }
    };
}

private Subscription createSubscription(final Tailer tailer) {
    return new Subscription() {
        @Override
        public void unsubscribe() {
            tailer.stop();
        }
    };
}

@Override
public synchronized void start() {

  logger.info("FileSource source {} starting.", getName());
  logger.info("FileSource source starting with file path: {}", filePath);

  this.queue = new LinkedList<String>();
  this.mutex = new Semaphore(1, true);

  /* Creates a single-threaded executor that can schedule commands
   * to run after a given delay */
  this.fileTailerExecutor = Executors.newSingleThreadScheduledExecutor();
  this.eventExtractorExecutor = Executors.newSingleThreadScheduledExecutor();

  final FileSourceEventProducer listener = new FileSourceEventProducer();

  this.tailer = new Tailer(new File(filePath), listener, delayMillis, startFromEnd);
  FileSourceEventConsumer eventExtractorCommand = new FileSourceEventConsumer();

  /**
   * Creates and executes a periodic action that becomes enabled first
   * after the given initial delay, and subsequently with the
   * given delay between the termination of one execution and the
   * commencement of the next.  If any execution of the task
   * encounters an exception, subsequent executions are suppressed.
   * Otherwise, the task will only terminate via cancellation or
   * termination of the executor. */
  this.fileTailerExecutor.scheduleWithFixedDelay(tailer, INITIAL_DELAY_MS, POLL_DELAY_MS, TimeUnit.MILLISECONDS);
  this.eventExtractorExecutor.scheduleWithFixedDelay(eventExtractorCommand, INITIAL_DELAY_MS, POLL_DELAY_MS, TimeUnit.MILLISECONDS);

  super.start();
  sourceCounter.start();
}

@Override
public synchronized void start() {
    log.info("Starting {} with {} ...", this);

    this.tailer = new Tailer(new File(filePath), new FlumeTailerSourceListener(this.getChannelProcessor()));
    this.tailer.run();
    super.start();
}

public void setLogFile(String pathname) {
    logFile = new File(pathname);

    tailerListener = new TailerListener();
    tailer = Tailer.create(logFile, tailerListener, 450, false);

    Thread thread = new Thread(tailer);
    thread.setDaemon(true);
    thread.start();

    if (logFile.exists() == false){
        logger.warn("Coud not open log file: " + pathname);
    }
}

@Autowired
public TailService(@Value("${geoip2.cityfile}") String cityFile,
        @Value("${access.logs}") String accessLogs,
        SimpMessageSendingOperations messagingTemplate) {
    this.messagingTemplate = messagingTemplate;

    String databaseFile = cityFile;
    if (databaseFile != null) {
        Path database = Paths.get(databaseFile);
        if (Files.exists(database)) {
            try {
                this.reader = new DatabaseReader.Builder(database.toFile()).build();
            }
            catch (IOException e) {
                LoggerFactory.getLogger(getClass()).error("GeoIPCityService init", e);
            }
        }
    }

    this.tailers = new ArrayList<>();

    for (String logFile : accessLogs.split(",")) {
        Path p = Paths.get(logFile.trim());
        this.tailers.add(new Tailer(p.toFile(), new ListenerAdapter()));
    }

    this.executor = Executors.newFixedThreadPool(this.tailers.size());
    for (Tailer tailer : this.tailers) {
        this.executor.execute(tailer);
    }
}

@PreDestroy
public void preDestroy() {
    if (this.tailers != null) {
        for (Tailer tailer : this.tailers) {
            tailer.stop();
        }
    }

    if (this.executor != null) {
        this.executor.shutdown();
    }
}

public void start()
{
    for (Log log : logs)
    {
        TailerListener listener = new MyTailerListener(log);
        executor.execute(new Tailer(log.getFile(), listener, DELAY, false, false));
    }
}

public TailCollector(Aggregator aggregator, TailCollectorConfig config) {
    this.aggregator = aggregator;
    this.tailer = new Tailer(new File(config.getPath()), new TailCollectorListener(this), 100, true);
    this.work = false;
}

@Override
public void init(Tailer tailer) {
    LOGGER.info("Init LogTailerListener");
}

@Override
public void init(Tailer tailer) {
    System.err.println("Starting error trapper with " + tailer.getFile().getName());
}

@Override
public void init(Tailer tailer) {
    running = true;
}

@Override
public void init(Tailer tailer) {
    running = true;
}

void startTailer() {
    this.tailer = new Tailer(this.logFile, this.createListener(), DELAY_MILLIS, true);
    ServerLogsExecutors.INSTANCE.execute(this.tailer);
}

private void setupTailer() {
    TailerListener listener = new LogTailer();
    File tailFile = new File(DebugLogger.LOGGING_FILENAME);
    tailer = Tailer.create(tailFile, new LogTailer(), 500, true);
}

@Override
public void init(Tailer tailer) {

    this.tailer = tailer;
}

public void start() {
    if (tailer == null) {
        tailer = Tailer.create(getLogsFile(), this, 500L, true);
        LOG.info("Logs watcher started. File {} will be tracked for changes.", configurationMediator.getLogFileName());
    }
}

@Override
public void init(Tailer tailer) {}

public static void test(final TestParams testParams) throws IOException, ClassNotFoundException,
    NoSuchMethodException, InvocationTargetException, IllegalAccessException, InterruptedException {

  final File tempFile = File.createTempFile("test", "test");
  final File labelFile = File.createTempFile("result", "result");
  LOG.info("Starting test, output file is {}, test config is \n{}", tempFile.getAbsolutePath(), testParams.toString());  
  Executors.newSingleThreadExecutor().submit(new Callable<Void>() {
    @Override
    public Void call() throws Exception {
      try {          
        Class.forName(testParams.getTaskClassName())
              .getMethod("main", String[].class)
              .invoke(null, (Object) String.format(
                  testParams.getCliStringTemplate(),
                  tempFile.getAbsolutePath(),
                  testParams.getInputInstances(),
                  testParams.getSamplingSize(),
                  testParams.getInputDelayMicroSec(),
                  labelFile.getAbsolutePath(),
                  testParams.getLabelSamplingSize()
                  ).split("[ ]"));
        } catch (Exception e) {
          LOG.error("Cannot execute test {} {}", e.getMessage(), e.getCause().getMessage());
        }
        return null;
      }
    });

  Thread.sleep(TimeUnit.SECONDS.toMillis(testParams.getPrePollWaitSeconds()));

  CountDownLatch signalComplete = new CountDownLatch(1);

  final Tailer tailer = Tailer.create(tempFile, new TestResultsTailerAdapter(signalComplete), 1000);
  new Thread(new Runnable() {
    @Override
    public void run() {
      tailer.run();
    }
  }).start();

  signalComplete.await();
  tailer.stop();

  assertResults(tempFile, testParams);
  if (testParams.getLabelFileCreated())
    assertLabels(labelFile, testParams);
}

@Override
public void init(Tailer tailer) {
    this.tailer = tailer;
}

@Override
public void init(Tailer tailer) {
}

@Override
public void init(final Tailer tailer)
{
    LOG.info("Tailing file {}", file.getAbsolutePath());
}

@Override
    public void init(Tailer tailer) {
        startupLatch.countDown();
//        logger.debug(System.currentTimeMillis()+": Started!");
    }

@Override
public void init(Tailer tailer)
{
    super.init(tailer);
}

public void start ()  {
    OutputFileListener listener = new OutputFileListener(this);
    tailer = Tailer.create(fileToTail, listener, 500);
}

public static void test(final TestParams testParams) throws IOException, ClassNotFoundException, NoSuchMethodException, InvocationTargetException, IllegalAccessException, InterruptedException {

        final File tempFile = File.createTempFile("test", "test");

        LOG.info("Starting test, output file is {}, test config is \n{}", tempFile.getAbsolutePath(), testParams.toString());

        Executors.newSingleThreadExecutor().submit(new Callable<Void>() {

            @Override
            public Void call() throws Exception {
                try {
                    Class.forName(testParams.getTaskClassName())
                            .getMethod("main", String[].class)
                            .invoke(null, (Object) String.format(
                                    testParams.getCliStringTemplate(),
                                    tempFile.getAbsolutePath(),
                                    testParams.getInputInstances(),
                                    testParams.getSamplingSize(),
                                    testParams.getInputDelayMicroSec()
                                    ).split("[ ]"));
                } catch (Exception e) {
                    LOG.error("Cannot execute test {} {}", e.getMessage(), e.getCause().getMessage());
                }
                return null;
            }
        });

        Thread.sleep(TimeUnit.SECONDS.toMillis(testParams.getPrePollWaitSeconds()));

        CountDownLatch signalComplete = new CountDownLatch(1);

        final Tailer tailer = Tailer.create(tempFile, new TestResultsTailerAdapter(signalComplete), 1000);
        new Thread(new Runnable() {
            @Override
            public void run() {
                tailer.run();
            }
        }).start();

        signalComplete.await();
        tailer.stop();

        assertResults(tempFile, testParams);
    }