oracle9persistencemanager.java

来自「jsr170接口的java实现。是个apache的开源项目。」· Java 代码 · 共 285 行

/*
 * 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.bundle;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.jackrabbit.core.persistence.PMContext;
import org.apache.jackrabbit.core.persistence.bundle.util.NodePropBundle;
import org.apache.jackrabbit.core.persistence.util.Serializer;
import org.apache.jackrabbit.core.state.ItemStateException;
import org.apache.jackrabbit.core.state.NodeReferences;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.lang.reflect.Method;
import java.sql.Blob;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;

/**
 * <code>OracleLegacyPersistenceManager</code> provides support for Oracle jdbc
 * drivers prior to version 10 which require special handling of BLOB data.
 * <p/>
 * Configuration:<br>
 * <ul>
 * <li>&lt;param name="{@link #setBundleCacheSize(String) bundleCacheSize}" value="8"/>
 * <li>&lt;param name="{@link #setConsistencyCheck(String) consistencyCheck}" value="false"/>
 * <li>&lt;param name="{@link #setMinBlobSize(String) minBlobSize}" value="16384"/>
 * <li>&lt;param name="{@link #setDriver(String) driver}" value="oracle.jdbc.OracleDriverr"/>
 * <li>&lt;param name="{@link #setUrl(String) url}" value="jdbc:oracle:thin:@127.0.0.1:1521:xe"/>
 * <li>&lt;param name="{@link #setUser(String) user}" value="crx"/>
 * <li>&lt;param name="{@link #setPassword(String) password}" value="crx"/>
 * <li>&lt;param name="{@link #setSchema(String) schema}" value="oracle"/>
 * <li>&lt;param name="{@link #setSchemaObjectPrefix(String) schemaObjectPrefix}" value="${wsp.name}_"/>
 * <li>&lt;param name="{@link #setErrorHandling(String) errorHandling}" value=""/>
 * </ul>
 */
public class Oracle9PersistenceManager extends OraclePersistenceManager {

    /**
     * the cvs/svn id
     */
    static final String CVS_ID = "$URL: http://svn.apache.org/repos/asf/jackrabbit/tags/1.3.3/jackrabbit-core/src/main/java/org/apache/jackrabbit/core/persistence/bundle/Oracle9PersistenceManager.java $ $Rev: 517625 $ $Date: 2007-03-13 11:59:30 +0200 (Tue, 13 Mar 2007) $";

    /**
     * the default logger
     */
    private static Logger log = LoggerFactory.getLogger(Oracle9PersistenceManager.class);

    private Class blobClass;
    private Integer DURATION_SESSION_CONSTANT;
    private Integer MODE_READWRITE_CONSTANT;

    public Oracle9PersistenceManager() {
    }

    //-----------------------------------< OraclePersistenceManager overrides >
    /**
     * {@inheritDoc}
     * <p/>
     * Retrieve the <code>oracle.sql.BLOB</code> class via reflection, and
     * initialize the values for the <code>DURATION_SESSION</code> and
     * <code>MODE_READWRITE</code> constants defined there.
     *
     * @see oracle.sql.BLOB#DURATION_SESSION
     * @see oracle.sql.BLOB#MODE_READWRITE
     */
    public void init(PMContext context) throws Exception {
        super.init(context);

        // initialize oracle.sql.BLOB class & constants

        // use the Connection object for using the exact same
        // class loader that the Oracle driver was loaded with
        blobClass = con.getClass().getClassLoader().loadClass("oracle.sql.BLOB");
        DURATION_SESSION_CONSTANT =
                new Integer(blobClass.getField("DURATION_SESSION").getInt(null));
        MODE_READWRITE_CONSTANT =
                new Integer(blobClass.getField("MODE_READWRITE").getInt(null));
    }

    /**
     * @inheritDoc
     */
    protected BundleDbPersistenceManager.CloseableBLOBStore createDBBlobStore(PMContext context) throws Exception {
        return new OracleBLOBStore();
    }

