databasepersistencemanager.java

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

            PreparedStatement stmt = (PreparedStatement) preparedStatements.get(sql);
            try {
                for (int i = 0; i < params.length; i++) {
                    if (params[i] instanceof SizedInputStream) {
                        SizedInputStream in = (SizedInputStream) params[i];
                        stmt.setBinaryStream(i + 1, in, (int) in.getSize());
                    } else {
                        stmt.setObject(i + 1, params[i]);
                    }
                }
                stmt.execute();
                resetStatement(stmt);
                return stmt;
            } catch (SQLException se) {
                if (--trials == 0) {
                    // no more trials, re-throw
                    throw se;
                }
                log.warn("execute failed, about to reconnect...", se.getMessage());

                // try to reconnect
                if (reestablishConnection()) {
                    // reconnect succeeded; check whether it's possible to
                    // re-execute the prepared stmt with the given parameters
                    for (int i = 0; i < params.length; i++) {
                        if (params[i] instanceof SizedInputStream) {
                            SizedInputStream in = (SizedInputStream) params[i];
                            if (in.isConsumed()) {
                                // we're unable to re-execute the prepared stmt
                                // since an InputStream paramater has already
                                // been 'consumed';
                                // re-throw previous SQLException
                                throw se;
                            }
                        }
                    }

                    // try again to execute the statement
                    continue;
                } else {
                    // reconnect failed, re-throw previous SQLException
                    throw se;
                }
            }
        }
    }

    /**
     * Resets the given <code>PreparedStatement</code> by clearing the parameters
     * and warnings contained.
     * <p/>
     * NOTE: This method MUST be called in a synchronized context as neither
     * this method nor the <code>PreparedStatement</code> instance on which it
     * operates are thread safe.
     *
     * @param stmt The <code>PreparedStatement</code> to reset. If
     *             <code>null</code> this method does nothing.
     */
    protected void resetStatement(PreparedStatement stmt) {
        if (stmt != null) {
            try {
                stmt.clearParameters();
                stmt.clearWarnings();
            } catch (SQLException se) {
                logException("failed resetting PreparedStatement", se);
            }
        }
    }

    protected void closeResultSet(ResultSet rs) {
        if (rs != null) {
            try {
                rs.close();
            } catch (SQLException se) {
                logException("failed closing ResultSet", se);
            }
        }
    }

    protected void closeStream(InputStream in) {
        if (in != null) {
            try {
                in.close();
            } catch (IOException ignore) {
            }
        }
    }

    protected void closeStatement(Statement stmt) {
        if (stmt != null) {
            try {
                stmt.close();
            } catch (SQLException se) {
                logException("failed closing Statement", se);
            }
        }
    }

    protected void logException(String message, SQLException se) {
        if (message != null) {
            log.error(message);
        }
        log.error("    reason: " + se.getMessage());
        log.error("state/code: " + se.getSQLState() + "/" + se.getErrorCode());
        log.debug("      dump:", se);
    }

    /**
     * Makes sure that <code>schemaObjectPrefix</code> does only consist of
     * characters that are allowed in names on the target database. Illegal
     * characters will be escaped as necessary.
     *
     * @throws Exception if an error occurs
     */
    protected void prepareSchemaObjectPrefix() throws Exception {
        DatabaseMetaData metaData = con.getMetaData();
        String legalChars = metaData.getExtraNameCharacters();
        legalChars += "ABCDEFGHIJKLMNOPQRSTUVWXZY0123456789_";

        String prefix = schemaObjectPrefix.toUpperCase();
        StringBuffer escaped = new StringBuffer();
        for (int i = 0; i < prefix.length(); i++) {
            char c = prefix.charAt(i);
            if (legalChars.indexOf(c) == -1) {
                escaped.append("_x");
                String hex = Integer.toHexString(c);
                escaped.append("0000".toCharArray(), 0, 4 - hex.length());
                escaped.append(hex);
                escaped.append("_");
            } else {
                escaped.append(c);
            }
        }
        schemaObjectPrefix = escaped.toString();
    }

    /**
     * Checks if the required schema objects exist and creates them if they
     * don't exist yet.
     *
     * @throws Exception if an error occurs
     */
    protected void checkSchema() throws Exception {
        DatabaseMetaData metaData = con.getMetaData();
        String tableName = schemaObjectPrefix + "NODE";
        if (metaData.storesLowerCaseIdentifiers()) {
            tableName = tableName.toLowerCase();
        } else if (metaData.storesUpperCaseIdentifiers()) {
            tableName = tableName.toUpperCase();
        }

        ResultSet rs = metaData.getTables(null, null, tableName, null);
        boolean schemaExists;
        try {
            schemaExists = rs.next();
        } finally {
            rs.close();
        }

        if (!schemaExists) {
            // read ddl from resources
            InputStream in = getSchemaDDL();
            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) {
                    // Skip comments and empty lines
                    if (!sql.startsWith("#") && sql.length() > 0) {
                        // replace prefix variable
                        sql = Text.replace(sql, SCHEMA_OBJECT_PREFIX_VARIABLE, schemaObjectPrefix);
                        // execute sql stmt
                        stmt.executeUpdate(sql);
                    }
                    // read next sql stmt
                    sql = reader.readLine();
                }
                // commit the changes
                con.commit();
            } finally {
                closeStream(in);
                closeStatement(stmt);
            }
        }
    }

    /**
     * Returns an input stream to the schema DDL resource.
     * @return an input stream to the schema DDL resource.
     */
    protected InputStream getSchemaDDL() {
        // JCR-595: Use the class explicitly instead of using getClass()
        // to avoid problems when subclassed in a different package
        return DatabasePersistenceManager.class.getResourceAsStream(schema + ".ddl");
    }

    /**
     * Builds the SQL statements
     */
    protected void buildSQLStatements() {
        nodeStateInsertSQL = "insert into "
                + schemaObjectPrefix + "NODE (NODE_DATA, NODE_ID) values (?, ?)";

        nodeStateUpdateSQL = "update "
                + schemaObjectPrefix + "NODE set NODE_DATA = ? where NODE_ID = ?";
        nodeStateSelectSQL = "select NODE_DATA from "
                + schemaObjectPrefix + "NODE where NODE_ID = ?";
        nodeStateSelectExistSQL = "select 1 from "
                + schemaObjectPrefix + "NODE where NODE_ID = ?";
        nodeStateDeleteSQL = "delete from "
                + schemaObjectPrefix + "NODE where NODE_ID = ?";

        propertyStateInsertSQL = "insert into "
                + schemaObjectPrefix + "PROP (PROP_DATA, PROP_ID) values (?, ?)";
        propertyStateUpdateSQL = "update "
                + schemaObjectPrefix + "PROP set PROP_DATA = ? where PROP_ID = ?";
        propertyStateSelectSQL = "select PROP_DATA from "
                + schemaObjectPrefix + "PROP where PROP_ID = ?";
        propertyStateSelectExistSQL = "select 1 from "
                + schemaObjectPrefix + "PROP where PROP_ID = ?";
        propertyStateDeleteSQL = "delete from "
                + schemaObjectPrefix + "PROP where PROP_ID = ?";

        nodeReferenceInsertSQL = "insert into "
                + schemaObjectPrefix + "REFS (REFS_DATA, NODE_ID) values (?, ?)";
        nodeReferenceUpdateSQL = "update "
                + schemaObjectPrefix + "REFS set REFS_DATA = ? where NODE_ID = ?";
        nodeReferenceSelectSQL = "select REFS_DATA from "
                + schemaObjectPrefix + "REFS where NODE_ID = ?";
        nodeReferenceSelectExistSQL = "select 1 from "
                + schemaObjectPrefix + "REFS where NODE_ID = ?";
        nodeReferenceDeleteSQL = "delete from "
                + schemaObjectPrefix + "REFS where NODE_ID = ?";

        if (!externalBLOBs) {
            blobInsertSQL = "insert into "
                    + schemaObjectPrefix + "BINVAL (BINVAL_DATA, BINVAL_ID) values (?, ?)";
            blobUpdateSQL = "update "
                    + schemaObjectPrefix + "BINVAL set BINVAL_DATA = ? where BINVAL_ID = ?";
            blobSelectSQL =
                    "select BINVAL_DATA from "
                    + schemaObjectPrefix + "BINVAL where BINVAL_ID = ?";
            blobSelectExistSQL =
                    "select 1 from "
                    + schemaObjectPrefix + "BINVAL where BINVAL_ID = ?";
            blobDeleteSQL = "delete from "
                    + schemaObjectPrefix + "BINVAL where BINVAL_ID = ?";
        }
    }

    /**
     * Initializes the map of prepared statements.
     *
     * @throws SQLException if an error occurs
     */
    protected void initPreparedStatements() throws SQLException {
        preparedStatements.put(
                nodeStateInsertSQL, con.prepareStatement(nodeStateInsertSQL));
        preparedStatements.put(
                nodeStateUpdateSQL, con.prepareStatement(nodeStateUpdateSQL));
        preparedStatements.put(
                nodeStateSelectSQL, con.prepareStatement(nodeStateSelectSQL));
        preparedStatements.put(
                nodeStateSelectExistSQL, con.prepareStatement(nodeStateSelectExistSQL));
        preparedStatements.put(
                nodeStateDeleteSQL, con.prepareStatement(nodeStateDeleteSQL));

        preparedStatements.put(
                propertyStateInsertSQL, con.prepareStatement(propertyStateInsertSQL));
        preparedStatements.put(
                propertyStateUpdateSQL, con.prepareStatement(propertyStateUpdateSQL));
        preparedStatements.put(
                propertyStateSelectSQL, con.prepareStatement(propertyStateSelectSQL));
        preparedStatements.put(
                propertyStateSelectExistSQL, con.prepareStatement(propertyStateSelectExistSQL));
        preparedStatements.put(
                propertyStateDeleteSQL, con.prepareStatement(propertyStateDeleteSQL));

        preparedStatements.put(
                nodeReferenceInsertSQL, con.prepareStatement(nodeReferenceInsertSQL));
        preparedStatements.put(
                nodeReferenceUpdateSQL, con.prepareStatement(nodeReferenceUpdateSQL));
        preparedStatements.put(
                nodeReferenceSelectSQL, con.prepareStatement(nodeReferenceSelectSQL));
        preparedStatements.put(
                nodeReferenceSelectExistSQL, con.prepareStatement(nodeReferenceSelectExistSQL));
        preparedStatements.put(
                nodeReferenceDeleteSQL, con.prepareStatement(nodeReferenceDeleteSQL));

        if (!externalBLOBs) {
            preparedStatements.put(blobInsertSQL, con.prepareStatement(blobInsertSQL));
            preparedStatements.put(blobUpdateSQL, con.prepareStatement(blobUpdateSQL));
            preparedStatements.put(blobSelectSQL, con.prepareStatement(blobSelectSQL));
            preparedStatements.put(blobSelectExistSQL, con.prepareStatement(blobSelectExistSQL));
            preparedStatements.put(blobDeleteSQL, con.prepareStatement(blobDeleteSQL));
        }
    }

    //--------------------------------------------------------< inner classes >

    class SizedInputStream extends FilterInputStream {
        private final long size;
        private boolean consumed = false;

        SizedInputStream(InputStream in, long size) {
            super(in);
            this.size = size;
        }

        long getSize() {
            return size;
        }

        boolean isConsumed() {
            return consumed;
        }

        public int read() throws IOException {
            consumed = true;
            return super.read();
        }

        public long skip(long n) throws IOException {
            consumed = true;
            return super.skip(n);
        }

        public int read(byte b[]) throws IOException {
            consumed = true;
            return super.read(b);
        }

        public int read(byte b[], int off, int len) throws IOException {
            consumed = true;
            return super.read(b, off, len);
        }
    }

    class DbBLOBStore implements BLOBStore {
        /**
         * {@inheritDoc}
         */
        public String createId(PropertyId id, int index) {
            // the blobId is a simple string concatenation of id plus index
            StringBuffer sb = new StringBuffer();
            sb.append(id.toString());
            sb.append('[');
            sb.append(index);
            sb.append(']');
            return sb.toString();
        }

        /**
         * {@inheritDoc}
         */
        public InputStream get(String blobId) throws Exception {
            synchronized (blobSelectSQL) {
                Statement stmt = executeStmt(blobSelectSQL, new Object[]{blobId});
                final ResultSet rs = stmt.getResultSet();
                if (!rs.next()) {
                    closeResultSet(rs);
                    throw new Exception("no such BLOB: " + blobId);
                }
                InputStream in = rs.getBinaryStream(1);
                if (in == null) {
                    // some databases treat zero-length values as NULL;
                    // return empty InputStream in such a case
                    closeResultSet(rs);
                    return new ByteArrayInputStream(new byte[0]);
                }

                /**
                 * return an InputStream wrapper in order to
                 * close the ResultSet when the stream is closed
                 */
                return new FilterInputStream(in) {
                    public void close() throws IOException {
                        in.close();
                        // now it's safe to close ResultSet
                        closeResultSet(rs);
                    }
                };
            }
        }

        /**
         * {@inheritDoc}
         */
        public synchronized void put(String blobId, InputStream in, long size)
                throws Exception {
            Statement stmt = executeStmt(blobSelectExistSQL, new Object[]{blobId});
            ResultSet rs = stmt.getResultSet();
            // a BLOB exists if the result has at least one entry
            boolean exists = rs.next();
            closeResultSet(rs);

            String sql = (exists) ? blobUpdateSQL : blobInsertSQL;
            executeStmt(sql, new Object[]{new SizedInputStream(in, size), blobId});
        }

        /**
         * {@inheritDoc}
         */
        public synchronized boolean remove(String blobId) throws Exception {
            Statement stmt = executeStmt(blobDeleteSQL, new Object[]{blobId});
            return stmt.getUpdateCount() == 1;
        }
    }
}