rdbmsadapter.java

来自「OpenJMS是一个开源的Java Message Service API 1.」· Java 代码 · 共 959 行 · 第 1/3 页

/**
 * 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-2005 (C) Exoffice Technologies Inc. All Rights Reserved.
 *
 * $Id: RDBMSAdapter.java,v 1.6 2005/08/31 05:45:50 tanderson Exp $
 */
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.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.DatabaseConfiguration;
import org.exolab.jms.config.RdbmsDatabaseConfiguration;
import org.exolab.jms.events.BasicEventManager;
import org.exolab.jms.events.EventHandler;
import org.exolab.jms.message.MessageImpl;
import org.exolab.jms.messagemgr.MessageHandle;


/**
 * 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.
 *
 * @author <a href="mailto:[e-mail]">Jim Mourikis</a>
 * @version $Revision: 1.6 $ $Date: 2005/08/31 05:45:50 $
 */

public class RDBMSAdapter
    extends PersistenceAdapter
    implements EventHandler {

    /**
     * The seed generator.
     */
    private final SeedGenerator _seeds;

    /**
     * The destination manager.
     */
    private final Destinations _destinations;

    /**
     * The consumer manager.
     */
    private final Consumers _consumers;

    /**
     * The message manager.
     */
    private final Messages _messages;

    /**
     * The message handles manager.
     */
    private final MessageHandles _handles;

    /**
     * The user manager.
     */
    private final Users _users;

    /**
     * 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 DBConnectionManager _connectionManager = null;

    /**
     * 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();

    /**
     * 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
     */
    public RDBMSAdapter(DatabaseConfiguration dbConfig, String driver, String url,
                 String userName, String password)
            throws PersistenceException {

        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");
            }

            _seeds = new SeedGenerator();
            _consumers = new Consumers(_seeds, connection);
            _destinations = new Destinations(_seeds, _consumers, connection);
            _consumers.setDestinations(_destinations);
            _messages = new Messages(_destinations);
            _handles = new MessageHandles(_destinations, _consumers);
            _users = new Users();
            connection.commit();
        } 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.
     */
    public void close() {
        _consumers.close();
        _destinations.close();
    }

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

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

            if (result.next()) {
                lastId = result.getInt(1);
            } 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 {
            _destinationLock.readLock().acquire();
            _messages.add(connection, message);
        } catch (InterruptedException exception) {
            throw new PersistenceException("Failed to acquire lock",
                                           exception);
        } finally {
            _destinationLock.readLock().release();

            if (_log.isDebugEnabled()) {
                _log.debug("addMessage," +
                           (System.currentTimeMillis() - start));
            }
        }
    }

    // implementation of PersistenceMessage.addMessage
    public void updateMessage(Connection connection, MessageImpl message)