bundledbpersistencemanager.java

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

     * Returns the miminum blob size in bytes.
     * @return the miminum blob size in bytes.
     */
    public String getMinBlobSize() {
        return String.valueOf(minBlobSize);
    }

    /**
     * Sets the minumum blob size. This size defines the threshhold of which
     * size a property is included in the bundle or is stored in the blob store.
     *
     * @param minBlobSize the minimum blobsize in bytes.
     */
    public void setMinBlobSize(String minBlobSize) {
        this.minBlobSize = Integer.decode(minBlobSize).intValue();
    }

    /**
     * Sets the error handling behaviour of this manager. See {@link ErrorHandling}
     * for details about the flags.
     *
     * @param errorHandling the error handling flags
     */
    public void setErrorHandling(String errorHandling) {
        this.errorHandling = new ErrorHandling(errorHandling);
    }

    /**
     * Returns the error handling configuration of this manager
     * @return the error handling configuration of this manager
     */
    public String getErrorHandling() {
        return errorHandling.toString();
    }

    /**
     * Returns <code>true</code> if the blobs are stored in the DB.
     * @return <code>true</code> if the blobs are stored in the DB.
     */
    public boolean useDbBlobStore() {
        return !externalBLOBs;
    }

    /**
     * Returns <code>true</code> if the blobs are stored in the local fs.
     * @return <code>true</code> if the blobs are stored in the local fs.
     */
    public boolean useLocalFsBlobStore() {
        return externalBLOBs;
    }

    /**
     * Checks if the required schema objects exist and creates them if they
     * don't exist yet.
     *
     * @throws SQLException if an SQL error occurs.
     * @throws RepositoryException if an error occurs.
     */
    protected void checkSchema() throws SQLException, RepositoryException {
        if (!checkTablesExist()) {
            // read ddl from resources
            InputStream in = BundleDbPersistenceManager.class.getResourceAsStream(schema + ".ddl");
            if (in == null) {
                String msg = "Configuration error: unknown schema '" + schema + "'";
                log.debug(msg);
                throw new RepositoryException(msg);
            }
            BufferedReader reader = new BufferedReader(new InputStreamReader(in));
            Statement stmt = con.createStatement();
            try {
                String sql = reader.readLine();
                while (sql != null) {
                    // replace prefix variable
                    sql = Text.replace(sql, SCHEMA_OBJECT_PREFIX_VARIABLE, schemaObjectPrefix).trim();
                    if (sql.length() > 0 && (sql.indexOf("BINVAL") < 0 || useDbBlobStore())) {
                        // only create blob related tables of db blob store configured
                        // execute sql stmt
                        stmt.executeUpdate(sql);
                    }
                    // read next sql stmt
                    sql = reader.readLine();
                }
            } catch (IOException e) {
                String msg = "Configuration error: unable to read schema '" + schema + "': " + e;
                log.debug(msg);
                throw new RepositoryException(msg, e);
            } finally {
                try {
                    in.close();
                } catch (IOException e) {
                    // ignore
                }
                stmt.close();
            }
        }
    }

    /**
     * Checks if the database table exist.
     *
     * @return <code>true</code> if the tables exist;
     *         <code>false</code> otherwise.
     *
     * @throws SQLException if an SQL erro occurs.
     */
    protected boolean checkTablesExist() throws SQLException {
        DatabaseMetaData metaData = con.getMetaData();
        String tableName = schemaObjectPrefix + "BUNDLE";
        if (metaData.storesLowerCaseIdentifiers()) {
            tableName = tableName.toLowerCase();
        } else if (metaData.storesUpperCaseIdentifiers()) {
            tableName = tableName.toUpperCase();
        }
        String userName = checkTablesWithUser() ? metaData.getUserName() : null;
        ResultSet rs = metaData.getTables(null, userName, tableName, null);
        try {
            return rs.next();
        } finally {
            rs.close();
        }
    }

    /**
     * Indicates if the username should be included when retrieving the tables
     * during {@link #checkTablesExist()}.
     * <p/>
     * Please note that this currently only needs to be changed for oracle based
     * persistence managers.
     *
     * @return <code>false</code>
     */
    protected boolean checkTablesWithUser() {
        return false;
    }

    /**
     * {@inheritDoc}
     *
     * Basically wrapps a JDBC transaction around super.store().
     */
    public synchronized void store(ChangeLog changeLog)
            throws ItemStateException {

        try {
            con.setAutoCommit(false);
            super.store(changeLog);
        } catch (SQLException e) {
            String msg = "setting autocommit failed.";
            log.error(msg, e);
            throw new ItemStateException(msg, e);
        } catch (ItemStateException e) {
            // storing the changes failed, rollback changes
            try {
                con.rollback();
            } catch (SQLException e1) {
                String msg = "rollback of change log failed";
                log.error(msg, e1);
            }
            // re-throw original exception
            throw e;
        }

        // storing the changes succeeded, now commit the changes
        try {
            con.commit();
            con.setAutoCommit(true);
        } catch (SQLException e) {
            String msg = "committing change log failed";
            log.error(msg, e);
            throw new ItemStateException(msg, e);
        }
    }

    /**
     * {@inheritDoc}
     */
    public void init(PMContext context) throws Exception {
        if (initialized) {
            throw new IllegalStateException("already initialized");
        }
        super.init(context);

        this.name = context.getHomeDir().getName();

        // setup jdbc connection
        // Note: Explicit creation of new instance of the driver is required
        // in order to re-register the driver in the DriverManager after a
        // repository shutdown.
        Driver drv = (Driver) Class.forName(driver).newInstance();
        log.info("JDBC driver created: {}", drv);
        con = DriverManager.getConnection(url, user, password);
        con.setAutoCommit(true);

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

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

        // create correct blob store
        blobStore = createBlobStore();

        // prepare statements
        if (getStorageModel() == SM_BINARY_KEYS) {
            bundleInsert = con.prepareStatement("insert into " + schemaObjectPrefix + "BUNDLE (BUNDLE_DATA, NODE_ID) values (?, ?)");
            bundleUpdate = con.prepareStatement("update " + schemaObjectPrefix + "BUNDLE set BUNDLE_DATA = ? where NODE_ID = ?");
            bundleSelect = con.prepareStatement("select BUNDLE_DATA from " + schemaObjectPrefix + "BUNDLE where NODE_ID = ?");
            bundleDelete = con.prepareStatement("delete from " + schemaObjectPrefix + "BUNDLE where NODE_ID = ?");

            nodeReferenceInsert = con.prepareStatement("insert into " + schemaObjectPrefix + "REFS (REFS_DATA, NODE_ID) values (?, ?)");
            nodeReferenceUpdate = con.prepareStatement("update " + schemaObjectPrefix + "REFS set REFS_DATA = ? where NODE_ID = ?");
            nodeReferenceSelect = con.prepareStatement("select REFS_DATA from " + schemaObjectPrefix + "REFS where NODE_ID = ?");
            nodeReferenceDelete = con.prepareStatement("delete from " + schemaObjectPrefix + "REFS where NODE_ID = ?");
        } else {
            bundleInsert = con.prepareStatement("insert into " + schemaObjectPrefix + "BUNDLE (BUNDLE_DATA, NODE_ID_HI, NODE_ID_LO) values (?, ?, ?)");
            bundleUpdate = con.prepareStatement("update " + schemaObjectPrefix + "BUNDLE set BUNDLE_DATA = ? where NODE_ID_HI = ? and NODE_ID_LO = ?");
            bundleSelect = con.prepareStatement("select BUNDLE_DATA from " + schemaObjectPrefix + "BUNDLE where NODE_ID_HI = ? and NODE_ID_LO = ?");
            bundleDelete = con.prepareStatement("delete from " + schemaObjectPrefix + "BUNDLE where NODE_ID_HI = ? and NODE_ID_LO = ?");

            nodeReferenceInsert = con.prepareStatement("insert into " + schemaObjectPrefix + "REFS (REFS_DATA, NODE_ID_HI, NODE_ID_LO) values (?, ?, ?)");
            nodeReferenceUpdate = con.prepareStatement("update " + schemaObjectPrefix + "REFS set REFS_DATA = ? where NODE_ID_HI = ? and NODE_ID_LO = ?");
            nodeReferenceSelect = con.prepareStatement("select REFS_DATA from " + schemaObjectPrefix + "REFS where NODE_ID_HI = ? and NODE_ID_LO = ?");
            nodeReferenceDelete = con.prepareStatement("delete from " + schemaObjectPrefix + "REFS where NODE_ID_HI = ? and NODE_ID_LO = ?");
        }
        // load namespaces
        binding = new BundleBinding(errorHandling, blobStore, getNsIndex(), getNameIndex());
        binding.setMinBlobSize(minBlobSize);

        initialized = true;

        if (consistencyCheck) {
            checkConsistency();
        }
    }

    /**
     * Creates a suitable blobstore
     * @return a blobstore
     * @throws Exception if an unspecified error occurs
     */
    protected CloseableBLOBStore createBlobStore() throws Exception {
        if (useLocalFsBlobStore()) {
            return createLocalFSBlobStore(context);
        } else {
            return createDBBlobStore(context);
        }
    }

    /**
     * Returns the local name index
     * @return the local name index
     * @throws IllegalStateException if an error occurs.
     */
    public StringIndex getNameIndex() {
        try {
            if (nameIndex == null) {
                FileSystemResource res = new FileSystemResource(context.getFileSystem(), RES_NAME_INDEX);
                if (res.exists()) {
                    nameIndex = super.getNameIndex();
                } else {
                    // create db nameindex
                    nameIndex = createDbNameIndex();
                }
            }
            return nameIndex;
        } catch (Exception e) {
            throw new IllegalStateException("Unable to create nsIndex: " + e);
        }
    }

    /**
     * Retruns a new instance of a DbNameIndex.
     * @return a new instance of a DbNameIndex.
     * @throws SQLException if an SQL error occurs.
     */
    protected DbNameIndex createDbNameIndex() throws SQLException {
        return new DbNameIndex(con, schemaObjectPrefix);
    }

    /**
     * returns the storage model
     * @return the storage model
     */
    public int getStorageModel() {
        return SM_BINARY_KEYS;
    }

    /**
     * Creates a blob store that is based on a local fs. This is called by
     * init if {@link #useLocalFsBlobStore()} returns <code>true</code>.
     *
     * @param context the persistence manager context
     * @return a blob store
     * @throws Exception if an error occurs.
     */
    protected CloseableBLOBStore createLocalFSBlobStore(PMContext context)
            throws Exception {
        /**
         * store blob's 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();
        return new FSBlobStore(blobFS);
    }

    /**
     * Creates a blob store that uses the database. This is called by
     * init if {@link #useDbBlobStore()} returns <code>true</code>.
     *
     * @param context the persistence manager context
     *
     * @return a blob store
     * @throws Exception if an error occurs.
     */
    protected CloseableBLOBStore createDBBlobStore(PMContext context)
            throws Exception {
        return new DbBlobStore();
    }

    /**
     * Performs a consistency check.
     */
    private void checkConsistency() {
        int count = 0;
        int total = 0;
        log.info("{}: checking workspace consistency...", name);

        Collection modifications = new ArrayList();
        PreparedStatement stmt = null;
        ResultSet rs = null;
        DataInputStream din = null;
        try {
            if (getStorageModel() == SM_BINARY_KEYS) {
                stmt = con.prepareStatement(
                        "select NODE_ID, BUNDLE_DATA from "
                        + schemaObjectPrefix + "BUNDLE");
            } else {