private final DocumentSnapshotRepositoryMonitorManager      snapshotRepositoryMonitorManager;  private final TraversalContextManager traversalContextManager;  /**   * Boolean to mark TraversalManager as invalid.   * It‘s possible for Connector Manager to keep a reference to   * an outdated TraversalManager (after a new one has been given   * previous TraversalManagers are invalid to use).   */  private boolean isActive = true;  /**   * Creates a {@link DiffingConnectorTraversalManager}.   *   * @param snapshotRepositoryMonitorManager the   *        {@link DocumentSnapshotRepositoryMonitorManager}   *        for use accessing a {@link ChangeSource}   * @param traversalContextManager {@link TraversalContextManager}   *        that holds the current {@link TraversalContext}   */  public DiffingConnectorTraversalManager(      DocumentSnapshotRepositoryMonitorManager snapshotRepositoryMonitorManager,      TraversalContextManager traversalContextManager) {    this.snapshotRepositoryMonitorManager = snapshotRepositoryMonitorManager;    this.traversalContextManager = traversalContextManager;  }

/* @Override */  public synchronized DocumentList resumeTraversal(String checkpoint)      throws RepositoryException {    /* Exhaustive list of method‘s use:     resumeTraversal(null) from startTraversal:       monitors get started from null     resumeTraversal(null) from Connector Manager sometime after startTraversal:       monitors already started from previous resumeTraversal call     resumeTraversal(cp) from Connector Manager without a startTraversal:       means there was a shutdown or turn off       monitors get started from cp; should use state     resumeTraversal(cp) from Connector Manager sometime after some uses:       is most common case; roll    */    if (isActive()) {        //启动snapshotRepositoryMonitorManager      if (!snapshotRepositoryMonitorManager.isRunning()) {        snapshotRepositoryMonitorManager.start(checkpoint);      }      return newDocumentList(checkpoint);    } else {      throw new RepositoryException(          "Inactive FileTraversalManager referanced.");    }  }

private DocumentList newDocumentList(String checkpoint)      throws RepositoryException {    //获取队列 CheckpointAndChangeQueue(队列 CheckpointAndChangeQueue只由snapshotRepositoryMonitorManager引用)    CheckpointAndChangeQueue checkpointAndChangeQueue =        snapshotRepositoryMonitorManager.getCheckpointAndChangeQueue();    try {      DiffingConnectorDocumentList documentList = new DiffingConnectorDocumentList(          checkpointAndChangeQueue,          CheckpointAndChangeQueue.initializeCheckpointStringIfNull(              checkpoint));      //Map<String, MonitorCheckpoint>      Map<String, MonitorCheckpoint> guaranteesMade =          checkpointAndChangeQueue.getMonitorRestartPoints();            snapshotRepositoryMonitorManager.acceptGuarantees(guaranteesMade);      return new ConfirmActiveDocumentList(documentList);    } catch (IOException e) {      throw new RepositoryException("Failure when making DocumentList.", e);    }  }

/** * An implementation of {@link DocumentList} for the {@link DiffingConnector}. * * @since 2.8 */public class DiffingConnectorDocumentList implements DocumentList {  private final Iterator<CheckpointAndChange> checkpointAndChangeIterator;  private String checkpoint;  /**   * Creates a document list that returns a batch of documents from the provided   * {@link CheckpointAndChangeQueue}.   *   * @param queue a CheckpointAndChangeQueue containing document changes   * @param checkpoint point into the change queue after which to start   *        returning documents   * @throws IOException if persisting fails   */  public DiffingConnectorDocumentList(CheckpointAndChangeQueue queue,      String checkpoint) throws IOException {      //CheckpointAndChangeQueue queued的resume方法获取List<CheckpointAndChange>      //本DocumentList批次数据已经加载于内存    List<CheckpointAndChange> guaranteedChanges = queue.resume(checkpoint);    checkpointAndChangeIterator = guaranteedChanges.iterator();    this.checkpoint = checkpoint;  }    /**   * 调用方获取该状态并持久化，迭代完毕即为最后的checkpoint   */  /* @Override */  public String checkpoint() {    return checkpoint;  }  /* @Override */  public Document nextDocument() throws RepositoryException {    if (checkpointAndChangeIterator.hasNext()) {      CheckpointAndChange checkpointAndChange =        checkpointAndChangeIterator.next();      //更新checkpoint      checkpoint = checkpointAndChange.getCheckpoint().toString();      return checkpointAndChange.getChange().getDocumentHandle().getDocument();    } else {      return null;    }  }}

//Map<String, MonitorCheckpoint>      Map<String, MonitorCheckpoint> guaranteesMade =          checkpointAndChangeQueue.getMonitorRestartPoints();

 /**   * 监视器管理对象收到CheckpointAndChangeQueue对象反馈，分发给对应的监视器处理MonitorCheckpoint   */  /* @Override */  public void acceptGuarantees(Map<String, MonitorCheckpoint> guarantees) {    for (Map.Entry<String, MonitorCheckpoint> entry : guarantees.entrySet()) {      String monitorName = entry.getKey();      MonitorCheckpoint checkpoint = entry.getValue();      DocumentSnapshotRepositoryMonitor monitor = fileSystemMonitorsByName.get(monitorName);      if (monitor != null) {        // Signal is asynch.  Let monitor figure out how to use.          //回调        monitor.acceptGuarantee(checkpoint);      }    }  }

/**   * 监视器收到反馈 [MonitorCheckpoint接收确认]   * @param cp   */  // Public for DocumentSnapshotRepositoryMonitorTest  @VisibleForTesting  public void acceptGuarantee(MonitorCheckpoint cp) {    snapshotStore.acceptGuarantee(cp);    guaranteeCheckpoint = cp;  }

/**   * 反馈MonitorCheckpoint处理   * @param cp   */  void acceptGuarantee(MonitorCheckpoint cp) {    long readSnapshotNumber = cp.getSnapshotNumber();    if (readSnapshotNumber < 0) {      throw new IllegalArgumentException("Received invalid snapshot in: " + cp);    }    if (oldestSnapshotToKeep > readSnapshotNumber) {      LOG.warning("Received an older snapshot than " + oldestSnapshotToKeep + ": " + cp);    } else {      oldestSnapshotToKeep = readSnapshotNumber;    }  }