기업 검색엔진 개발의 연결기connector(27)

/** 
   *    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());
    }
  }