dbtransaction.java

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

  }

  private final void saveInLog(byte[] buf,
                               String name,
                               Hashtable log,
                               boolean copy) throws IOException {
    DBOperation op = DBOperation.alloc(DBOperation.SAVE, name, buf);
    DBOperation old = (DBOperation) log.put(name, op);
    if (copy) {
      if ((old != null) &&
          (old.type == DBOperation.SAVE) &&
          (old.value.length == buf.length)) {
        // reuse old buffer
        op.value = old.value;
      } else {
        // alloc a new one
        op.value = new byte[buf.length];
      }
      System.arraycopy(buf, 0, op.value, 0, buf.length);
    }
    if (old != null) old.free();
  }

  public Object load(String dirName, String name) throws IOException, ClassNotFoundException {
    return load(fname(dirName, name));
  }

  public Object load(String name) throws IOException, ClassNotFoundException {
    byte[] buf = loadByteArray(name);
    if (buf != null) {
      ByteArrayInputStream bis = new ByteArrayInputStream(buf);
      ObjectInputStream ois = new ObjectInputStream(bis);	  
      return ois.readObject();
    }
    return null;
  }

  public byte[] loadByteArray(String dirName, String name) throws IOException {
    return loadByteArray(fname(dirName, name));
  }

  public synchronized byte[] loadByteArray(String name) throws IOException {
    if (logmon.isLoggable(BasicLevel.DEBUG))
      logmon.log(BasicLevel.DEBUG, "DBTransaction, loadByteArray(" + name + ")");

    // Searchs in the log a new value for the object.
    Hashtable log = ((Context) perThreadContext.get()).log;
    DBOperation op = (DBOperation) log.get(name);
    if (op != null) {
      if (op.type == DBOperation.SAVE) {
	return op.value;
      } else if (op.type == DBOperation.DELETE) {
	// The object was deleted.
	return null;
      }
    }

    try {
      // Creating a statement lets us issue commands against the connection.
      Statement s = conn.createStatement();
      //
      ResultSet rs = s.executeQuery("SELECT content FROM JoramDB WHERE name='" + name + "'");

      if (!rs.next()) return null;

       byte[] content = rs.getBytes(1);

       rs.close();
       s.close();

       return content;
    } catch (SQLException sqle) {
      throw new IOException(sqle.getMessage());
    }
  }

  public void delete(String dirName, String name) {
    delete(fname(dirName, name));
  }

  public void delete(String name) {
    if (logmon.isLoggable(BasicLevel.DEBUG))
      logmon.log(BasicLevel.DEBUG,
                 "DBTransaction, delete(" + name + ")");

    Hashtable log = ((Context) perThreadContext.get()).log;
    DBOperation op = DBOperation.alloc(DBOperation.DELETE, name);
    op = (DBOperation) log.put(name, op);
    if (op != null) op.free();
  }

  public final synchronized void commit() throws IOException {
    if (phase != RUN)
      throw new IllegalStateException("Can not commit.");

    if (logmon.isLoggable(BasicLevel.DEBUG))
      logmon.log(BasicLevel.DEBUG, "DBTransaction, commit");
    
    Hashtable log = ((Context) perThreadContext.get()).log;
    if (! log.isEmpty()) {
      DBOperation op = null;
      for (Enumeration e = log.elements(); e.hasMoreElements(); ) {
        op = (DBOperation) e.nextElement();
        if (op.type == DBOperation.SAVE) {
          if (logmon.isLoggable(BasicLevel.DEBUG))
            logmon.log(BasicLevel.DEBUG,
                       "DBTransaction, commit.save (" + op.name + ')');

          try {
            insertStmt.setString(1, op.name);
            insertStmt.setBytes(2, op.value);
            insertStmt.executeUpdate();
          } catch (SQLException sqle1) {
            try {
              updateStmt.setBytes(1, op.value);
              updateStmt.setString(2, op.name);
              updateStmt.executeUpdate();
            } catch (SQLException sqle) {
              throw new IOException(sqle.getMessage());
            }
          }
        } else if (op.type == DBOperation.DELETE) {
          if (logmon.isLoggable(BasicLevel.DEBUG))
            logmon.log(BasicLevel.DEBUG,
                       "DBTransaction, commit.delete (" + op.name + ')');

          try {
            deleteStmt.setString(1, op.name);
            deleteStmt.executeUpdate();
          } catch (SQLException sqle) {
            throw new IOException(sqle.getMessage());
          }
        }
        op.free();
      }
      log.clear();

      try {
        conn.commit();
      } catch (SQLException sqle) {
        throw new IOException(sqle.getMessage());
      }
    }

    if (logmon.isLoggable(BasicLevel.DEBUG))
      logmon.log(BasicLevel.DEBUG, "DBTransaction, committed");

    setPhase(COMMIT);
  }

  public final synchronized void rollback() throws IOException {
    if (phase != RUN)
      throw new IllegalStateException("Can not rollback.");

    if (logmon.isLoggable(BasicLevel.DEBUG))
      logmon.log(BasicLevel.DEBUG, "DBTransaction, rollback");

    setPhase(ROLLBACK);
    ((Context) perThreadContext.get()).log.clear();
  }

  public final synchronized void release() throws IOException {
    if ((phase != RUN) && (phase != COMMIT) && (phase != ROLLBACK))
      throw new IllegalStateException("Can not release transaction.");

    // Change the transaction state.
    setPhase(FREE);
    // wake-up an eventually user's thread in begin
    notify();
  }

  /**
   * Stops the transaction module.
   * It waits all transactions termination, then the module is kept
   * in a FREE 'ready to use' state.
   */
  public synchronized void stop() {
    if (logmon.isLoggable(BasicLevel.INFO))
      logmon.log(BasicLevel.INFO, "DBTransaction, stops");

    while (phase != FREE) {
      // Wait for the transaction subsystem to be free
      try {
        wait();
      } catch (InterruptedException exc) {
      }
    }
    setPhase(FINALIZE);

//     server.shutdown();
    try {
//       org.hsqldb.DatabaseManager.closeDatabases(0);
      // Creating a statement lets us issue commands against the connection.
      Statement s = conn.createStatement();
      // .
//       s.execute("CHECKPOINT DEFRAG");
      conn.commit();
    logmon.log(BasicLevel.INFO, "DBTransaction, TBR stop#3");
      s.executeUpdate("SHUTDOWN COMPACT");
    logmon.log(BasicLevel.INFO, "DBTransaction, TBR stop#4");
      s.close();
    logmon.log(BasicLevel.INFO, "DBTransaction, TBR stop#5");
    } catch (SQLException sqle) {
// AF: TODO
//       throw new IOException(sqle.getMessage());
      logmon.log(BasicLevel.ERROR, "DBTransaction, stop#6", sqle);
    } catch (Throwable t) {
      logmon.log(BasicLevel.ERROR, "DBTransaction, stop#7", t);
    } finally {
      logmon.log(BasicLevel.INFO, "DBTransaction, stop#8");
    }
    logmon.log(BasicLevel.INFO, "DBTransaction, TBR stop#9");
    setPhase(FREE);

    if (logmon.isLoggable(BasicLevel.INFO)) {
      logmon.log(BasicLevel.INFO, "NTransaction, stopped: ");
    }
  }

  /**
   * Close the transaction module.
   * It waits all transactions termination, the module will be initialized
   * anew before reusing it.
   */
  public synchronized void close() {
    if (logmon.isLoggable(BasicLevel.INFO))
      logmon.log(BasicLevel.INFO, "DBTransaction, close");

    if (phase == INIT) return;

    while (phase != FREE) {
      // Wait for the transaction subsystem to be free
      try {
        wait();
      } catch (InterruptedException exc) {
      }
    }

    setPhase(FINALIZE);
    try {
      // Creating a statement lets us issue commands against the connection.
      Statement s = conn.createStatement();
      // .
      s.execute("SHUTDOWN COMPACT");
      s.close();
    } catch (SQLException sqle) {
// AF: TODO
//       throw new IOException(sqle.getMessage());
      logmon.log(BasicLevel.ERROR, "DBTransaction, close", sqle);
    }
    setPhase(INIT);

    if (logmon.isLoggable(BasicLevel.INFO)) {
      logmon.log(BasicLevel.INFO, "DBTransaction, closed: ");
    }
  }
}

