ntransaction.java

一个类似于openJMS分布在ObjectWeb之下的JMS消息中间件。

/*
 * Copyright (C) 2004 - 2006 ScalAgent Distributed Technologies
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or any later version.
 * 
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 * 
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307
 * USA.
 *
 * Initial developer(s): ScalAgent Distributed Technologies
 * Contributor(s): 
 */
package fr.dyade.aaa.util;

import java.io.*;
import java.util.*;

import org.objectweb.util.monolog.api.BasicLevel;
import org.objectweb.util.monolog.api.Logger;

import fr.dyade.aaa.agent.Debug;

public final class NTransaction implements Transaction, NTransactionMBean {
  // Logging monitor
  private static Logger logmon = null;

  /**
   *  Global in memory log initial capacity, by default 4096.
   *  This value can be adjusted for a particular server by setting
   * <code>NTLogMemoryCapacity</code> specific property.
   * <p>
   *  These property can be fixed either from <code>java</code> launching
   * command, or in <code>a3servers.xml</code> configuration file.
   */
  static int LogMemoryCapacity = 4096;

  /**
   * Returns the initial capacity of global in memory log (by default 4096).
   *
   * @return The initial capacity of global in memory log.
   */
  public int getLogMemoryCapacity() {
    return LogMemoryCapacity;
  }

  /**
   *  Maximum size of memory log, by default 2048Kb.
   *  This value can be adjusted (Kb) for a particular server by setting
   * <code>NTLogMemorySize</code> specific property.
   * <p>
   *  These property can be fixed either from <code>java</code> launching
   * command, or in <code>a3servers.xml</code> configuration file.
   */
  static int MaxLogMemorySize = 2048 * Kb;

  /**
   * Returns the maximum size of memory log in Kb, by default 2048Kb.
   *
   * @return The maximum size of memory log in Kb.
   */
  public int getMaxLogMemorySize() {
    return MaxLogMemorySize/Mb;
  }

  /**
   * Sets the maximum size of memory log in Kb.
   *
   * @param size The maximum size of memory log in Kb.
   */
  public void setMaxLogMemorySize(int size) {
    if (size > 0) MaxLogMemorySize = size *Kb;
  }

  /**
   * Returns the size of memory log in byte.
   *
   * @return The size of memory log in byte.
   */
  public int getLogMemorySize() {
    return logFile.logMemorySize;
  }

  /**
   *  Size of disk log in Mb, by default 16Mb.
   *  This value can be adjusted (Mb) for a particular server by setting
   * <code>NTLogFileSize</code> specific property.
   * <p>
   *  These property can be fixed either from <code>java</code> launching
   * command, or in <code>a3servers.xml</code> configuration file.
   */
  static int LogFileSize = 16 * Mb;

  /**
   * Returns the size of disk log in Mb, by default 16Mb.
   *
   * @return The size of disk log in Mb.
   */
  public int getLogFileSize() {
    return LogFileSize/Mb;
  }

  /**
   * Sets the size of disk log in Mb.
   *
   * @param size The size of disk log in Mb.
   */
  public void setLogFileSize(int size) {
    if (size > 0) LogFileSize = size *Mb;
  }

  /**
   *  Number of pooled operation, by default 1000.
   *  This value can be adjusted for a particular server by setting
   * <code>NTLogThresholdOperation</code> specific property.
   * <p>
   *  These property can be fixed either from <code>java</code> launching
   * command, or in <code>a3servers.xml</code> configuration file.
   */
  static int LogThresholdOperation = 1000;

  /**
   * Returns the pool size for <code>operation</code> objects, by default 1000.
   *
   * @return The pool size for <code>operation</code> objects.
   */
  public int getLogThresholdOperation() {
    return LogThresholdOperation;
  }

  /**
   * Returns the number of commit operation since starting up.
   *
   * @return The number of commit operation.
   */
  public int getCommitCount() {
    return logFile.commitCount;
  }

  /**
   * Returns the number of garbage operation since starting up.
   *
   * @return The number of garbage operation.
   */
  public int getGarbageCount() {
    return logFile.garbageCount;
  }

  long startTime = 0L;

  /**
   * Returns the starting time.
   *
   * @return The starting time.
   */
  public long getStartTime() {
    return startTime;
  }

  /**
   * Returns the cumulated time of garbage operations since starting up.
   *
   * @return The cumulated time of garbage operations since starting up.
   */
  public long getGarbageTime() {
    return logFile.garbageTime;
  }

