rdbmsadapter.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-2004 (C) Exoffice Technologies Inc. All Rights Reserved.
 *
 * $Id: RDBMSAdapter.java,v 1.54 2004/01/08 05:55:07 tanderson Exp $
 *
 * Date         Author  Changes
 * $Date	    jimm    Created
 */
package org.exolab.jms.persistence;

import java.sql.Connection;
import java.sql.Date;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.Vector;

import EDU.oswego.cs.dl.util.concurrent.FIFOReadWriteLock;
import EDU.oswego.cs.dl.util.concurrent.ReadWriteLock;

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

import org.exolab.core.foundation.HandleIfc;
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.ConfigurationManager;
import org.exolab.jms.config.DatabaseConfiguration;
import org.exolab.jms.config.RdbmsDatabaseConfiguration;
import org.exolab.jms.events.EventHandler;
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.
 *
 * @version     $Revision: 1.54 $ $Date: 2004/01/08 05:55:07 $
 * @author      <a href="mailto:[e-mail]">Jim Mourikis</a>
 */

public class RDBMSAdapter
    extends PersistenceAdapter
    implements EventHandler {

    /**
     * The schema version number. Note this must be incremented whenever
     * The schema changes.
     */
    public static final String SCHEMA_VERSION = "V0.7.6";

    /**
     *  The JDBC ConnectionManager
     */
    private static DBConnectionManager _connectionManager = null;

    /**
     * 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 used to track the incremental garbage collection. It
     * tracks the last blocks examined.
     */
    private long _lastTime = 0;

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

    /**
     * Lock to help prevent deadlocks when administratively removing
     * destinations, while producers and consumers are actively sending
     * and receiving messages. It ensures that when a destination is in the
     * process of being removed, no other changes are occuring on the
     * messages and message_handles tables.
     */
    private ReadWriteLock _destinationLock = new FIFOReadWriteLock();

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

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


    /**
     * Connects to the given db.
     *
     * @throws PersistenceException if a connection cannot be establised to
     * the database
     */
    RDBMSAdapter(String driver, String url, String userName, String password)
        throws PersistenceException {

        DatabaseConfiguration dbConfig = 
            ConfigurationManager.getConfig().getDatabaseConfiguration();
        RdbmsDatabaseConfiguration config = 
            dbConfig.getRdbmsDatabaseConfiguration();

        // create the connection manager, and configure it
        _connectionManager = getConnectionManager(config.getClazz());
        _connectionManager.setUser(userName);
        _connectionManager.setPassword(password);
        _connectionManager.setDriver(driver);
        _connectionManager.setURL(url);
        _connectionManager.setMaxActive(config.getMaxActive());
        _connectionManager.setMaxIdle(config.getMaxIdle());
        _connectionManager.setMinIdleTime(config.getMinIdleTime());
        _connectionManager.setEvictionInterval(config.getEvictionInterval());
        _connectionManager.setTestQuery(config.getTestQuery());
        _connectionManager.setTestBeforeUse(config.getTestBeforeUse());

        // initialisze the connection manager
        _connectionManager.init();

        Connection connection = null;
        try {
            // initialize the various caches and helper classes used to
            // execute the various SQL.
            connection = getConnection();

            String version = getSchemaVersion(connection);
            if (version == null) {
                initSchemaVersion(connection);
            } else if (!version.equals(SCHEMA_VERSION)) {
                throw new PersistenceException(
                    "Schema needs to be converted from version=" + version
                    + " to version=" + SCHEMA_VERSION
                    + "\nBack up your database, and run 'dbtool -migrate'"
                    + "to convert the schema");
            }

            SeedGenerator.initialise();
            Destinations.initialise(connection);
            Consumers.initialise(connection);
            Messages.initialise();
            MessageHandles.initialise();
            connection.commit();
            Users.initialise(connection);
        } catch (PersistenceException exception) {
            SQLHelper.rollback(connection);
            throw exception;
        } catch (Exception exception) {
            throw new PersistenceException(
                "Failed to initialise database adapter", exception);
        } finally {
            SQLHelper.close(connection);
            
        }

        // check whether we should initiate automatic garbage collection
        if (dbConfig.hasGarbageCollectionInterval()) {
            _gcInterval = dbConfig.getGarbageCollectionInterval() * 1000;
            registerEvent();
        }

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

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

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

        if (Destinations.instance() != null) {
            Destinations.instance().close();
        }

        if (Consumers.instance() != null) {
            Consumers.instance().close();
        }

        if (Messages.instance() != null) {
            Messages.instance().close();
        }

        if (MessageHandles.instance() != null) {
            MessageHandles.instance().close();
        }

        if (Users.instance() != null) {
            Users.instance().close();
        }
    }

    // implementation of PersistenceAdapter.getLastId
    public long getLastId(Connection connection)
        throws PersistenceException {

        long lastId = -1;
        boolean successful = false;
        PreparedStatement query = null;
        ResultSet result = null;
        PreparedStatement insert = null;
        try {
            query = connection.prepareStatement(
                "select * from message_id where id = 1");
            result = query.executeQuery();

            if (result.next()) {
                lastId = result.getInt("maxId");
            } else {
                // first entry create.
                insert = connection.prepareStatement(
                    "insert into message_id values (?,?)");
                insert.setInt(1, 1);
                insert.setLong(2, 0);
                insert.executeUpdate();
                lastId = 0;
            }
        } catch (Exception exception) {
            throw new PersistenceException("Failed to get last message id",
                exception);
        } finally {
            SQLHelper.close(result);
            SQLHelper.close(insert);
            SQLHelper.close(query);        	
        }

        return lastId;
    }

    // implementation of PersistenceAdapter.updateIds
    public void updateIds(Connection connection, long id)
        throws PersistenceException {
		PreparedStatement insert = null;
        try {
            insert = connection.prepareStatement(
                "update message_id set maxId = ? where id = 1");

            insert.setLong(1, id);
            insert.executeUpdate();
        } catch (Exception exception) {
            throw new PersistenceException("Failed to update message id",
                                           exception);
        } finally {
            SQLHelper.close(insert);
        }
    }

    // implementation of PersistenceMessage.addMessage
    public void addMessage(Connection connection, MessageImpl message)
        throws PersistenceException {

        long start = 0;

        if (_log.isDebugEnabled()) {
            start = System.currentTimeMillis();
        }

        try {