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;
  }
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.