  /**
   * Returns the ratio of garbage operations since starting up.
   *
   * @return The ratio of garbage operations since starting up.
   */
  public int getGarbageRatio() {
    return (int) ((logFile.garbageTime *100) / (System.currentTimeMillis() - startTime));
  }

  /**
   *  The Repository classname implementation.
   *  This value can be set for a particular server by setting the
   * <code>NTRepositoryImpl</code> specific property. By default its value
   * is "fr.dyade.aaa.util.FileRepository".
   * <p>
   *  These property can be fixed either from <code>java</code> launching
   * command, or in <code>a3servers.xml</code> configuration file.
   */
  String repositoryImpl = "fr.dyade.aaa.util.FileRepository";

  /**
   * Returns the Repository classname implementation.
   *
   * @return The Repository classname implementation.
   */
  public String getRepositoryImpl() {
    return repositoryImpl;
  }

  /**
   * Returns the number of save operation to repository.
   *
   * @return The number of save operation to repository.
   */
  public int getNbSavedObjects() {
    return repository.getNbSavedObjects();
  }

  /**
   * Returns the number of delete operation on repository.
   *
   * @return The number of delete operation on repository.
   */
  public int getNbDeletedObjects() {
    return repository.getNbDeletedObjects();
  }

  /**
   * Returns the number of useless delete operation on repository.
   *
   * @return The number of useless delete operation on repository.
   */
  public int getNbBadDeletedObjects() {
    return repository.getNbBadDeletedObjects();
  }

  /**
   * Returns the number of load operation from repository.
   *
   * @return The number of load operation from repository.
   */
  public int getNbLoadedObjects() {
    return repository.getNbLoadedObjects();
  }

  /** Log context associated with each Thread using NTransaction. */
  private class Context {
    Hashtable log = null;
    ByteArrayOutputStream bos = null;
    ObjectOutputStream oos = null;

    Context() {
      log = new Hashtable(15);
      bos = new ByteArrayOutputStream(256);
    }
  }

  File dir = null;

  LogFile logFile = null;

  Repository repository = null;

  /**
   *  ThreadLocal variable used to get the log to associate state with each
   * thread. The log contains all operations do by the current thread since
   * the last <code>commit</code>. On commit, its content is added to current
   * log (memory + disk), then it is freed.
   */
  private ThreadLocal perThreadContext = null;

  static final boolean debug = false;

  public NTransaction() {}

  /**
   * Tests if the Transaction component is persistent.
   *
   * @return true.
   */
  public boolean isPersistent() {
    return true;
  }

  public final void init(String path) throws IOException {
    phase = INIT;

    logmon = Debug.getLogger(Debug.A3Debug + ".Transaction");
    if (logmon.isLoggable(BasicLevel.INFO))
      logmon.log(BasicLevel.INFO, "NTransaction, init()");

    LogMemoryCapacity = Integer.getInteger("NTLogMemoryCapacity",
                                           LogMemoryCapacity).intValue();
    LogFileSize = Integer.getInteger("NTLogFileSize",
                                     LogFileSize /Mb).intValue() *Mb;
    MaxLogMemorySize = Integer.getInteger("NTLogMemorySize",
                                          MaxLogMemorySize /Kb).intValue() *Kb;

    dir = new File(path);
    if (!dir.exists()) dir.mkdir();
    if (!dir.isDirectory())
      throw new FileNotFoundException(path + " is not a directory.");

    // Saves the transaction classname in order to prevent use of a
    // different one after restart (see AgentServer.init).
    DataOutputStream ldos = null;
    try {
      File tfc = new File(dir, "TFC");
      if (! tfc.exists()) {
        ldos = new DataOutputStream(new FileOutputStream(tfc));
        ldos.writeUTF(getClass().getName());
        ldos.flush();
      }
    } finally {
      if (ldos != null) ldos.close();
    }

    try {
      repositoryImpl = System.getProperty("NTRepositoryImpl", repositoryImpl);
      repository = (Repository) Class.forName(repositoryImpl).newInstance();
      repository.init(dir);
    } catch (ClassNotFoundException exc) {
      logmon.log(BasicLevel.FATAL,
                 "NTransaction, cannot initializes the repository ", exc);
      throw new IOException(exc.getMessage());
    } catch (InstantiationException exc) {
      logmon.log(BasicLevel.FATAL,
                 "NTransaction, cannot initializes the repository ", exc);
      throw new IOException(exc.getMessage());
    } catch (IllegalAccessException exc) {
      logmon.log(BasicLevel.FATAL,
                 "NTransaction, cannot initializes the repository ", exc);
      throw new IOException(exc.getMessage());
    }
    logFile = new LogFile(dir, repository);

    perThreadContext = new ThreadLocal() {
        protected synchronized Object initialValue() {
          return new Context();
        }
      };

    startTime = System.currentTimeMillis();

    if (logmon.isLoggable(BasicLevel.INFO))
      logmon.log(BasicLevel.INFO, "NTransaction, initialized " + startTime);

    /* The Transaction subsystem is ready */
    setPhase(FREE);
  }

