cm.java

JXTA&#8482 is a set of open, generalized peer-to-peer (P2P) protocols that allow any networked devi

/*
 * Copyright (c) 2001-2007 Sun Microsystems, Inc.  All rights reserved.
 *  
 *  The Sun Project JXTA(TM) Software License
 *  
 *  Redistribution and use in source and binary forms, with or without 
 *  modification, are permitted provided that the following conditions are met:
 *  
 *  1. Redistributions of source code must retain the above copyright notice,
 *     this list of conditions and the following disclaimer.
 *  
 *  2. Redistributions in binary form must reproduce the above copyright notice, 
 *     this list of conditions and the following disclaimer in the documentation 
 *     and/or other materials provided with the distribution.
 *  
 *  3. The end-user documentation included with the redistribution, if any, must 
 *     include the following acknowledgment: "This product includes software 
 *     developed by Sun Microsystems, Inc. for JXTA(TM) technology." 
 *     Alternately, this acknowledgment may appear in the software itself, if 
 *     and wherever such third-party acknowledgments normally appear.
 *  
 *  4. The names "Sun", "Sun Microsystems, Inc.", "JXTA" and "Project JXTA" must 
 *     not be used to endorse or promote products derived from this software 
 *     without prior written permission. For written permission, please contact 
 *     Project JXTA at http://www.jxta.org.
 *  
 *  5. Products derived from this software may not be called "JXTA", nor may 
 *     "JXTA" appear in their name, without prior written permission of Sun.
 *  
 *  THIS SOFTWARE IS PROVIDED ``AS IS'' AND ANY EXPRESSED OR IMPLIED WARRANTIES,
 *  INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND 
 *  FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL SUN 
 *  MICROSYSTEMS OR ITS CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, 
 *  INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT 
 *  LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, 
 *  OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF 
 *  LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING 
 *  NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, 
 *  EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 *  
 *  JXTA is a registered trademark of Sun Microsystems, Inc. in the United 
 *  States and other countries.
 *  
 *  Please see the license information page at :
 *  <http://www.jxta.org/project/www/license.html> for instructions on use of 
 *  the license in source files.
 *  
 *  ====================================================================
 *  
 *  This software consists of voluntary contributions made by many individuals 
 *  on behalf of Project JXTA. For more information on Project JXTA, please see 
 *  http://www.jxta.org.
 *  
 *  This license is based on the BSD license adopted by the Apache Foundation. 
 */
package net.jxta.impl.cm;

import net.jxta.discovery.DiscoveryService;
import net.jxta.document.Advertisement;
import net.jxta.document.AdvertisementFactory;
import net.jxta.document.Element;
import net.jxta.document.MimeMediaType;
import net.jxta.document.StructuredDocument;
import net.jxta.document.StructuredDocumentFactory;
import net.jxta.document.StructuredTextDocument;
import net.jxta.document.XMLDocument;
import net.jxta.impl.util.JxtaHash;
import net.jxta.impl.util.TimeUtils;
import net.jxta.impl.xindice.core.DBException;
import net.jxta.impl.xindice.core.data.Key;
import net.jxta.impl.xindice.core.data.Record;
import net.jxta.impl.xindice.core.data.Value;
import net.jxta.impl.xindice.core.filer.BTreeCallback;
import net.jxta.impl.xindice.core.filer.BTreeFiler;
import net.jxta.impl.xindice.core.indexer.IndexQuery;
import net.jxta.impl.xindice.core.indexer.NameIndexer;
import net.jxta.logging.Logging;
import net.jxta.protocol.PeerAdvertisement;
import net.jxta.protocol.PeerGroupAdvertisement;
import net.jxta.protocol.SrdiMessage;

import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.StringWriter;
import java.lang.reflect.UndeclaredThrowableException;
import java.math.BigInteger;
import java.net.URI;
import java.text.MessageFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.ResourceBundle;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * This class implements a limited document caching mechanism
 * intended to provide cache for services that have a need for cache
 * to search and exchange jxta documents.
 * <p/>
 * Only Core Services are intended to use this mechanism.
 */
public final class Cm implements Runnable {

    /**
     * Logger.
     */
    private final static Logger LOG = Logger.getLogger(Cm.class.getName());

    /**
     * the name we will use for the base directory
     */
    final File ROOTDIRBASE;

    /**
     * adv types
     */
    private static final String[] DIRNAME = {"Peers", "Groups", "Adv", "Raw"};

