databasepersistencemanager.java

来自「jsr170接口的java实现。是个apache的开源项目。」· Java 代码 · 共 1,245 行 · 第 1/3 页

/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.jackrabbit.core.persistence.db;

import org.apache.jackrabbit.core.NodeId;
import org.apache.jackrabbit.core.PropertyId;
import org.apache.jackrabbit.core.fs.FileSystem;
import org.apache.jackrabbit.core.fs.local.LocalFileSystem;
import org.apache.jackrabbit.core.persistence.AbstractPersistenceManager;
import org.apache.jackrabbit.core.persistence.PMContext;
import org.apache.jackrabbit.core.persistence.util.BLOBStore;
import org.apache.jackrabbit.core.persistence.util.FileSystemBLOBStore;
import org.apache.jackrabbit.core.persistence.util.Serializer;
import org.apache.jackrabbit.core.state.ChangeLog;
import org.apache.jackrabbit.core.state.ItemState;
import org.apache.jackrabbit.core.state.ItemStateException;
import org.apache.jackrabbit.core.state.NoSuchItemStateException;
import org.apache.jackrabbit.core.state.NodeReferences;
import org.apache.jackrabbit.core.state.NodeReferencesId;
import org.apache.jackrabbit.core.state.NodeState;
import org.apache.jackrabbit.core.state.PropertyState;
import org.apache.jackrabbit.core.value.BLOBFileValue;
import org.apache.jackrabbit.core.value.InternalValue;
import org.apache.jackrabbit.util.Text;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.jcr.PropertyType;
import javax.jcr.RepositoryException;
import java.io.BufferedReader;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.FilterInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.HashMap;
import java.util.Iterator;

/**
 * Abstract base class for database persistence managers. This class
 * contains common functionality for database persistence manager subclasses
 * that normally differ only in the way the database connection is acquired.
 * Subclasses should override the {@link #getConnection()} method to return
 * the configured database connection.
 * <p>
 * See the {@link SimpleDbPersistenceManager} for a detailed description
 * of the available configuration options and database behaviour.
 */
public abstract class DatabasePersistenceManager extends AbstractPersistenceManager {

    /**
     * Logger instance
     */
    private static Logger log = LoggerFactory.getLogger(DatabasePersistenceManager.class);

    protected static final String SCHEMA_OBJECT_PREFIX_VARIABLE =
            "${schemaObjectPrefix}";

    protected boolean initialized;

    protected String schema;
    protected String schemaObjectPrefix;

    protected boolean externalBLOBs;

    // initial size of buffer used to serialize objects
    protected static final int INITIAL_BUFFER_SIZE = 1024;

    // jdbc connection
    protected Connection con;

    // internal flag governing whether an automatic reconnect should be
    // attempted after a SQLException had been encountered
    protected boolean autoReconnect = true;
    // time to sleep in ms before a reconnect is attempted
    protected static final int SLEEP_BEFORE_RECONNECT = 10000;

    // the map of prepared statements (key: sql stmt, value: prepared stmt)
    private HashMap preparedStatements = new HashMap();

    // SQL statements for NodeState management
    protected String nodeStateInsertSQL;
    protected String nodeStateUpdateSQL;
    protected String nodeStateSelectSQL;
    protected String nodeStateSelectExistSQL;
    protected String nodeStateDeleteSQL;

    // SQL statements for PropertyState management
    protected String propertyStateInsertSQL;
    protected String propertyStateUpdateSQL;
    protected String propertyStateSelectSQL;
    protected String propertyStateSelectExistSQL;
    protected String propertyStateDeleteSQL;

    // SQL statements for NodeReference management
    protected String nodeReferenceInsertSQL;
    protected String nodeReferenceUpdateSQL;
    protected String nodeReferenceSelectSQL;
    protected String nodeReferenceSelectExistSQL;
    protected String nodeReferenceDeleteSQL;

    // SQL statements for BLOB management
    // (if <code>externalBLOBs==false</code>)
    protected String blobInsertSQL;
    protected String blobUpdateSQL;
    protected String blobSelectSQL;
    protected String blobSelectExistSQL;
    protected String blobDeleteSQL;



    /**
     * file system where BLOB data is stored
     * (if <code>externalBLOBs==true</code>)
     */
    protected FileSystem blobFS;
    /**
     * BLOBStore that manages BLOB data in the file system
     * (if <code>externalBLOBs==true</code>)
     */
    protected BLOBStore blobStore;

    /**
     * Creates a new <code>DatabasePersistenceManager</code> instance.
     */
    public DatabasePersistenceManager() {
        schema = "default";
        schemaObjectPrefix = "";
        externalBLOBs = true;
        initialized = false;
    }

    //----------------------------------------------------< setters & getters >

    public String getSchemaObjectPrefix() {
        return schemaObjectPrefix;
    }

    public void setSchemaObjectPrefix(String schemaObjectPrefix) {
        // make sure prefix is all uppercase
        this.schemaObjectPrefix = schemaObjectPrefix.toUpperCase();
    }

    public String getSchema() {
        return schema;
    }

    public void setSchema(String schema) {
        this.schema = schema;
    }

    public boolean isExternalBLOBs() {
        return externalBLOBs;
    }

    public void setExternalBLOBs(boolean externalBLOBs) {
        this.externalBLOBs = externalBLOBs;
    }

    public void setExternalBLOBs(String externalBLOBs) {
        this.externalBLOBs = Boolean.valueOf(externalBLOBs).booleanValue();
    }