  public final File getDir() {
    return dir;
  }

  /**
   * Returns the path of persistence directory.
   *
   * @return The path of persistence directory.
   */
  public String getPersistenceDir() {
    return dir.getPath();
  }

  // State of the transaction monitor.
  private int phase = INIT;
  String phaseInfo = PhaseInfo[phase];

  /**
   *
   */
  public int getPhase() {
    return phase;
  }

  public String getPhaseInfo() {
    return phaseInfo;
  }

  private final void setPhase(int newPhase) {
    phase = newPhase;
    phaseInfo = PhaseInfo[phase];
  }

  public final synchronized void begin() throws IOException {
    while (phase != FREE) {
      try {
	wait();
      } catch (InterruptedException exc) {
      }
    }
    // Change the transaction state.
    setPhase(RUN);
  }

  /**
   *  Returns an array of strings naming the persistent objects denoted by
   * a name that satisfy the specified prefix. Each string is an object name.
   * 
   * @param prefix	the prefix
   * @return		An array of strings naming the persistent objects
   *		 denoted by a name that satisfy the specified prefix. The
   *		 array will be empty if no names match.
   */
  public synchronized String[] getList(String prefix) {
    String[] list1 = null;
    try {
      list1 = repository.list(prefix);
    } catch (IOException exc) {
      // AF: TODO
    }
    if (list1 == null) list1 = new String[0];
    Object[] list2 = logFile.log.keySet().toArray();
    int nb = list1.length;
    for (int i=0; i<list2.length; i++) {
      if ((list2[i] instanceof String) &&
          (((String) list2[i]).startsWith(prefix))) {
        int j=0;
        for (; j<list1.length; j++) {
          if (list2[i].equals(list1[j])) break;
        }
        if (j<list1.length) {
          // The file is already in the directory list, it must be count
          // at most once.
          if (((Operation) logFile.log.get(list2[i])).type == Operation.DELETE) {
            // The file is deleted in transaction log.
            list1[j] = null;
            nb -= 1;
          }
          list2[i] = null;
        } else if (((Operation) logFile.log.get(list2[i])).type == Operation.SAVE) {
          // The file is added in transaction log
          nb += 1;
        } else {
          list2[i] = null;
        }
      } else {
        list2[i] = null;
      }
    }
    String[] list = new String[nb];
    for (int i=list1.length-1; i>=0; i--) {
      if (list1[i] != null) list[--nb] = list1[i];
    }
    for (int i=list2.length-1; i>=0; i--) {
      if (list2[i] != null) list[--nb] = (String) list2[i];
    }
        
    return list;
  }

  public final void save(Serializable obj, String name) throws IOException {
    save(obj, null, name);
  }

  static private final byte[] OOS_STREAM_HEADER = {
    (byte)((ObjectStreamConstants.STREAM_MAGIC >>> 8) & 0xFF),
    (byte)((ObjectStreamConstants.STREAM_MAGIC >>> 0) & 0xFF),
    (byte)((ObjectStreamConstants.STREAM_VERSION >>> 8) & 0xFF),
    (byte)((ObjectStreamConstants.STREAM_VERSION >>> 0) & 0xFF)
  };

  public final void save(Serializable obj,
                         String dirName, String name) throws IOException {
    if (logmon.isLoggable(BasicLevel.DEBUG))
      logmon.log(BasicLevel.DEBUG, "NTransaction, save(" + dirName + ", " + name + ")");

    Context ctx = (Context) perThreadContext.get();
    if (ctx.oos == null) {
      ctx.bos.reset();
      ctx.oos = new ObjectOutputStream(ctx.bos);
    } else {
      ctx.oos.reset();