企业搜索引擎开发之连接器connector（二十七）

 * A source of {@link Change} objects.
 *
 * @since 2.8
 */
public interface ChangeSource {
  /**
   * @return the next change, or {@code null} if there is no change available
   */
  public Change getNextChange();
}

/**
   * 初始化阻塞队列pendingChanges
   * @param size
   * @param sleepInterval
   * @param introduceDelayAfterEachScan
   * @param activityLogger
   */
  private ChangeQueue(int size, long sleepInterval, 
      boolean introduceDelayAfterEachScan, CrawlActivityLogger activityLogger) {
    pendingChanges = new ArrayBlockingQueue<Change>(size);
    this.sleepInterval = sleepInterval;
    this.activityLogger = activityLogger;
    this.introduceDelayAfterEveryScan = introduceDelayAfterEachScan;
  }

/**
   * 获取阻塞队列pendingChanges元素
   * Gets the next available change from the ChangeQueue.  Will wait up to
   * 1/4 second for a change to appear if none is immediately available.
   *
   * @return the next available change, or {@code null} if no changes are
   *         available
   */
  public Change getNextChange() {
    try {
      return pendingChanges.poll(250L, TimeUnit.MILLISECONDS);
    } catch (InterruptedException ie) {
      return null;
    }
  }

 private final AtomicInteger maximumQueueSize =
      new AtomicInteger(DEFAULT_MAXIMUM_QUEUE_SIZE);
  private final List<CheckpointAndChange> checkpointAndChangeList;
  private final ChangeSource changeSource;
  private final DocumentHandleFactory internalDocumentHandleFactory;
  private final DocumentHandleFactory clientDocumentHandleFactory;

  private volatile DiffingConnectorCheckpoint lastCheckpoint;
  private final File persistDir;  // place to persist enqueued values
  private MonitorRestartState monitorPoints = new MonitorRestartState();

public CheckpointAndChangeQueue(ChangeSource changeSource, File persistDir,
      DocumentHandleFactory internalDocumentHandleFactory,
      DocumentHandleFactory clientDocumentHandleFactory) {
    this.changeSource = changeSource;
    this.checkpointAndChangeList
        = Collections.synchronizedList(
            new ArrayList<CheckpointAndChange>(maximumQueueSize.get()));
    this.persistDir = persistDir;
    this.internalDocumentHandleFactory = internalDocumentHandleFactory;
    this.clientDocumentHandleFactory = clientDocumentHandleFactory;
    ensurePersistDirExists();
  }

 /**
   * 从ChangeSource拉取Change，加入checkpointAndChangeList
   */
  private void loadUpFromChangeSource() {
    int max = maximumQueueSize.get();
    if (checkpointAndChangeList.size() < max) {
      lastCheckpoint = lastCheckpoint.nextMajor();
    }   
    while (checkpointAndChangeList.size() < max) {
      Change newChange = changeSource.getNextChange();
      if (newChange == null) {
        break;
      }
      lastCheckpoint = lastCheckpoint.next();
      checkpointAndChangeList.add(new CheckpointAndChange(
          lastCheckpoint, newChange));      
    }
  }

/**
   * 获取List<CheckpointAndChange>队列
   * Returns an {@link Iterator} for currently available
   * {@link CheckpointAndChange} objects that occur after the passed in
   * checkpoint. The {@link String} form of a {@link DiffingConnectorCheckpoint}
   * passed in is produced by calling
   * {@link DiffingConnectorCheckpoint#toString()}. As a side effect, Objects
   * up to and including the object with the passed in checkpoint are removed
   * from this queue.
   *
   * @param checkpointString null means return all {@link CheckpointAndChange}
   *        objects and a non null value means to return
   *        {@link CheckpointAndChange} objects with checkpoints after the
   *        passed in value.
   * @throws IOException if error occurs while manipulating recovery state
   */
  synchronized List<CheckpointAndChange> resume(String checkpointString)
      throws IOException {
      //移除已完成队列
    removeCompletedChanges(checkpointString);
    //从ChangeSource拉取Change，加入checkpointAndChangeList
    loadUpFromChangeSource();
    //更新monitorPoints
    monitorPoints.updateOnGuaranteed(checkpointAndChangeList);
    try {
        //持久化checkpointAndChangeList到队列文件
        //一次resume即生成一文件
      writeRecoveryState();
    } finally {
      // TODO: Enahnce with mechanism that remembers
      // information about recovery files to avoid re-reading.
        //移除冗余的队列文件 （已经消费完成的）
      removeExcessRecoveryState();
    }
    return getList();
  }