final class DBOperation implements Serializable {
  static final int SAVE = 1;
  static final int DELETE = 2;
  static final int COMMIT = 3;
  static final int END = 127;
 
  int type;
  String name;
  byte[] value;

  private DBOperation(int type, String name, byte[] value) {
    this.type = type;
    this.name = name;
    this.value = value;
  }

  /**
   * Returns a string representation for this object.
   *
   * @return	A string representation of this object. 
   */
  public String toString() {
    StringBuffer strbuf = new StringBuffer();

    strbuf.append('(').append(super.toString());
    strbuf.append(",type=").append(type);
    strbuf.append(",name=").append(name);
    strbuf.append(')');
    
    return strbuf.toString();
  }

  private static Pool pool = null;

  static {
    pool = new Pool("DBTransaction$Operation",
                    Integer.getInteger("DBLogThresholdOperation",
                                       DBTransaction.LogThresholdOperation).intValue());
  }

  static DBOperation alloc(int type, String name) {
    return alloc(type, name, null);
  }

  static DBOperation alloc(int type, String name, byte[] value) {
    DBOperation op = null;
    
    try {
      op = (DBOperation) pool.allocElement();
    } catch (Exception exc) {
      return new DBOperation(type, name, value);
    }
    op.type = type;
    op.name = name;
    op.value = value;

    return op;
  }

  void free() {
    /* to let gc do its work */
    name = null;
    value = null;
    pool.freeElement(this);
  }
}