aggregatelist.java

pastry的java实现的2.0b版

            Id adcKey = factory.buildIdFromToString(parts[4]);
            long lifetime = Long.parseLong(parts[5]);
            AggregateDescriptor adc = (AggregateDescriptor) aggregateList.get(adcKey);
            if (adc == null) {
              throw new AggregationException("Cannot find aggregate (" + adcKey.toStringFull() + ": " + line);
            }

            adc.currentLifetime = lifetime;
            for (int i = 0; i < adc.objects.length; i++) {
              adc.objects[i].currentLifetime = adc.objects[i].refreshedLifetime;
            }
            if (logger.level <= Logger.FINER) {
              logger.log(
                " - refresh=" + lifetime + " in ADC " + adcKey.toStringFull());
            }
          } else if (parts[3].equals("removeAggregate")) {
            Id adcKey = factory.buildIdFromToString(parts[4]);
            AggregateDescriptor adc = (AggregateDescriptor) aggregateList.get(adcKey);
            if (adc == null) {
              throw new AggregationException("Cannot find aggregate (" + adcKey.toStringFull() + ": " + line);
            }

            removeAggregateDescriptor(adc, false);
            if (logger.level <= Logger.FINER) {
              logger.log(
                " - remove ADC " + adcKey.toStringFull());
            }
          } else if (parts[3].equals("addAggregate")) {
            Id adcKey = factory.buildIdFromToString(parts[4]);
            AggregateDescriptor adc = (AggregateDescriptor) aggregateList.get(adcKey);
            if (adc != null) {
              throw new AggregationException("Aggregate already present (" + adcKey.toStringFull() + ": " + line);
            }

            adc = readAggregate(logFile, adcKey);
            addAggregateDescriptor(adc, false);
            if (logger.level <= Logger.FINER) {
              logger.log(
                " - add ADC " + adcKey.toStringFull());
            }
          } else {
            throw new AggregationException("Unknown command (" + parts[3] + "): " + line);
          }

          nextSerial++;
        } else {
          /*
           *  skip addAggregate if necessary
           */
          if (parts[3].equals("addAggregate")) {
            readAggregate(logFile, factory.buildIdFromToString(parts[4]));
          }
        }
      }

    } catch (FileNotFoundException fnfe) {
      if (logger.level <= Logger.WARNING) {
        logger.log(
          "No aggregate log found; using configuration file only");
      }
    } catch (Exception e) {
      if (logger.level <= Logger.WARNING) {
        logger.logException(
          "Exception while recovering aggregate log: ", e);
      }
      System.exit(1);
    }

    if (entriesReplayed > 0) {
      writeToDisk();
      if (logger.level <= Logger.WARNING) {
        logger.log(
          entriesReplayed + " entries replayed from aggregate log");
      }
    }
  }

  /**
   * DESCRIBE THE METHOD
   *
   * @param entry DESCRIBE THE PARAMETER
   */
  public void logEntry(String entry) {
    if (loggingEnabled) {
      if (logFile != null) {
        logFile.println("$|" + (nextSerial++) + "|" + environment.getTimeSource().currentTimeMillis() + "|" + entry + "|@");
        logFile.flush();
      } else {
        if (logger.level <= Logger.WARNING) {
          logger.log(
            "Aggregation cannot write to log: " + entry);
        }
      }
    }
  }

  /**
   * DESCRIBE THE METHOD
   */
  public void clear() {
    aggregateList.clear();
    rootKey = null;
  }

  /**
   * DESCRIBE THE METHOD
   */
  public void resetMarkers() {
    Enumeration enumerationeration = aggregateList.elements();
    while (enumerationeration.hasMoreElements()) {
      AggregateDescriptor aggr = (AggregateDescriptor) enumerationeration.nextElement();
      aggr.marker = false;
    }
  }

  /**
   * DESCRIBE THE METHOD
   *
   * @param adc DESCRIBE THE PARAMETER
   * @param lifetime DESCRIBE THE PARAMETER
   */
  public void refreshAggregate(AggregateDescriptor adc, long lifetime) {
    logEntry("refresh|" + adc.key.toStringFull() + "|" + lifetime);
    adc.currentLifetime = lifetime;
    for (int i = 0; i < adc.objects.length; i++) {
      adc.objects[i].currentLifetime = adc.objects[i].refreshedLifetime;
    }
  }

  /**
   * Adds a feature to the AggregateDescriptor attribute of the AggregateList
   * object
   *
   * @param aggr The feature to be added to the AggregateDescriptor attribute
   */
  public void addAggregateDescriptor(AggregateDescriptor aggr) {
    addAggregateDescriptor(aggr, true);
  }

  /**
   * Adds a feature to the AggregateDescriptor attribute of the AggregateList
   * object
   *
   * @param aggr The feature to be added to the AggregateDescriptor attribute
   * @param logThis The feature to be added to the AggregateDescriptor attribute
   */
  private void addAggregateDescriptor(AggregateDescriptor aggr, boolean logThis) {
    if ((logFile != null) && logThis && loggingEnabled) {
      logEntry("addAggregate|" + aggr.key.toStringFull());
      writeAggregate(logFile, aggr);
      logFile.flush();
    }

    aggregateList.put(aggr.key, aggr);

    for (int i = 0; i < aggr.objects.length; i++) {
      aggregateList.put(new VersionKey(aggr.objects[i].key, aggr.objects[i].version), aggr);
      AggregateDescriptor prevDesc = (AggregateDescriptor) aggregateList.get(aggr.objects[i].key);
      int objDescIndex = (prevDesc == null) ? -1 : prevDesc.lookupNewest(aggr.objects[i].key);
      if ((objDescIndex < 0) || (prevDesc.objects[objDescIndex].version <= aggr.objects[i].version)) {
        aggregateList.put(aggr.objects[i].key, aggr);
      }
    }

    for (int i = 0; i < aggr.pointers.length; i++) {
      AggregateDescriptor ref = (AggregateDescriptor) aggregateList.get(aggr.pointers[i]);
      if (ref != null) {
        ref.addReference();
      }
    }
  }

  /**
   * DESCRIBE THE METHOD
   *
   * @param aggr DESCRIBE THE PARAMETER
   */
  public void removeAggregateDescriptor(AggregateDescriptor aggr) {
    removeAggregateDescriptor(aggr, true);
  }

  /**
   * DESCRIBE THE METHOD
   *
   * @param aggr DESCRIBE THE PARAMETER
   * @param logThis DESCRIBE THE PARAMETER
   */
  private void removeAggregateDescriptor(AggregateDescriptor aggr, boolean logThis) {
    if (logThis) {
      logEntry("removeAggregate|" + aggr.key.toStringFull());
    }

    aggregateList.remove(aggr.key);

    for (int i = 0; i < aggr.objects.length; i++) {
      VersionKey vkey = new VersionKey(aggr.objects[i].key, aggr.objects[i].version);
      AggregateDescriptor prevDesc1 = (AggregateDescriptor) aggregateList.get(vkey);
      if ((prevDesc1 != null) && prevDesc1.key.equals(aggr.key)) {
        aggregateList.remove(vkey);
      }
      AggregateDescriptor prevDesc2 = (AggregateDescriptor) aggregateList.get(aggr.objects[i].key);
      if ((prevDesc2 != null) && prevDesc2.key.equals(aggr.key)) {
        aggregateList.remove(aggr.objects[i].key);
      }
    }

    if (aggregateList.containsValue(aggr)) {
      if (logger.level <= Logger.WARNING) {
        logger.log(
          "Removal from aggregate list incomplete: " + aggr.key.toStringFull());
      }
    }
  }

  /**
   * DESCRIBE THE METHOD
   *
   * @param excludes DESCRIBE THE PARAMETER
   */
  public void recalculateReferenceCounts(Id[] excludes) {
    Enumeration enumeration = aggregateList.elements();
    while (enumeration.hasMoreElements()) {
      AggregateDescriptor aggr = (AggregateDescriptor) enumeration.nextElement();
      aggr.referenceCount = 0;
      aggr.marker = false;

      /*
       *  References from excluded aggregates don't count
       */
      if (excludes != null) {
        for (int i = 0; i < excludes.length; i++) {
          if (excludes[i].equals(aggr.key)) {
            aggr.marker = true;
          }
        }
      }
    }

    enumeration = aggregateList.elements();
    while (enumeration.hasMoreElements()) {
      AggregateDescriptor aggr = (AggregateDescriptor) enumeration.nextElement();
      if (!aggr.marker) {
        aggr.marker = true;
        for (int i = 0; i < aggr.pointers.length; i++) {
          AggregateDescriptor ref = (AggregateDescriptor) aggregateList.get(aggr.pointers[i]);
          if (ref != null) {
            ref.addReference();
          }
        }
      }
    }
  }

  /**
   * DESCRIBE THE METHOD
   *
   * @param br DESCRIBE THE PARAMETER
   * @return DESCRIBE THE RETURN VALUE
   * @exception IOException DESCRIBE THE EXCEPTION
   */
  private String readLineSkipComments(BufferedReader br) throws IOException {
    while (true) {
      String line = br.readLine();
      if ((line != null) && ((line.length() == 0) || (line.charAt(0) == '#'))) {
        continue;
      }
      return line;
    }
  }

  /**
   * DESCRIBE THE METHOD
   *
   * @param reader DESCRIBE THE PARAMETER
   * @param aggrKey DESCRIBE THE PARAMETER
   * @return DESCRIBE THE RETURN VALUE
   * @exception AggregationException DESCRIBE THE EXCEPTION
   * @exception IOException DESCRIBE THE EXCEPTION
   */
  private AggregateDescriptor readAggregate(BufferedReader reader, Id aggrKey) throws AggregationException, IOException {
    String[] expiresS = readLineSkipComments(reader).split("=");
    if (!expiresS[0].equals("expires")) {
      throw new AggregationException("Cannot find expiration date: " + expiresS[0]);
    }
    long expires = Long.parseLong(expiresS[1]);

    String[] objectNumS = readLineSkipComments(reader).split("=");
    if (!objectNumS[0].equals("objects")) {
      throw new AggregationException("Cannot find number of objects: " + objectNumS[0]);
    }
    int numObjects = Integer.parseInt(objectNumS[1]);

    ObjectDescriptor[] objects = new ObjectDescriptor[numObjects];
    for (int i = 0; i < numObjects; i++) {
      String[] objS = readLineSkipComments(reader).split("=");
      String[] objArgS = objS[1].split(";");
      String[] objIdS = objArgS[0].split("v");
      objects[i] = new ObjectDescriptor(
        factory.buildIdFromToString(objIdS[0]),
        Long.parseLong(objIdS[1]),
        Long.parseLong(objArgS[1]),
        Long.parseLong(objArgS[2]),
        Integer.parseInt(objArgS[3])
        );
    }

    String[] pointerNumS = readLineSkipComments(reader).split("=");
    if (!pointerNumS[0].equals("pointers")) {
      throw new AggregationException("Cannot find number of pointers: " + pointerNumS[0]);
    }
    int numPointers = Integer.parseInt(pointerNumS[1]);

    Id[] pointers = new Id[numPointers];
    for (int i = 0; i < numPointers; i++) {
      String[] ptrS = readLineSkipComments(reader).split("=");
      pointers[i] = factory.buildIdFromToString(ptrS[1]);
    }

    return new AggregateDescriptor(aggrKey, expires, objects, pointers);
  }

  /**
   * DESCRIBE THE METHOD
   *
   * @return DESCRIBE THE RETURN VALUE
   */
  public boolean readFromDisk() {
    rootKey = null;
    aggregateList.clear();

    String fileName;
    if ((new File(configFileName)).exists()) {
      fileName = configFileName;
    } else if ((new File(configFileName + ".new")).exists()) {
      fileName = configFileName + ".new";
    } else {
      return false;
    }

    BufferedReader configFile = null;
    boolean readSuccessful = false;

    try {
      configFile = new BufferedReader(new FileReader(fileName));

      String[] root = readLineSkipComments(configFile).split("=");
      if (root[0].equals("nextid")) {
        nextSerial = Long.parseLong(root[1]);
        root = readLineSkipComments(configFile).split("=");
      }
      if (!root[0].equals("root")) {
        throw new AggregationException("Cannot read root key: " + root[0]);
      }
      rootKey = factory.buildIdFromToString(root[1]);

      while (true) {
        String aggrKeyLine = readLineSkipComments(configFile);
        if (aggrKeyLine == null) {
          readSuccessful = true;
          break;
        }

        String[] aggrKeyS = aggrKeyLine.split("\\[|\\]");
        Id aggrKey = factory.buildIdFromToString(aggrKeyS[1]);
        AggregateDescriptor adc = readAggregate(configFile, aggrKey);

        addAggregateDescriptor(adc, false);
      }
    } catch (Exception e) {
      if (logger.level <= Logger.WARNING) {
        logger.logException(
          "Cannot read configuration file: " + configFileName + " (e=" + e + ")", e);
      }
    }

    if (configFile != null) {
      try {
        configFile.close();
      } catch (Exception e) {
      }
    }

    if (!readSuccessful) {
      rootKey = null;
      aggregateList.clear();
    }

    return readSuccessful;
  }

  /**
   * DESCRIBE THE METHOD
   *
   * @param stream DESCRIBE THE PARAMETER
   * @param adc DESCRIBE THE PARAMETER
   */
  private void writeAggregate(PrintStream stream, AggregateDescriptor adc) {
    stream.println("expires=" + adc.currentLifetime);
    stream.println("objects=" + adc.objects.length);
    for (int i = 0; i < adc.objects.length; i++) {
      stream.println("obj" + i + "=" +
        adc.objects[i].key.toStringFull() + "v" +
        adc.objects[i].version + ";" +
        adc.objects[i].currentLifetime + ";" +
        adc.objects[i].refreshedLifetime + ";" +
        adc.objects[i].size
        );
    }
    stream.println("pointers=" + adc.pointers.length);
    for (int i = 0; i < adc.pointers.length; i++) {
      stream.println("ptr" + i + "=" + adc.pointers[i].toStringFull());
    }
  }

  /**
   * DESCRIBE THE METHOD
   */
  public void writeToDisk() {
    if (rootKey == null) {
      return;
    }

    try {
      PrintStream configFile = new PrintStream(new FileOutputStream(configFileName + ".new"));
      Enumeration enumeration = aggregateList.elements();

      resetMarkers();
      configFile.println("# Aggregate list at " + label + " (" + (new Date()) + ")");
      configFile.println();
      configFile.println("nextid=" + nextSerial);
      configFile.println("root=" + rootKey.toStringFull());
      configFile.println();

      while (enumeration.hasMoreElements()) {
        AggregateDescriptor aggr = (AggregateDescriptor) enumeration.nextElement();
        if (!aggr.marker) {
          configFile.println("[" + aggr.key.toStringFull() + "]");
          writeAggregate(configFile, aggr);
          configFile.println("");

          aggr.marker = true;
        }
      }

      configFile.close();
      (new File(configFileName)).delete();
      (new File(configFileName + ".new")).renameTo(new File(configFileName));
    } catch (IOException ioe) {
      if (logger.level <= Logger.WARNING) {
        logger.logException(
          "AggregationImpl cannot write to its aggregate list: " + configFileName + " (" + ioe + ")", ioe);
      }
    }
  }
}