/** 
   * 持久化json队列
   * @throws IOException
   */
  private void writeRecoveryState() throws IOException {
    // TODO(pjo): Move this method into RecoveryFile.
    File recoveryFile = new RecoveryFile(persistDir);
    FileOutputStream outStream = new FileOutputStream(recoveryFile);
    Writer writer = new OutputStreamWriter(outStream, Charsets.UTF_8);
    try {
      try {
        writeJson(writer);
      } catch (JSONException e) {
        throw IOExceptionHelper.newIOException("Failed writing recovery file.", e);
      }
      writer.flush();
      outStream.getFD().sync();
    } finally {
      writer.close();
    }
  }

/** 
   * 可用于比较时间的队列文件
   * A File that has some of the recovery logic. 
   *  Original recovery files‘ names contained a single nanosecond timestamp,
   *  eg.  recovery.10220010065599398 .  These turned out to be flawed
   *  because nanosecond times can go "back in time" between JVM restarts.
   *  Updated recovery files‘ names contain a wall clock millis timestamp 
   *  followed by an underscore followed by a nanotimestamp, eg.
   *  recovery.702522216012_10220010065599398 .
   */
  static class RecoveryFile extends File {
    final static long NO_TIME_AVAIL = -1;
    long milliTimestamp = NO_TIME_AVAIL;
    long nanoTimestamp;

    long parseTime(String s) throws IOException {
      try {
        return Long.parseLong(s);
      } catch(NumberFormatException e) {
        throw new LoggingIoException("Invalid recovery filename: "
            + getAbsolutePath());
      }
    }
    
    /**
     * 解析文件名称中包含的时间
     * @throws IOException
     */
    void parseOutTimes() throws IOException {
      try {
        String basename = getName();
        if (!basename.startsWith(RECOVERY_FILE_PREFIX)) {
          throw new LoggingIoException("Invalid recovery filename: "
              + getAbsolutePath());
        } else {
          String extension = basename.substring(RECOVERY_FILE_PREFIX.length());
          if (!extension.contains("_")) {  // Original name format.
            nanoTimestamp = parseTime(extension);
          } else {  // Updated name format.
            String timeParts[] = extension.split("_");
            if (2 != timeParts.length) {
              throw new LoggingIoException("Invalid recovery filename: "
                  + getAbsolutePath());
            }
            milliTimestamp = parseTime(timeParts[0]);
            nanoTimestamp = parseTime(timeParts[1]);
          }
        }
      } catch(IndexOutOfBoundsException e) {
        throw new LoggingIoException("Invalid recovery filename: "
            + getAbsolutePath());
      }
    }

    RecoveryFile(File persistanceDir) throws IOException {
      super(persistanceDir, RECOVERY_FILE_PREFIX + System.currentTimeMillis()
          + "_" + System.nanoTime());
      parseOutTimes();
    }
    
    /**
     * 该构造函数用于先获得文件绝对路径
     * @param absolutePath
     * @throws IOException
     */
    RecoveryFile(String absolutePath) throws IOException {
      super(absolutePath);
      parseOutTimes();
    }

    boolean isOlder(RecoveryFile other) {
      boolean weHaveMillis = milliTimestamp != NO_TIME_AVAIL;
      boolean otherHasMillis = other.milliTimestamp != NO_TIME_AVAIL;
      boolean bothHaveMillis = weHaveMillis && otherHasMillis;
      boolean neitherHasMillis = (!weHaveMillis) && (!otherHasMillis);
      if (bothHaveMillis) {
        if (this.milliTimestamp < other.milliTimestamp) {
          return true;
        } else if (this.milliTimestamp > other.milliTimestamp) {
          return false;
        } else {
          return this.nanoTimestamp < other.nanoTimestamp;
        }
      } else if (neitherHasMillis) {
        return this.nanoTimestamp < other.nanoTimestamp;
      } else if (weHaveMillis) {  // and other doesn‘t; we are newer.
        return false;
      } else {  // other has millis; other is newer.
        return true;
      }
    }
    
    /** A delete method that logs failures. */
    /**
     * 删除文件
     */
    public void logOnFailDelete() {
      boolean deleted = super.delete();
      if (!deleted) {
        LOG.severe("Failed to delete: " + getAbsolutePath());
      }
    }
    // TODO(pjo): Move more recovery logic into this class.
  }

 /**
   * Initialize to start processing from after the passed in checkpoint
   * or from the beginning if the passed in checkpoint is null.  Part of
   * making DocumentSnapshotRepositoryMonitorManager go from "cold" to "warm".
   */
  public synchronized void start(String checkpointString) throws IOException {
    LOG.info("Starting CheckpointAndChangeQueue from " + checkpointString);
    //创建队列目录
    ensurePersistDirExists();
    checkpointAndChangeList.clear();
    lastCheckpoint = constructLastCheckpoint(checkpointString);
    if (null == checkpointString) {
        //删除队列文件
      removeAllRecoveryState();
    } else {
      RecoveryFile current = removeExcessRecoveryState();
      //加载monitorPoints和checkpointAndChangeList队列
      loadUpFromRecoveryState(current);
      //this.monitorPoints.points.entrySet();
      
    }
  }

