You are viewing a plain text version of this content. The canonical link for it is here.

Posted to dev@jackrabbit.apache.org by Hui Lin <hl...@consumer.org> on 2012/06/06 16:14:58 UTC

Mongodb bundle persistence manager

hey folks, i am working on mongodb persistence manager in CQ. I implemented
all the methods but the repository shutdown due to unknown error during the
startup. Can anybody help me look at the code?


import com.mongodb.*;
import com.mongodb.gridfs.GridFS;
import com.mongodb.gridfs.GridFSDBFile;
import com.mongodb.gridfs.GridFSInputFile;
import org.apache.jackrabbit.core.id.NodeId;
import org.apache.jackrabbit.core.id.PropertyId;
import org.apache.jackrabbit.core.persistence.PMContext;
import
org.apache.jackrabbit.core.persistence.bundle.AbstractBundlePersistenceManager;
import org.apache.jackrabbit.core.persistence.util.*;
import org.apache.jackrabbit.core.state.ItemStateException;
import org.apache.jackrabbit.core.state.NoSuchItemStateException;
import org.apache.jackrabbit.core.state.NodeReferences;
import org.bson.types.Binary;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.jcr.RepositoryException;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;

/**
 * Created with IntelliJ IDEA.
 * Date: 6/4/12
 * Time: 1:55 PM
 * To change this template use File | Settings | File Templates.
 */