    // garbage collect once an hour
    public static final long DEFAULT_GC_MAX_INTERVAL = 1 * TimeUtils.ANHOUR;

    /*
     *  expiration db
     */
    private BTreeFiler cacheDB = null;
    private Indexer indexer = null;
    private final static String databaseFileName = "advertisements";

    private boolean stop = false;

    private boolean trackDeltas = false;
    private final Map<String, List<SrdiMessage.Entry>> deltaMap = new HashMap<String, List<SrdiMessage.Entry>>(3);

    /**
     * file descriptor for the root of the cm
     */
    protected File rootDir;

    private Thread gcThread = null;
    private long gcTime = 0;
    private final long gcMinInterval = 1000L * 60L;
    private long gcMaxInterval = DEFAULT_GC_MAX_INTERVAL;
    
    
    private final int maxInconvenienceLevel = 1000;
    private volatile int inconvenienceLevel = 0;

    /**
     * Constructor for cm
     *
     * @param areaName  the name of the cm sub-dir to create
     *                  <p/>
     *                  NOTE: Default garbage interval once an hour
     * @param storeRoot store root dir
     */
    public Cm(URI storeRoot, String areaName) {
        // Default garbage collect once an hour
        this(Thread.currentThread().getThreadGroup(), storeRoot, areaName, DEFAULT_GC_MAX_INTERVAL, false);
    }

    /**
     * Constructor for cm
     *
     * @param threadGroup     the thread group
     * @param storeRoot   persistence location
     * @param gcinterval  garbage collect max interval
     * @param trackDeltas when true deltas are tracked
     * @param areaName    storage area name
     */
    public Cm(ThreadGroup threadGroup, URI storeRoot, String areaName, long gcinterval, boolean trackDeltas) {
        this.trackDeltas = trackDeltas;
        this.gcMaxInterval = gcinterval;
        this.gcTime = System.currentTimeMillis() + gcMaxInterval;

        ROOTDIRBASE = new File(new File(storeRoot), "cm");

        try {
            rootDir = new File(ROOTDIRBASE, areaName);
            rootDir = new File(rootDir.getAbsolutePath());
            if (!rootDir.exists()) {
                // We need to create the directory
                if (!rootDir.mkdirs()) {
                    throw new RuntimeException("Cm cannot create directory " + rootDir);
                }
            }

            /*
             * to avoid inconsistent database state, it is highly recommended that
             * checkpoint is true by default, which causes fd.sync() on every write
             * operation.  In transitory caches such as SrdiCache it makes perfect sense
             */
            boolean chkPoint = true;
            ResourceBundle jxtaRsrcs = ResourceBundle.getBundle("net.jxta.user");
            String checkpointStr = jxtaRsrcs.getString("impl.cm.defferedcheckpoint");

            if (checkpointStr != null) {
                chkPoint = !(checkpointStr.equalsIgnoreCase("true"));
            }

            // Storage
            cacheDB = new BTreeFiler();
            // no deffered checkpoint
            cacheDB.setSync(chkPoint);
            cacheDB.setLocation(rootDir.getAbsolutePath(), databaseFileName);

            if (!cacheDB.open()) {
                cacheDB.create();
                // now open it
                cacheDB.open();
            }

            // Index
            indexer = new Indexer(chkPoint);
            indexer.setLocation(rootDir.getAbsolutePath(), databaseFileName);

            if (!indexer.open()) {
                indexer.create();
                // now open it
                indexer.open();
            }

            if (System.getProperty("net.jxta.impl.cm.index.rebuild") != null) {
                rebuildIndex();
            }
            gcThread = new Thread(threadGroup, this, "CM GC Thread interval : " + gcMinInterval);
            gcThread.setDaemon(true);
            gcThread.start();

            if (Logging.SHOW_CONFIG && LOG.isLoggable(Level.CONFIG)) {
                LOG.config("Instantiated Cm for: " + rootDir.getAbsolutePath());
            }
        } catch (DBException de) {
            if (Logging.SHOW_SEVERE && LOG.isLoggable(Level.SEVERE)) {
                LOG.log(Level.SEVERE, "Unable to Initialize databases", de);
            }
            throw new UndeclaredThrowableException(de, "Unable to Initialize databases");
        } catch (Throwable e) {
            if (Logging.SHOW_SEVERE && LOG.isLoggable(Level.SEVERE)) {
                LOG.log(Level.SEVERE, "Unable to create Cm", e);
            }
            if (e instanceof RuntimeException) {
                throw (RuntimeException) e;
            } else if (e instanceof Error) {
                throw (Error) e;
            } else {
                throw new UndeclaredThrowableException(e, "Unable to create Cm");
            }
        }
    }

