databasefilesystem.java

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

        }
    }

    /**
     * {@inheritDoc}
     */
    public RandomAccessOutputStream getRandomAccessOutputStream(final String filePath)
            throws FileSystemException, UnsupportedOperationException {
        if (!initialized) {
            throw new IllegalStateException("not initialized");
        }

        FileSystemPathUtil.checkFormat(filePath);

        final String parentDir = FileSystemPathUtil.getParentDir(filePath);
        final String name = FileSystemPathUtil.getName(filePath);

        if (!isFolder(parentDir)) {
            throw new FileSystemException("path not found: " + parentDir);
        }

        if (isFolder(filePath)) {
            throw new FileSystemException("path denotes folder: " + filePath);
        }

        try {
            TransientFileFactory fileFactory = TransientFileFactory.getInstance();
            final File tmpFile = fileFactory.createTransientFile("bin", null, null);

            // @todo FIXME use java.sql.Blob

            if (isFile(filePath)) {
                // file entry exists, spool contents to temp file first
                InputStream in = getInputStream(filePath);
                OutputStream out = new FileOutputStream(tmpFile);
                try {
                    int read;
                    byte[] ba = new byte[8192];
                    while ((read = in.read(ba, 0, ba.length)) != -1) {
                        out.write(ba, 0, read);
                    }
                } finally {
                    out.close();
                    in.close();
                }
            }

            return new RandomAccessOutputStream() {
                private final RandomAccessFile raf =
                    new RandomAccessFile(tmpFile, "rw");

                public void close() throws IOException {
                    raf.close();

                    InputStream in = null;
                    try {
                        if (isFile(filePath)) {
                            synchronized (updateDataSQL) {
                                long length = tmpFile.length();
                                in = new FileInputStream(tmpFile);
                                executeStmt(updateDataSQL,
                                        new Object[]{
                                            new SizedInputStream(in, length),
                                            new Long(System.currentTimeMillis()),
                                            new Long(length),
                                            parentDir,
                                            name
                                        });
                            }
                        } else {
                            synchronized (insertFileSQL) {
                                long length = tmpFile.length();
                                in = new FileInputStream(tmpFile);
                                executeStmt(insertFileSQL,
                                        new Object[]{
                                            parentDir,
                                            name,
                                            new SizedInputStream(in, length),
                                            new Long(System.currentTimeMillis()),
                                            new Long(length)
                                        });
                            }
                        }

                    } catch (Exception e) {
                        throw new IOException(e.getMessage());
                    } finally {
                        if (in != null) {
                            in.close();
                        }
                        // temp file can now safely be removed
                        tmpFile.delete();
                    }
                }

                public void seek(long position) throws IOException {
                    raf.seek(position);
                }

                public void write(int b) throws IOException {
                    raf.write(b);
                }

                public void flush() /*throws IOException*/ {
                    // nop
                }

                public void write(byte[] b) throws IOException {
                    raf.write(b);
                }

                public void write(byte[] b, int off, int len) throws IOException {
                    raf.write(b, off, len);
                }
            };
        } catch (Exception e) {
            String msg = "failed to open output stream to file: " + filePath;
            log.error(msg, e);
            throw new FileSystemException(msg, e);
        }
    }

    /**
     * {@inheritDoc}
     */
    public void copy(String srcPath, String destPath) throws FileSystemException {
        if (!initialized) {
            throw new IllegalStateException("not initialized");
        }

        FileSystemPathUtil.checkFormat(srcPath);
        FileSystemPathUtil.checkFormat(destPath);

        if (isFolder(srcPath)) {
            // src is a folder
            copyDeepFolder(srcPath, destPath);
        } else {
            // src is a file
            copyFile(srcPath, destPath);
        }
    }

    /**
     * {@inheritDoc}
     */
    public void move(String srcPath, String destPath) throws FileSystemException {
        if (!initialized) {
            throw new IllegalStateException("not initialized");
        }

        FileSystemPathUtil.checkFormat(srcPath);
        FileSystemPathUtil.checkFormat(destPath);

        // @todo optimize move (use sql update stmts)
        copy(srcPath, destPath);
        if (isFile(srcPath)) {
            deleteFile(srcPath);
        } else {
            deleteFolder(srcPath);
        }
    }

    //----------------------------------< misc. helper methods & overridables >

    /**
     * Initializes the database connection used by this file system.
     * <p>
     * Subclasses should normally override the {@link #getConnection()}
     * method instead of this one. The default implementation calls
     * {@link #getConnection()} to get the database connection and disables
     * the autocommit feature.
     *
     * @throws Exception if an error occurs
     */
    protected void initConnection() throws Exception {
        con = getConnection();
        con.setAutoCommit(true);
    }

    /**
     * Abstract factory method for creating a new database connection. This
     * method is called by {@link #initConnection()} when the file system is
     * started. The returned connection should come with the default JDBC
     * settings, as the {@link #initConnection()} method will explicitly set
     * the <code>autoCommit</code> and other properties as needed.
     * <p>
     * Note that the returned database connection is kept during the entire
     * lifetime of the file system, after which it is closed by
     * {@link #close()} using the {@link #closeConnection(Connection)} method.
     *
     * @return new connection
     * @throws Exception if an error occurs
     */
    protected Connection getConnection() throws Exception {
        throw new UnsupportedOperationException("Override in a subclass!");
    }

    /**
     * Closes the given database connection. This method is called by
     * {@link #close()} to close the connection acquired using
     * {@link #getConnection()} when the file system was started.
     * <p>
     * The default implementation just calls the {@link Connection#close()}
     * method of the given connection, but subclasses can override this
     * method to provide more extensive database and connection cleanup.
     *
     * @param connection database connection
     * @throws SQLException if an error occurs
     */
    protected void closeConnection(Connection connection) throws SQLException {
        connection.close();
    }

    /**
     * Re-establishes the database connection. This method is called by
     * {@link #executeStmt(String, Object[])} after a <code>SQLException</code>
     * had been encountered.
     *
     * @return true if the connection could be successfully re-established,
     *         false otherwise.
     */
    protected synchronized boolean reestablishConnection() {
        // in any case try to shut down current connection
        // gracefully in order to avoid potential memory leaks

        // close shared prepared statements
        for (Iterator it = preparedStatements.values().iterator(); it.hasNext(); ) {
            closeStatement((PreparedStatement) it.next());
        }
        try {
            closeConnection(con);
        } catch (Exception ignore) {
        }

        // sleep for a while to give database a chance
        // to restart before a reconnect is attempted

        try {
            Thread.sleep(SLEEP_BEFORE_RECONNECT);
        } catch (InterruptedException ignore) {
        }

        // now try to re-establish connection

        try {
            initConnection();
            initPreparedStatements();
            return true;
        } catch (Exception e) {
            log.error("failed to re-establish connection", e);
            // reconnect failed
            return false;
        }
    }

    /**
     * Executes the given SQL statement with the specified parameters.
     * If a <code>SQLException</code> is encountered <i>one</i> attempt is made
     * to re-establish the database connection and re-execute the statement.
     *
     * @param sql    statement to execute
     * @param params parameters to set
     * @return the <code>Statement</code> object that had been executed
     * @throws SQLException if an error occurs
     */
    protected Statement executeStmt(String sql, Object[] params)
            throws SQLException {
        int trials = 2;
        while (true) {
            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;
                }
            }
        }
    }

    /**
     * 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 + "FSENTRY";
        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 = DatabaseFileSystem.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) {
                    // 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();
                }