/**
   * 加载队列
   * @param file
   * @throws IOException
   */
  private void loadUpFromRecoveryState(RecoveryFile file) throws IOException {
    // TODO(pjo): Move this method into RecoveryFile.
    new LoadingQueueReader().readJson(file);
  }

/**
   * 从json文件加载队列抽象类
   * Reads JSON recovery files. Uses the Template Method pattern to
   * delegate what to do with the parsed objects to subclasses.
   *
   * Note: This class uses gson for streaming support.
   */
  private abstract class AbstractQueueReader {
    public void readJson(File file) throws IOException {
      readJson(new BufferedReader(new InputStreamReader(
                  new FileInputStream(file), Charsets.UTF_8)));
    }

    /**
     * Reads and parses the stream, calling the abstract methods to
     * take whatever action is required. The given stream will be
     * closed automatically.
     *
     * @param reader the stream to parse
     */
    @VisibleForTesting
    void readJson(Reader reader) throws IOException {
      JsonReader jsonReader = new JsonReader(reader);
      try {
        readJson(jsonReader);
      } finally {
        jsonReader.close();
      }
    }

    /**
     * Reads and parses the stream, calling the abstract methods to
     * take whatever action is required.
     */
    private void readJson(JsonReader reader) throws IOException {
      JsonParser parser = new JsonParser();
      reader.beginObject();
      while (reader.hasNext()) {
        String name = reader.nextName();
        if (name.equals(MONITOR_STATE_JSON_TAG)) {
          readMonitorPoints(parser.parse(reader));
        } else if (name.equals(QUEUE_JSON_TAG)) {
          reader.beginArray();
          while (reader.hasNext()) {
            readCheckpointAndChange(parser.parse(reader));
          }
          reader.endArray();
        } else {
          throw new IOException("Read invalid recovery file.");
        }
      }
      reader.endObject();

      reader.setLenient(true);
      String name = reader.nextString();
      if (!name.equals(SENTINAL)) {
        throw new IOException("Read invalid recovery file.");
      }
    }

    protected abstract void readMonitorPoints(JsonElement gson)
        throws IOException;

    protected abstract void readCheckpointAndChange(JsonElement gson)
        throws IOException;
  }

/**
   * 检测队列文件的有效性
   * Verifies that a JSON recovery file is valid JSON with a
   * trailing sentinel.
   */
  private class ValidatingQueueReader extends AbstractQueueReader {
    protected void readMonitorPoints(JsonElement gson) throws IOException {
    }

    protected void readCheckpointAndChange(JsonElement gson)
        throws IOException {
    }
  }
   
  /**
   * 从json文件加载队列实现类
   */
  /** Loads the queue from a JSON recovery file. */
  /*
   * TODO(jlacey): Change everything downstream to gson. For now, we
   * reserialize the individual gson objects and deserialize them
   * using org.json.
   */
  @VisibleForTesting
  class LoadingQueueReader extends AbstractQueueReader {
    /**
     * 加载MonitorRestartState checkpoint(HashMap<String, MonitorCheckpoint> points)
     */
    protected void readMonitorPoints(JsonElement gson) throws IOException {
      try {
        JSONObject json = gsonToJson(gson);
        monitorPoints = new MonitorRestartState(json);
        //monitorPoints.updateOnGuaranteed(checkpointAndChangeList)
      } catch (JSONException e) {
        throw IOExceptionHelper.newIOException(
            "Failed reading persisted JSON queue.", e);
      }
    }
    
    /**
     * 加载checkpointAndChangeList
     */
    protected void readCheckpointAndChange(JsonElement gson)
        throws IOException {
      try {
        JSONObject json = gsonToJson(gson);
        checkpointAndChangeList.add(new CheckpointAndChange(json,
            internalDocumentHandleFactory, clientDocumentHandleFactory));
      } catch (JSONException e) {
        throw IOExceptionHelper.newIOException(
            "Failed reading persisted JSON queue.", e);
      }
    }

    // TODO(jlacey): This could be much more efficient, especially
    // with LOBs, if we directly transformed the objects with a little
    // recursive parser. This code is only used when recovering failed
    // batches, so I don‘t know if that‘s worth the effort.
    private JSONObject gsonToJson(JsonElement gson) throws JSONException {
      return new JSONObject(gson.toString());
    }
  }