    //---------------------------------------------------< PersistenceManager >
    /**
     * {@inheritDoc}
     */
    public void init(PMContext context) throws Exception {
        if (initialized) {
            throw new IllegalStateException("already initialized");
        }

        // setup jdbc connection
        initConnection();

        // make sure schemaObjectPrefix consists of legal name characters only
        prepareSchemaObjectPrefix();

        // check if schema objects exist and create them if necessary
        checkSchema();

        // build sql statements
        buildSQLStatements();

        // prepare statements
        initPreparedStatements();

        if (externalBLOBs) {
            /**
             * store BLOBs in local file system in a sub directory
             * of the workspace home directory
             */
            LocalFileSystem blobFS = new LocalFileSystem();
            blobFS.setRoot(new File(context.getHomeDir(), "blobs"));
            blobFS.init();
            this.blobFS = blobFS;
            blobStore = new FileSystemBLOBStore(blobFS);
        } else {
            /**
             * store BLOBs in db
             */
            blobStore = new DbBLOBStore();
        }

        initialized = true;
    }

    /**
     * {@inheritDoc}
     */
    public synchronized void close() throws Exception {
        if (!initialized) {
            throw new IllegalStateException("not initialized");
        }

        try {
            // close shared prepared statements
            for (Iterator it = preparedStatements.values().iterator(); it.hasNext(); ) {
                closeStatement((PreparedStatement) it.next());
            }
            preparedStatements.clear();

            if (externalBLOBs) {
                // close BLOB file system
                blobFS.close();
                blobFS = null;
            }
            blobStore = null;

            // close jdbc connection
            closeConnection(con);

        } finally {
            initialized = false;
        }
    }

    /**
     * {@inheritDoc}
     */
    public synchronized void store(ChangeLog changeLog)
            throws ItemStateException {
        // temporarily disable automatic reconnect feature
        // since the changes need to be persisted atomically
        autoReconnect = false;
        try {
            ItemStateException ise = null;
            // number of attempts to store the changes
            int trials = 2;
            while (trials > 0) {
                try {
                    super.store(changeLog);
                    break;
                } catch (ItemStateException e) {
                    // catch exception and fall through...
                    ise = e;
                }

                if (ise != null && ise.getCause() instanceof SQLException
                        && --trials > 0) {
                    // a SQLException has been thrown, try to reconnect
                    log.warn("storing changes failed, about to reconnect...", ise.getCause());

                    // try to reconnect
                    if (reestablishConnection()) {
                        // now let's give it another try
                        ise = null;
                        continue;
                    } else {
                        // reconnect failed, proceed with error processing
                        break;
                    }
                }
            }

            if (ise == null) {
                // storing the changes succeeded, now commit the changes
                try {
                    con.commit();
                } catch (SQLException e) {
                    String msg = "committing change log failed";
                    log.error(msg, e);
                    throw new ItemStateException(msg, e);
                }
            } else {
                // storing the changes failed, rollback changes
                try {
                    con.rollback();
                } catch (SQLException e) {
                    String msg = "rollback of change log failed";
                    log.error(msg, e);
                }
                // re-throw original exception
                throw ise;
            }
        } finally {
            // re-enable automatic reconnect feature
            autoReconnect = true;
        }
    }

    /**
     * {@inheritDoc}
     */
    public NodeState load(NodeId id)
            throws NoSuchItemStateException, ItemStateException {
        if (!initialized) {
            throw new IllegalStateException("not initialized");
        }

        synchronized (nodeStateSelectSQL) {
            ResultSet rs = null;
            InputStream in = null;
            try {
                Statement stmt = executeStmt(nodeStateSelectSQL, new Object[]{id.toString()});
                rs = stmt.getResultSet();
                if (!rs.next()) {
                    throw new NoSuchItemStateException(id.toString());
                }

                in = rs.getBinaryStream(1);
                NodeState state = createNew(id);
                Serializer.deserialize(state, in);

                return state;
            } catch (Exception e) {
                if (e instanceof NoSuchItemStateException) {
                    throw (NoSuchItemStateException) e;
                }
                String msg = "failed to read node state: " + id;
                log.error(msg, e);
                throw new ItemStateException(msg, e);
            } finally {
                closeStream(in);
                closeResultSet(rs);
            }
        }
    }

    /**
     * {@inheritDoc}
     */
    public PropertyState load(PropertyId id)
            throws NoSuchItemStateException, ItemStateException {
        if (!initialized) {
            throw new IllegalStateException("not initialized");
        }

        synchronized (propertyStateSelectSQL) {
            ResultSet rs = null;
            InputStream in = null;
            try {
                Statement stmt = executeStmt(propertyStateSelectSQL, new Object[]{id.toString()});
                rs = stmt.getResultSet();
                if (!rs.next()) {
                    throw new NoSuchItemStateException(id.toString());
                }

                in = rs.getBinaryStream(1);
                PropertyState state = createNew(id);
                Serializer.deserialize(state, in, blobStore);

                return state;
            } catch (Exception e) {
                if (e instanceof NoSuchItemStateException) {
                    throw (NoSuchItemStateException) e;
                }
                String msg = "failed to read property state: " + id;
                log.error(msg, e);
                throw new ItemStateException(msg, e);
            } finally {
                closeStream(in);
                closeResultSet(rs);
            }
        }
    }

    /**
     * {@inheritDoc}
     * <p/>
     * This method uses shared <code>PreparedStatement</code>s which must
     * be executed strictly sequentially. Because this method synchronizes on
     * the persistence manager instance there is no need to synchronize on the
     * shared statement. If the method would not be sychronized the shared
     * statements would have to be synchronized.
     */
    public synchronized void store(NodeState state) throws ItemStateException {
        if (!initialized) {
            throw new IllegalStateException("not initialized");
        }

        // check if insert or update
        boolean update = state.getStatus() != ItemState.STATUS_NEW;