commonlib/src/main/java/com/streamsets/pipeline/lib/io/MultiFileReader.java
public LiveFileChunk next(long waitMillis) {
    Utils.checkState(open, "Not open");
    waitMillis = (waitMillis > 0) ? waitMillis : 0;
    long startTime = System.currentTimeMillis();
    LiveFileChunk chunk = null;
    boolean exit = false;
    fileContextProvider.startNewLoop();
    while (!exit) {
      if (!fileContextProvider.didFullLoop()) {
        FileContext fileContext = fileContextProvider.next();
        try {
          LiveFileReader reader = fileContext.getReader();
          if (reader != null) {
            if (reader.hasNext()) {
              chunk = reader.next(0);
              if (LOG.isTraceEnabled()) {
                LOG.trace("next(): directory '{}', file '{}', offset '{}' got data '{}'",
                    fileContext.getMultiFileInfo().getFileFullPath(),
                    reader.getLiveFile(), reader.getOffset(), chunk != null);
              }
            } else {
              if (LOG.isTraceEnabled()) {
                LOG.trace("next(): directory '{}', file '{}', offset '{}' EOF reached",
                    fileContext.getMultiFileInfo().getFileFullPath(),
                    reader.getLiveFile(), reader.getOffset());
              }
            }
            fileContext.releaseReader(false);
          } else {
            if (LOG.isTraceEnabled()) {
              LOG.trace("next(): directory '{}', no reader available",
                  fileContext.getMultiFileInfo().getFileFullPath());
            }
          }
        } catch (IOException ex) {
          LOG.error("Error while reading file: {}", ex.toString(), ex);
          try {
            fileContext.releaseReader(true);
          } catch (IOException ex1) {
            LOG.warn("Error while releasing reader in error: {}", ex1.toString(), ex1);
          }
        }
      }

      // check exit conditions (we have a chunk, or we timed-out waitMillis)
      exit = chunk != null;
      if (!exit) {
        // if we looped thru all dir contexts in this call we yield CPU
        if (fileContextProvider.didFullLoop()) {
          exit = isTimeout(startTime, waitMillis);
          if (!exit && LOG.isTraceEnabled()) {
            LOG.trace("next(): looped through all directories, yielding CPU");
          }
          exit = exit || !ThreadUtil.sleep(Math.min(getRemainingWaitTime(startTime, waitMillis), MAX_YIELD_TIME));
          fileContextProvider.startNewLoop();
        }
      }
    }
    return chunk;
  }
public LiveFileChunk next(long waitMillis) {
    Utils.checkState(open, "Not open");
    waitMillis = (waitMillis > 0) ? waitMillis : 0;
    long startTime = System.currentTimeMillis();
    LiveFileChunk chunk = null;
    boolean exit = false;
    fileContextProvider.startNewLoop();
    while (!exit) {
      if (!fileContextProvider.didFullLoop()) {
        FileContext fileContext = fileContextProvider.next();
        try {
          LiveFileReader reader = fileContext.getReader();
          if (reader != null) {
            if (reader.hasNext()) {
              chunk = reader.next(0);
              if (LOG.isTraceEnabled()) {
                LOG.trace("next(): directory '{}', file '{}', offset '{}' got data '{}'",
                    fileContext.getMultiFileInfo().getFileFullPath(),
                    reader.getLiveFile(), reader.getOffset(), chunk != null);
              }
            } else {
              if (LOG.isTraceEnabled()) {
                LOG.trace("next(): directory '{}', file '{}', offset '{}' EOF reached",
                    fileContext.getMultiFileInfo().getFileFullPath(),
                    reader.getLiveFile(), reader.getOffset());
              }
            }
            fileContext.releaseReader(false);
          } else {
            if (LOG.isTraceEnabled()) {
              LOG.trace("next(): directory '{}', no reader available",
                  fileContext.getMultiFileInfo().getFileFullPath());
            }
          }
        } catch (IOException ex) {
          LOG.error("Error while reading file: {}", ex.toString(), ex);
          try {
            fileContext.releaseReader(true);
          } catch (IOException ex1) {
            LOG.warn("Error while releasing reader in error: {}", ex1.toString(), ex1);
          }
        }
      }

      // check exit conditions (we have a chunk, or we timed-out waitMillis)
      exit = chunk != null;
      if (!exit) {
        // if we looped thru all dir contexts in this call we yield CPU
        if (fileContextProvider.didFullLoop()) {
          exit = isTimeout(startTime, waitMillis);
          if (!exit && LOG.isTraceEnabled()) {
            LOG.trace("next(): looped through all directories, yielding CPU");
          }
          exit = exit || !ThreadUtil.sleep(Math.min(getRemainingWaitTime(startTime, waitMillis), MAX_YIELD_TIME));
          fileContextProvider.startNewLoop();
        }
      }
    }
    return chunk;
  }
public LiveFileChunk next ( long waitMillis ) { Utils . checkState ( open , "Not open" ) ; waitMillis = ( waitMillis > 0 ) ? waitMillis : 0 ; long startTime = System . currentTimeMillis ( ) ; LiveFileChunk chunk = null ; boolean exit = false ; fileContextProvider . startNewLoop ( ) ; while ( ! exit ) { if ( ! fileContextProvider . didFullLoop ( ) ) { FileContext fileContext = fileContextProvider . next ( ) ; try { LiveFileReader reader = fileContext . getReader ( ) ; if ( reader != null ) { if ( reader . hasNext ( ) ) { chunk = reader . next ( 0 ) ; if ( LOG . isTraceEnabled ( ) ) { LOG . trace ( "next(): directory '{}', file '{}', offset '{}' got data '{}'" , fileContext . getMultiFileInfo ( ) . getFileFullPath ( ) , reader . getLiveFile ( ) , reader . getOffset ( ) , chunk != null ) ; } } else { if ( LOG . isTraceEnabled ( ) ) { LOG . trace ( "next(): directory '{}', file '{}', offset '{}' EOF reached" , fileContext . getMultiFileInfo ( ) . getFileFullPath ( ) , reader . getLiveFile ( ) , reader . getOffset ( ) ) ; } } fileContext . releaseReader ( false ) ; } else { if ( LOG . isTraceEnabled ( ) ) { LOG . trace ( "next(): directory '{}', no reader available" , fileContext . getMultiFileInfo ( ) . getFileFullPath ( ) ) ; } } } catch ( IOException ex ) { LOG . error ( "Error while reading file: {}" , ex . toString ( ) , ex ) ; try { fileContext . releaseReader ( true ) ; } catch ( IOException ex1 ) { LOG . warn ( "Error while releasing reader in error: {}" , ex1 . toString ( ) , ex1 ) ; } } } // check exit conditions (we have a chunk, or we timed-out waitMillis) exit = chunk != null ; if ( ! exit ) { // if we looped thru all dir contexts in this call we yield CPU if ( fileContextProvider . didFullLoop ( ) ) { exit = isTimeout ( startTime , waitMillis ) ; if ( ! exit && LOG . isTraceEnabled ( ) ) { LOG . trace ( "next(): looped through all directories, yielding CPU" ) ; } exit = exit || ! ThreadUtil . sleep ( Math . min ( getRemainingWaitTime ( startTime , waitMillis ) , MAX_YIELD_TIME ) ) ; fileContextProvider . startNewLoop ( ) ; } } } return chunk ; }
Reads the next {@link LiveFileChunk} from the directories waiting the specified time for one.

@param waitMillis number of milliseconds to block waiting for a chunk.
@return the next chunk, or <code>null</code> if there is no next chunk and the waiting time passed.
Reads the next { @link LiveFileChunk } from the directories waiting the specified time for one .