bundledbpersistencemanager.java

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

                stmt = con.prepareStatement(
                        "select NODE_ID_HI, NODE_ID_LO, BUNDLE_DATA from "
                        + schemaObjectPrefix + "BUNDLE");
            }
            stmt.execute();
            rs = stmt.getResultSet();
            while (rs.next()) {
                NodeId id;
                Blob blob;
                if (getStorageModel() == SM_BINARY_KEYS) {
                    id = new NodeId(new UUID(rs.getBytes(1)));
                    blob = rs.getBlob(2);
                } else {
                    id = new NodeId(new UUID(rs.getLong(1), rs.getLong(2)));
                    blob = rs.getBlob(3);
                }
                din = new DataInputStream(blob.getBinaryStream());
                try {
                    NodePropBundle bundle = binding.readBundle(din, id);
                    Collection missingChildren = new ArrayList();
                    Iterator iter = bundle.getChildNodeEntries().iterator();
                    while (iter.hasNext()) {
                        NodePropBundle.ChildNodeEntry entry = (NodePropBundle.ChildNodeEntry) iter.next();
                        if (entry.getId().toString().endsWith("babecafebabe")) {
                            continue;
                        }
                        if (id.toString().endsWith("babecafebabe")) {
                            continue;
                        }
                        try {
                            NodePropBundle child = loadBundle(entry.getId());
                            if (child == null) {
                                log.error("NodeState " + id.getUUID() + " references unexistent child " + entry.getName() + " with id " + entry.getId().getUUID());
                                missingChildren.add(entry);
                            } else {
                                NodeId cp = child.getParentId();
                                if (cp == null) {
                                    log.error("ChildNode has invalid parent uuid: null");
                                } else if (!cp.equals(id)) {
                                    log.error("ChildNode has invalid parent uuid: " + cp + " (instead of " + id.getUUID() + ")");
                                }
                            }
                        } catch (ItemStateException e) {
                            log.error("Error while loading child node: " + e);
                        }
                    }
                    if (consistencyFix && !missingChildren.isEmpty()) {
                        Iterator iterator = missingChildren.iterator();
                        while (iterator.hasNext()) {
                            bundle.getChildNodeEntries().remove(iterator.next());
                        }
                        modifications.add(bundle);
                    }

                    NodeId parentId = bundle.getParentId();
                    if (parentId != null) {
                        if (!exists(parentId)) {
                            log.error("NodeState " + id + " references unexistent parent id " + parentId);
                        }
                    }
                } catch (IOException e) {
                    log.error("Error in bundle " + id + ": " + e);
                    din = new DataInputStream(blob.getBinaryStream());
                    binding.checkBundle(din);
                }
                count++;
                if (count % 1000 == 0) {
                    log.info(name + ": checked " + count + "/" + total + " bundles...");
                }
            }
        } catch (Exception e) {
            log.error("Error in bundle", e);
        } finally {
            closeStream(din);
            closeResultSet(rs);
            closeStatement(stmt);
        }

        if (consistencyFix && !modifications.isEmpty()) {
            log.info(name + ": Fixing " + modifications.size() + " inconsistent bundle(s)...");
            Iterator iterator = modifications.iterator();
            while (iterator.hasNext()) {
                NodePropBundle bundle = (NodePropBundle) iterator.next();
                try {
                    log.info(name + ": Fixing bundle " + bundle.getId());
                    bundle.markOld(); // use UPDATE instead of INSERT
                    storeBundle(bundle);
                } catch (ItemStateException e) {
                    log.error(name + ": Error storing fixed bundle: " + e);
                }
            }
        }

        log.info(name + ": checked " + count + "/" + total + " bundles.");
    }


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

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

        try {
            // close shared prepared statements
            closeStatement(bundleInsert);
            closeStatement(bundleUpdate);
            closeStatement(bundleSelect);
            closeStatement(bundleDelete);

            closeStatement(nodeReferenceInsert);
            closeStatement(nodeReferenceUpdate);
            closeStatement(nodeReferenceSelect);
            closeStatement(nodeReferenceDelete);

            if (nameIndex instanceof DbNameIndex) {
                ((DbNameIndex) nameIndex).close();
            }

            // close jdbc connection
            con.close();

            // close blob store
            blobStore.close();
            blobStore = null;
        } finally {
            initialized = false;
        }
    }

    /**
     * Sets the key parameters to the prepared statement, starting at
     * <code>pos</code> and returns the number of key parameters + pos.
     *
     * @param stmt the statement
     * @param uuid the uuid of the key
     * @param pos the position of the key parameter
     * @return the number of key parameters + <code>pos</code>
     * @throws SQLException if an SQL error occurs.
     */
    protected int setKey(PreparedStatement stmt, UUID uuid, int pos)
            throws SQLException {
        if (getStorageModel() == SM_BINARY_KEYS) {
            stmt.setBytes(pos++, uuid.getRawBytes());
        } else {
            stmt.setLong(pos++, uuid.getMostSignificantBits());
            stmt.setLong(pos++, uuid.getLeastSignificantBits());
        }
        return pos;
    }

    /**
     * {@inheritDoc}
     */
    protected synchronized NodePropBundle loadBundle(NodeId id)
            throws ItemStateException {
        PreparedStatement stmt = bundleSelect;
        ResultSet rs = null;
        InputStream in = null;
        try {
            setKey(stmt, id.getUUID(), 1);
            stmt.execute();
            rs = stmt.getResultSet();
            if (!rs.next()) {
                return null;
            }
            Blob b = rs.getBlob(1);
            // JCR-1039: pre-fetch/buffer blob data
            long length = b.length();
            byte[] bytes = new byte[(int) length];
            in = b.getBinaryStream();
            int read, pos = 0;
            while ((read = in.read(bytes, pos, bytes.length - pos)) > 0) {
                pos += read;
            }
            DataInputStream din = new DataInputStream(new ByteArrayInputStream(bytes));
            NodePropBundle bundle = binding.readBundle(din, id);
            bundle.setSize(length);
            return bundle;
        } catch (Exception e) {
            String msg = "failed to read bundle: " + id + ": " + e;
            log.error(msg);
            throw new ItemStateException(msg, e);
        } finally {
            closeStream(in);
            closeResultSet(rs);
            resetStatement(stmt);
        }
    }

    /**
     * {@inheritDoc}
     */
    protected synchronized boolean existsBundle(NodeId id) throws ItemStateException {
        PreparedStatement stmt = bundleSelect;
        ResultSet rs = null;
        try {
            setKey(stmt, id.getUUID(), 1);
            stmt.execute();
            rs = stmt.getResultSet();

            // a bundle exists, if the result has at least one entry
            return rs.next();
        } catch (Exception e) {
            String msg = "failed to check existence of bundle: " + id;
            log.error(msg, e);
            throw new ItemStateException(msg, e);
        } finally {
            closeResultSet(rs);
            resetStatement(stmt);
        }
    }

    /**
     * {@inheritDoc}
     */
    protected synchronized void storeBundle(NodePropBundle bundle) throws ItemStateException {
        PreparedStatement stmt = null;
        try {
            ByteArrayOutputStream out = new ByteArrayOutputStream(INITIAL_BUFFER_SIZE);
            DataOutputStream dout = new DataOutputStream(out);
            binding.writeBundle(dout, bundle);
            dout.close();

            if (bundle.isNew()) {
                stmt = bundleInsert;
            } else {
                stmt = bundleUpdate;
            }
            stmt.setBytes(1, out.toByteArray());
            setKey(stmt, bundle.getId().getUUID(), 2);
            stmt.execute();
        } catch (Exception e) {
            String msg = "failed to write bundle: " + bundle.getId();
            log.error(msg, e);
            throw new ItemStateException(msg, e);
        } finally {
            resetStatement(stmt);
        }
    }

    /**
     * {@inheritDoc}
     */
    protected synchronized void destroyBundle(NodePropBundle bundle) throws ItemStateException {
        PreparedStatement stmt = bundleDelete;
        try {
            setKey(stmt, bundle.getId().getUUID(), 1);
            stmt.execute();
            // also delete all
            bundle.removeAllProperties();
        } catch (Exception e) {
            if (e instanceof NoSuchItemStateException) {
                throw (NoSuchItemStateException) e;
            }
            String msg = "failed to delete bundle: " + bundle.getId();
            log.error(msg, e);
            throw new ItemStateException(msg, e);
        } finally {
            resetStatement(stmt);
        }
    }

    /**
     * {@inheritDoc}
     */
    public synchronized NodeReferences load(NodeReferencesId targetId)
            throws NoSuchItemStateException, ItemStateException {
        if (!initialized) {
            throw new IllegalStateException("not initialized");
        }

        PreparedStatement stmt = nodeReferenceSelect;
        ResultSet rs = null;
        InputStream in = null;
        try {
            setKey(stmt, targetId.getTargetId().getUUID(), 1);
            stmt.execute();
            rs = stmt.getResultSet();
            if (!rs.next()) {
                throw new NoSuchItemStateException(targetId.toString());
            }

            in = rs.getBinaryStream(1);
            NodeReferences refs = new NodeReferences(targetId);
            Serializer.deserialize(refs, in);

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

    /**
     * This method uses shared <code>PreparedStatements</code>, which must
     * be used 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
     * statement must be synchronized.