    /**
     * @inheritDoc
     */
    protected synchronized void storeBundle(NodePropBundle bundle)
            throws ItemStateException {
        PreparedStatement stmt = null;
        Blob blob = 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;
            }
            blob = createTemporaryBlob(new ByteArrayInputStream(out.toByteArray()));
            stmt.setBlob(1, blob);
            stmt.setBytes(2, bundle.getId().getUUID().getRawBytes());
            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);
            if (blob != null) {
                try {
                    freeTemporaryBlob(blob);
                } catch (Exception e1) {
                }
            }
        }
    }

    /**
     * @inheritDoc
     */
    public synchronized void store(NodeReferences refs)
            throws ItemStateException {
        if (!initialized) {
            throw new IllegalStateException("not initialized");
        }

        PreparedStatement stmt = null;
        Blob blob = null;
        try {
            // check if insert or update
            if (exists(refs.getId())) {
                stmt = nodeReferenceUpdate;
            } else {
                stmt = nodeReferenceInsert;
            }

            ByteArrayOutputStream out = new ByteArrayOutputStream(INITIAL_BUFFER_SIZE);
            // serialize references
            Serializer.serialize(refs, out);

            // we are synchronized on this instance, therefore we do not
            // not have to additionally synchronize on the preparedStatement

            blob = createTemporaryBlob(new ByteArrayInputStream(out.toByteArray()));
            stmt.setBlob(1, blob);
            stmt.setBytes(2, refs.getTargetId().getUUID().getRawBytes());
            stmt.execute();

            // there's no need to close a ByteArrayOutputStream
            //out.close();
        } catch (Exception e) {
            String msg = "failed to write property state: " + refs.getTargetId();
            log.error(msg, e);
            throw new ItemStateException(msg, e);
        } finally {
            resetStatement(stmt);
            if (blob != null) {
                try {
                    freeTemporaryBlob(blob);
                } catch (Exception e1) {
                }
            }
        }
    }

    //----------------------------------------< oracle-specific blob handling >
    /**
     * Creates a temporary oracle.sql.BLOB instance via reflection and spools
     * the contents of the specified stream.
     */
    protected Blob createTemporaryBlob(InputStream in) throws Exception {
        /*
        BLOB blob = BLOB.createTemporary(con, false, BLOB.DURATION_SESSION);
        blob.open(BLOB.MODE_READWRITE);
        OutputStream out = blob.getBinaryOutputStream();
        ...
        out.flush();
        out.close();
        blob.close();
        return blob;
        */
        Method createTemporary = blobClass.getMethod("createTemporary",
                new Class[]{Connection.class, Boolean.TYPE, Integer.TYPE});
        Object blob = createTemporary.invoke(null,
                new Object[]{con, Boolean.FALSE, DURATION_SESSION_CONSTANT});
        Method open = blobClass.getMethod("open", new Class[]{Integer.TYPE});
        open.invoke(blob, new Object[]{MODE_READWRITE_CONSTANT});
        Method getBinaryOutputStream = blobClass.getMethod("getBinaryOutputStream", new Class[0]);
        OutputStream out = (OutputStream) getBinaryOutputStream.invoke(blob, null);
        try {
            int read;
            byte[] buf = new byte[8192];
            while ((read = in.read(buf, 0, buf.length)) > -1) {
                out.write(buf, 0, read);
            }
        } finally {
            try {
                out.flush();
            } catch (IOException ioe) {
            }
            out.close();
        }
        Method close = blobClass.getMethod("close", new Class[0]);
        close.invoke(blob, null);
        return (Blob) blob;
    }

    /**
     * Frees a temporary oracle.sql.BLOB instance via reflection.
     */
    protected void freeTemporaryBlob(Object blob) throws Exception {
        // blob.freeTemporary();
        Method freeTemporary = blobClass.getMethod("freeTemporary", new Class[0]);
        freeTemporary.invoke(blob, null);
    }

    //--------------------------------------------------------< inner classes >
    class OracleBLOBStore extends DbBlobStore {

        public OracleBLOBStore() throws SQLException {
        }

        /**
         * {@inheritDoc}
         */
        public synchronized void put(String blobId, InputStream in, long size)
                throws Exception {
            PreparedStatement stmt = blobSelectExist;
            Blob blob = null;
            try {
                stmt.setString(1, blobId);
                stmt.execute();
                ResultSet rs = stmt.getResultSet();
                // a BLOB exists if the result has at least one entry
                boolean exists = rs.next();
                resetStatement(stmt);
                closeResultSet(rs);

                stmt = (exists) ? blobUpdate : blobInsert;

                blob = createTemporaryBlob(in);
                stmt.setBlob(1, blob);
                stmt.setString(2, blobId);
                stmt.executeUpdate();
            } finally {
                resetStatement(stmt);
                if (blob != null) {
                    try {
                        freeTemporaryBlob(blob);
                    } catch (Exception e1) {
                    }
                }
            }
        }
    }
}