objectadapter.java

实现了Jms的服务器源码,支持多种适配器,DB,FTP,支持多种数据库

/**
 * Redistribution and use of this software and associated documentation
 * ("Software"), with or without modification, are permitted provided
 * that the following conditions are met:
 *
 * 1. Redistributions of source code must retain copyright
 *    statements and notices.  Redistributions must also contain a
 *    copy of this document.
 *
 * 2. Redistributions in binary form must reproduce the
 *    above copyright notice, this list of conditions and the
 *    following disclaimer in the documentation and/or other
 *    materials provided with the distribution.
 *
 * 3. The name "Exolab" must not be used to endorse or promote
 *    products derived from this Software without prior written
 *    permission of Exoffice Technologies.  For written permission,
 *    please contact [e-mail].
 *
 * 4. Products derived from this Software may not be called "Exolab"
 *    nor may "Exolab" appear in their names without prior written
 *    permission of Exoffice Technologies. Exolab is a registered
 *    trademark of Exoffice Technologies.
 *
 * 5. Due credit should be given to the Exolab Project
 *    (http://www.exolab.org/).
 *
 * THIS SOFTWARE IS PROVIDED BY EXOFFICE TECHNOLOGIES AND CONTRIBUTORS
 * ``AS IS'' AND ANY EXPRESSED OR IMPLIED WARRANTIES, INCLUDING, BUT
 * NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND
 * FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.  IN NO EVENT SHALL
 * EXOFFICE TECHNOLOGIES OR ITS CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
 * INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
 * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
 * SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
 * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
 * STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED
 * OF THE POSSIBILITY OF SUCH DAMAGE.
 *
 * Copyright 2000-2003 (C) Exoffice Technologies Inc. All Rights Reserved.
 *
 * $Id: ObjectAdapter.java,v 1.30 2003/08/17 01:32:25 tanderson Exp $
 *
 * Date         Author  Changes
 * $Date	    jimm    Created
 */


package org.exolab.jms.persistence;

import java.io.Externalizable;
import java.io.IOException;
import java.io.ObjectInput;
import java.io.ObjectOutput;
import java.sql.Connection;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.Vector;

import javax.jms.JMSException;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import org.exolab.core.database.recman.PMDHandle;
import org.exolab.core.database.recman.PMDHashMap;
import org.exolab.core.database.recman.PMDSessionManager;
import org.exolab.core.database.recman.PMDVector;
import org.exolab.core.database.recman.PageManagedDatabase;
import org.exolab.core.foundation.DatabaseIOException;
import org.exolab.core.foundation.FailedToAcquireLockException;
import org.exolab.core.foundation.FailedToCreateCollectionException;
import org.exolab.core.foundation.FailedToCreateDatabaseException;
import org.exolab.core.foundation.FailedToCreateLockException;
import org.exolab.core.foundation.FailedToCreateSessionException;
import org.exolab.core.foundation.FailedToDestroySessionException;
import org.exolab.core.foundation.HandleIfc;
import org.exolab.core.foundation.ObjectNameExistsException;
import org.exolab.core.foundation.PersistentObject;
import org.exolab.core.foundation.SessionIfc;
import org.exolab.core.foundation.TransactionException;
import org.exolab.core.foundation.TransactionInProgressException;
import org.exolab.core.foundation.TransactionNotInProgressException;
import org.exolab.jms.authentication.User;
import org.exolab.jms.client.JmsDestination;
import org.exolab.jms.client.JmsQueue;
import org.exolab.jms.client.JmsTopic;
import org.exolab.jms.config.Configuration;
import org.exolab.jms.config.ConfigurationManager;
import org.exolab.jms.config.DatabaseConfiguration;
import org.exolab.jms.events.BasicEventManager;
import org.exolab.jms.events.Event;
import org.exolab.jms.events.EventHandler;
import org.exolab.jms.events.IllegalEventDefinedException;
import org.exolab.jms.message.DestinationImpl;
import org.exolab.jms.message.MessageImpl;
import org.exolab.jms.messagemgr.PersistentMessageHandle;


/**
 * This adapter is a wrapper class around the persistency mechanism.
 * It isolates the client from the working specifics of the database, by
 * providing a simple straight forward interface. Furure changes to
 * the database will only require changes to the adapter.
 *
 * <P>For this release the PageManager is used as the underlying database.
 *
 * @version     $Revision: 1.30 $ $Date: 2003/08/17 01:32:25 $
 * @author      <a href="mailto:[e-mail]">Jim Mourikis</a>
 * @see org.exolab.core.database.recman.PMDSessionManager
 * @see org.exolab.core.database.recman.PMDHashMap
 * @see	org.exolab.core.database.recman.PMDVector
 * @see org.exolab.core.foundation.PersistentObject
 */