public class MongoBundlePersistenceManager extends
AbstractBundlePersistenceManager {

    protected DB itsDb;
    protected DBCollection itsNodeCollection;


    /**
     * flag for error handling
     */
    protected ErrorHandling errorHandling = new ErrorHandling();
    protected BundleBinding binding;
    /**
     * the minimum size of a property until it gets written to the blob
store
     */
    private int minBlobSize = 0x1000;

    private  String itsSchema;

    /** the default logger */
    private static Logger log =
LoggerFactory.getLogger(org.apache.jackrabbit.core.persistence.mongo.MongoBundlePersistenceManager.class);
    protected DBCollection itsRefCollection;
    protected BLOBStore itsBlobStore;

    @Override
    protected NodePropBundle loadBundle(NodeId id) throws
ItemStateException {

        BasicDBObject query = new BasicDBObject();
        query.put("_id", id.toString());
        BasicDBObject theDBObject = (BasicDBObject)
itsNodeCollection.findOne(query);
        if (theDBObject == null) {
            return null;
        }

        Binary theD = (Binary) theDBObject.get("d");
        InputStream in = new ByteArrayInputStream(theD.getData());
        try {
            return binding.readBundle(in, id);
        } catch (IOException e) {
            String msg = "failed to read bundle: " + id + ": " + e;
            log.error(msg,e);
            throw new ItemStateException(msg, e);
        }
    }

    /** initial size of buffer used to serialize objects */
    protected static final int INITIAL_BUFFER_SIZE = 1024;
    @Override
    protected synchronized void storeBundle(NodePropBundle bundle) throws
ItemStateException {
        try {
            DBObject theQuery = new BasicDBObject();
            theQuery.put("_id", bundle.getId().toString());

            DBObject theDbObject = new BasicDBObject();
            theDbObject.put("_id", bundle.getId().toString());
            ByteArrayOutputStream out = new
ByteArrayOutputStream(INITIAL_BUFFER_SIZE);
            binding.writeBundle(out, bundle);
            theDbObject.put("d", out.toByteArray());

            if( bundle.getParentId() != null) {
                theDbObject.put("p", bundle.getParentId().toString());
            }

            itsNodeCollection.update(theQuery, theDbObject, true, false);

        } catch (IOException e) {
            String msg = "failed to write bundle: " + bundle.getId() + ": "
+ e;
            log.error(msg, e);
            throw new ItemStateException(msg, e);
        }
    }

    @Override
    protected synchronized void destroyBundle(NodePropBundle bundle) throws
ItemStateException {
        //todo: delete entire subtree in single query
        DBObject theDBObject = new BasicDBObject();
        theDBObject.put("_id", bundle.getId().toString());
        itsNodeCollection.remove(theDBObject);
    }

    @Override
    protected synchronized void destroy(NodeReferences refs) throws
ItemStateException {
        DBObject theDBObject = new BasicDBObject();
        theDBObject.put("_id", refs.getTargetId().toString());
        itsRefCollection.remove(theDBObject);
    }

    @Override
    protected synchronized void store(NodeReferences refs) throws
ItemStateException {

        DBObject theQuery = new BasicDBObject();
        theQuery.put("_id", refs.getTargetId().toString());

        DBObject theRef = new BasicDBObject();
        theRef.put("_id", refs.getTargetId().toString());

        ByteArrayOutputStream out = new
ByteArrayOutputStream(INITIAL_BUFFER_SIZE);
        // serialize references
        try {

org.apache.jackrabbit.core.persistence.util.Serializer.serialize(refs, out);
            theRef.put("d", out.toByteArray());
            itsRefCollection.update(theQuery, theRef, true, false);
        } catch (Exception e) {
            String msg = "failed to write " + refs;
            log.error(msg, e);
            throw new ItemStateException(msg, e);
        }


    }


    public void init(PMContext inContext) throws Exception {
        super.init(inContext);
        log.info("init mongo db persistance manager: " +
getClass().getName());
        Mongo m = new Mongo("localhost");
        itsDb = m.getDB(getSchema());
        itsNodeCollection = itsDb.getCollection("node");
        itsRefCollection = itsDb.getCollection("ref");
        itsBlobStore = new MongoGridFSBLOBStore();
        // load namespaces
        binding = new BundleBinding(errorHandling, itsBlobStore,
getNsIndex(), getNameIndex(), context.getDataStore());
        binding.setMinBlobSize(minBlobSize);
        log.info("complete init mongo db persistance manager: " +
getClass().getName());
    }

    public NodeReferences loadReferencesTo(NodeId id) throws
NoSuchItemStateException, ItemStateException {

        BasicDBObject theQuery = new BasicDBObject();
        theQuery.put("_id", id.toString());

        BasicDBObject theRefDb = (BasicDBObject)
itsRefCollection.findOne(theQuery);
        if(theRefDb == null) {
            throw new NoSuchItemStateException(id.toString());
        }

        Binary theD = (Binary) theRefDb.get("d");
        InputStream in = new ByteArrayInputStream(theD.getData());

        NodeReferences theRef = new NodeReferences(id);
        try {

org.apache.jackrabbit.core.persistence.util.Serializer.deserialize(theRef,
in);
            return theRef;
        } catch (Exception e) {
            if (e instanceof NoSuchItemStateException) {
                throw (NoSuchItemStateException) e;
            }
            String msg = "failed to read references: " + id;
            log.error(msg, e);
            throw new ItemStateException(msg, e);
        }
    }

    public synchronized boolean existsReferencesTo(NodeId targetId) throws
ItemStateException {

        BasicDBObject theDBObject = new BasicDBObject();
        theDBObject.put("_id", targetId.toString());
        BasicDBObject theRefDb = (BasicDBObject)
itsRefCollection.findOne(theDBObject);
        return theRefDb != null;
    }

    public Iterable<NodeId> getAllNodeIds(NodeId after, int maxCount)
throws ItemStateException, RepositoryException {
        DBObject theSort = new BasicDBObject();
        theSort.put("_id", "1");

        DBObject theQuery = new BasicDBObject();
        theQuery.put("_id", after.toString());

        theQuery =        QueryBuilder.start().greaterThan(theQuery).get();
        DBCursor theCursor =
itsRefCollection.find(theQuery).sort(theSort).limit(maxCount);

        ArrayList<NodeId> result = new ArrayList<NodeId>();

        for (DBObject theDBObject : theCursor) {
            result.add(NodeId.valueOf(theDBObject.get("_id").toString()));
        }

        return result;
    }


    @Override
    protected BLOBStore getBlobStore() {
        return itsBlobStore;  //To change body of implemented methods use
File | Settings | File Templates.
    }

    public String getSchema() {
        return itsSchema;
    }

    public void setSchema(String inSchema) {
        itsSchema = inSchema;
    }


    private  class MongoGridFSBLOBStore implements BLOBStore {

        private GridFS itsGripFS;

        public MongoGridFSBLOBStore() {

            itsGripFS = new GridFS( itsDb );

        }

        public String createId(PropertyId id, int index) {

            StringBuffer buf = new StringBuffer();
                      buf.append(id.getParentId().toString());
                      buf.append('.');

buf.append(getNsIndex().stringToIndex(id.getName().getNamespaceURI()));
                      buf.append('.');

buf.append(getNameIndex().stringToIndex(id.getName().getLocalName()));
                      buf.append('.');
                      buf.append(index);
                      return buf.toString();
        }

        public void put(String blobId, InputStream in, long size) throws
Exception {
            GridFSInputFile theGridFSInputFile = itsGripFS.createFile( in,
true );
            theGridFSInputFile.put("_id", blobId);
            theGridFSInputFile.save();


        }

        public InputStream get(String blobId) throws Exception {
            GridFSDBFile out = itsGripFS.findOne( new BasicDBObject( "_id"
, blobId ) );
            return out.getInputStream();
        }

        public boolean remove(String blobId) throws Exception {
           itsGripFS.remove( new BasicDBObject( "_id" , blobId ) );
            return true;
        }
    }
}


- Steven



*****
This e-mail message is intended only for the designated recipient(s) named above. The information contained in this e-mail and any attachments may be confidential or legally privileged. If you are not the intended recipient, you may not review, retain, copy, redistribute or use this e-mail or any attachment for any purpose, or disclose all or any part of its contents. If you have received this e-mail in error, please immediately notify the sender by reply e-mail and permanently delete this e-mail and any attachments from your computer system.
*****