    @Override
    public String toString() {
        return "CM for " + rootDir.getAbsolutePath() + "[" + super.toString() + "]";
    }

    private static String getDirName(Advertisement adv) {
        if (adv instanceof PeerAdvertisement) {
            return DIRNAME[DiscoveryService.PEER];
        } else if (adv instanceof PeerGroupAdvertisement) {
            return DIRNAME[DiscoveryService.GROUP];
        }
        return DIRNAME[DiscoveryService.ADV];
    }

    /**
     * Generates a random file name using doc hashcode
     *
     * @param doc to hash to generate a unique name
     * @return String a random file name
     */
    public static String createTmpName(StructuredTextDocument doc) {
        try {
            StringWriter out = new StringWriter();

            doc.sendToWriter(out);
            out.close();

            JxtaHash digester = new JxtaHash(out.toString());
            BigInteger hash = digester.getDigestInteger();

            if (hash.compareTo(BigInteger.ZERO) < 0) {
                hash = hash.negate();
            }
            return "cm" + hash.toString(16);
        } catch (IOException ex) {
            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.log(Level.WARNING, "Exception creating tmp name: ", ex);
            }
            throw new IllegalStateException("Could not generate name from document");
        }
    }

    /**
     * Gets the list of all the files into the given folder
     *
     * @param dn          contains the name of the folder
     * @param threshold   the max number of results
     * @param expirations List to contain expirations
     * @return List Strings containing the name of the
     *         files
     */
    public List<InputStream> getRecords(String dn, int threshold, List<Long> expirations) {
        return getRecords(dn, threshold, expirations, false);
    }

    public synchronized List<InputStream> getRecords(String dn, int threshold, List<Long> expirations, boolean purge) {
        List<InputStream> res = new ArrayList<InputStream>();

        if (dn == null) {
            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.fine("null directory name");
            }
            return res;
        } else {
            IndexQuery iq = new IndexQuery(IndexQuery.SW, new Value(dn));
            try {
                cacheDB.query(iq, new SearchCallback(cacheDB, indexer, res, expirations, threshold, purge));
            } catch (DBException dbe) {
                if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                    LOG.log(Level.FINE, "Exception during getRecords(): ", dbe);
                }
            } catch (IOException ie) {
                if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                    LOG.log(Level.FINE, "Exception during getRecords(): ", ie);
                }
            }
            return res;
        }
    }

    public void garbageCollect() {
        // calling getRecords is good enough since it removes
        // expired entries
        Map map = indexer.getIndexers();
        Iterator it = map.keySet().iterator();
        long t0;

        while (it != null && it.hasNext()) {
            t0 = System.currentTimeMillis();
            String indexName = (String) it.next();
            getRecords(indexName, Integer.MAX_VALUE, null, true);
            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.fine("Cm garbageCollect :" + indexName + " in :" + (System.currentTimeMillis() - t0));
            }
        }
    }

    /**
     * Returns the relative time in milliseconds at which the file
     * will expire.
     *
     * @param dn contains the name of the folder
     * @param fn contains the name of the file
     * @return the absolute time in milliseconds at which this
     *         document will expire. -1 is returned if the file is not
     *         recognized or already expired.
     */
    public synchronized long getLifetime(String dn, String fn) {
        try {
            Key key = new Key(dn + "/" + fn);
            Record record = cacheDB.readRecord(key);

            if (record == null) {
                return -1;
            }
            Long life = (Long) record.getMetaData(Record.LIFETIME);

            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.fine("Lifetime for :" + fn + "  " + life.toString());
            }
            if (life < System.currentTimeMillis()) {
                if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                    LOG.fine("Removing expired record :" + fn);
                }
                try {
                    remove(dn, fn);
                } catch (IOException e) {
                    if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                        LOG.log(Level.FINE, "Failed to remove record", e);
                    }
                }
            }
            return TimeUtils.toRelativeTimeMillis(life);
        } catch (DBException de) {
            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.log(Level.WARNING, "failed to remove " + dn + "/" + fn, de);
            }
            return -1;
        }
    }

    /**
     * Returns the maximum duration in milliseconds for which this
     * document should cached by those other than the publisher. This
     * value is either the cache lifetime or the remaining lifetime
     * of the document, whichever is less.
     *