public class ObjectAdapter
    extends PersistenceAdapter
    implements EventHandler {

    // the database instance
    private PageManagedDatabase db_ = null;

    // The name of this database.
    private String dbname_ = null;

    // The max time to wait to aquire a lock
    static private int MAX_WAIT_TIME = 5000;

    // The root name for storing messages
    final static private String MESSAGES = "MESSAGES";

    // The root name for storing destinations
    final static private String DESTINATIONS = "DESTINATIONS";

    // All consumers or destinations have a handle
    final static private String HANDLES = "HANDLES";

    // All topics destinations are prefixed with
    final static private String TOPIC = "T_";

    // All queue destinations are prefixed with
    final static private String QUEUE = "Q_";

    //The root name of the max id storage location
    final static private String IDSTORAGE = "IDSTORAGE";

    //The database schema version root
    final static private String VERSIONID = "VERSIONID";

    // The schema version number. Note this must be incremented whenever
    // The schema changes.
    final static private long VERSIONNUM = 7;

    // The minimum persistent consumer list size
    private int minConsumerListSize_ = 10000;

    // The minimum persistent message list size
    private int minMessageListSize_ = 20000;

    /**
     * Reference to a dummy connection object, used to satisfy the fact
     * that the object adapter can look like an Connection object
     */
    private NullConnection _connection = new NullConnection();

    /**
     * This is the interval that the automatic garbage collector will
     * execute, if specified. It is specified in seconds.
     */
    private int _gcInterval = 600;

    /**
     * This is the block size that is used during purging.
     */
    private int _gcBlockSize = 500;

    /**
     * This is the thread priority for the GC Thread
     */
    private int _gcThreadPriority = Thread.NORM_PRIORITY;

    /**
     * This is the event that is fired to initiate garbage collection
     * in the database
     */
    private static final int COLLECT_DATABASE_GARBAGE_EVENT = 1;

    /**
     * The size of the CACHE used by the JDBM
     */
    private static final int MIN_CACHE_SIZE = 512;

    /**
     * The logger
     */
    private static final Log _log = LogFactory.getLog(ObjectAdapter.class);


    /**
     * Constructs a databse with the given name, if one does not exist,
     * otherwise opens the existing database.
     * The database name is checked to see if it already has a ".db" at the
     * end and stripped off if it does. The PageManagedDatabase always
     * prefixes ".db" to any name passed in.
     *
     * @throws PersistenceException for any database error
     *
     */
    public ObjectAdapter(String dbName, int minListSize, int minMsgSize,
                         int minCacheSize)
        throws PersistenceException {

        minConsumerListSize_ = Math.max(minConsumerListSize_, minListSize);
        minMessageListSize_ = Math.max(minMessageListSize_, minMsgSize);

        dbname_ = fixName(dbName);
        try {
            db_ = new PageManagedDatabase(dbname_);

            // set the cache object size
            //db_.setObjectCacheSize(Math.max(MIN_CACHE_SIZE, minCacheSize));

            // initialise the session manager with the created database and
            // then
            // check the version to ensure that it is compatible.
            PMDSessionManager.init(db_);
        } catch (FailedToCreateDatabaseException exception) {
            throw new PersistenceException(
                "Failed to initialise database adapter", exception);
        }

        checkVersion();
        _log.debug("minConsumerListSize = " + minConsumerListSize_);
        _log.debug("minMessageListSize = " + minMessageListSize_);

        // create the roots if they do not already exist
        // jima - we really need to improve this part of the code
        try {
            createIdRoot();
        } catch (Exception exception) {
            throw new PersistenceException(
                "Failed to create database root", exception);
        }

        // check whether we should initiate automatic garbage collection
        DatabaseConfiguration config =
            ConfigurationManager.getConfig().getDatabaseConfiguration();
        if (config.hasGarbageCollectionInterval()) {
            _gcInterval =
                config.getGarbageCollectionInterval() * 1000;
            registerEvent();
        }

        if (config.hasGarbageCollectionBlockSize()) {
            _gcBlockSize =
                config.getGarbageCollectionBlockSize();
        }

        if (config.hasGarbageCollectionThreadPriority()) {
            _gcThreadPriority =
                config.getGarbageCollectionBlockSize();
            if (_gcThreadPriority < Thread.MIN_PRIORITY) {
                _gcThreadPriority = Thread.MIN_PRIORITY;
            } else if (_gcThreadPriority > Thread.MAX_PRIORITY) {
                _gcThreadPriority = Thread.MAX_PRIORITY;
            }
        }
    }

    /**
     * Fix up the database name, by striping off any ".xx" suffixes.
     *
     */
    private String fixName(String dbName) {
        String newDbName = dbName;
        newDbName.trim();

        int i = newDbName.lastIndexOf('.');
        if (i > 0 && i < (newDbName.length() - 1)) {
            newDbName = newDbName.substring(0, i);
        }
        return newDbName;
    }

    /**
     * Close the database if open.
     *
     */
    public void close() {
        if (db_ != null) {
            db_.close();
            dbname_ = null;
            db_ = null;
        }
    }

    /**
     * Check to see if the version is correct. if this is the first time the
     * database has been opened, then created the version root, and itialise
     * the version.
     *
     */
    private void checkVersion() {
        try {
            PMDVector vector;
            SessionIfc session = getSession();

            session.getCurrentTransaction().begin();
            if ((vector = (PMDVector) session.lookup(VERSIONID)) == null) {
                _log.debug("Setting Version Id");
                vector =
                    (PMDVector) session.getCollectionManager().createVector();
                session.createObject(vector);
                session.bind(VERSIONID, vector);
                session.getCurrentTransaction().commit();

                session.getCurrentTransaction().begin();
                vector = (PMDVector) session.lookup(VERSIONID);
                session.acquireLock(vector, 10);
                vector.addElement(new PMDLongInteger(VERSIONNUM));
                session.updateObject(vector);
                session.getCurrentTransaction().commit();
            } else {