cm.java

jxta_src_2.41b jxta 2.41b 最新版源码 from www.jxta.org

            if (results.size() >= threshold) {
                return false;
            }
            if (LOG.isEnabledFor(Level.DEBUG)) {
                LOG.debug("Found " + val.toString() + " at " + pos);
            }
            Record record = null;

            try {
                record = cacheDB.readRecord(pos);
            } catch (DBException ex) {
                if (LOG.isEnabledFor(Level.WARN)) {
                    LOG.warn("Exception while reading indexed", ex);
                }
                return false;
            }
            if (record == null) {
                return true;
            }
            long exp = calcExpiration(record);

            if (exp <= 0) {
                // skip expired and private entries
                return true;
            }
            Long life = (Long) record.getMetaData(Record.LIFETIME);
            SrdiMessage.Entry entry = new SrdiMessage.Entry(key, val.toString(), (life.longValue() - System.currentTimeMillis()));

            if (LOG.isEnabledFor(Level.DEBUG)) {
                LOG.debug(" key [" + entry.key + "] value [" + entry.value + "] exp [" + entry.expiration + "]");
            }
            results.addElement(entry);
            return true;
        }
    }


    private final class SearchCallback implements BTreeCallback {

        private BTreeFiler cacheDB = null;
        private Indexer indexer = null;
        private int threshold;
        private Vector results;
        private Vector expirations;
        private boolean purge;

        SearchCallback(BTreeFiler cacheDB, Indexer indexer, Vector results, Vector expirations, int threshold) {
            this(cacheDB, indexer, results, expirations, threshold, false);
        }

        SearchCallback(BTreeFiler cacheDB, Indexer indexer, Vector results, Vector expirations, int threshold, boolean purge) {
            this.cacheDB = cacheDB;
            this.indexer = indexer;
            this.results = results;
            this.threshold = threshold;
            this.expirations = expirations;
            this.purge = purge;
        }

        /**
         *  {@inheritDoc}
         */
        public boolean indexInfo(Value val, long pos) {
            if (results.size() >= threshold) {
                return false;
            }
            if (LOG.isEnabledFor(Level.DEBUG)) {
                LOG.debug("Found " + val.toString() + " at " + pos);
            }
            Record record = null;

            try {
                record = cacheDB.readRecord(pos);
            } catch (DBException ex) {
                if (LOG.isEnabledFor(Level.WARN)) {
                    LOG.warn("Exception while reading indexed", ex);
                }
                return false;
            }
            if (record == null) {
                return true;
            }

            /* too noisy
             if (LOG.isEnabledFor(Level.DEBUG)) {
             LOG.debug("Search callback record " + record.toString());
             }
             */
            long exp = calcExpiration(record);

            if (exp < 0) {
                if (purge) {
                    try {
                        indexer.purge(pos);
                        cacheDB.deleteRecord(record.getKey());
                    } catch (DBException ex) {
                        if (LOG.isEnabledFor(Level.WARN)) {
                            LOG.warn("Exception while reading indexed", ex);
                        }
                    } catch (IOException ie) {
                        if (LOG.isEnabledFor(Level.WARN)) {
                            LOG.warn("Exception while reading indexed", ie);
                        }
                    }
                } else {
                    ++inconvenienceLevel;
                }
                return true;
            }

            if (expirations != null) {
                expirations.addElement(new Long(exp));
            }
            results.addElement(record.getValue().getInputStream());
            return true;
        }
    }


    private static final class removeCallback implements BTreeCallback {

        private BTreeFiler cacheDB = null;
        private Indexer indexer = null;

        removeCallback(BTreeFiler cacheDB, Indexer indexer) {
            this.cacheDB = cacheDB;
            this.indexer = indexer;
        }

        /**
         *  {@inheritDoc}
         */
        public boolean indexInfo(Value val, long pos) {
            Record record = null;

            try {
                record = cacheDB.readRecord(pos);
            } catch (DBException ex) {
                if (LOG.isEnabledFor(Level.WARN)) {
                    LOG.warn("Exception while reading record", ex);
                }
                return false;
            }
            if (record == null) {
                return true;
            }
            try {
                if (LOG.isEnabledFor(Level.DEBUG)) {
                    LOG.debug("Removing Record at position :" + pos);
                }
                indexer.purge(pos);
                cacheDB.deleteRecord(record.getKey());
            } catch (DBException ex) {
                if (LOG.isEnabledFor(Level.WARN)) {
                    LOG.warn("Exception while reading indexed", ex);
                }
            } catch (IOException ie) {
                if (LOG.isEnabledFor(Level.WARN)) {
                    LOG.warn("Exception while reading indexed", ie);
                }
            }
            return true;
        }
    }

    protected static IndexQuery getIndexQuery(String value) {

        int operator = IndexQuery.ANY;

        if (value == null) {
            return null;
        } else if (value.length() == 0 || "*".equals(value)) {
            return null;
        } else if (value.indexOf("*") < 0) {
            operator = IndexQuery.EQ;
        } else if (value.charAt(0) == '*' && value.charAt(value.length() - 1) != '*') {
            operator = IndexQuery.EW;
            value = value.substring(1, value.length());
        } else if (value.charAt(value.length() - 1) == '*' && value.charAt(0) != '*') {
            operator = IndexQuery.SW;
            value = value.substring(0, value.length() - 1);
        } else {
            operator = IndexQuery.BWX;
            value = value.substring(1, value.length() - 1);
        }
        if (LOG.isEnabledFor(Level.DEBUG)) {
            LOG.debug("Index query operator :" + operator);
        }
        return new IndexQuery(operator, new Value(value.toUpperCase()));
    }

    /**
     * Search and recovers documents that contains at least
     * a macthing pair of tag/value.
     *
     * @param  dn         contains the name of the folder on which to 
     *                    perform the search
     * @param  value      contains the value to search on.
     * @param  attribute  attribute to search on
     * @param  threshold  threshold
     * @return            Enumeration containing of all the documents names
     */
    public synchronized Vector search(String dn, String attribute,
            String value, int threshold,
            Vector expirations) {

        Vector res = new Vector();
        IndexQuery iq = getIndexQuery(value);

        try {
            indexer.search(iq, dn + attribute, new SearchCallback(cacheDB, indexer, res, expirations, threshold));
        } catch (Exception ex) {
            if (LOG.isEnabledFor(Level.WARN)) {
                LOG.warn("Exception while searching in index", ex);
            }
        }
        return res;
    }

    /**
     * returns all entries that are cached
     *
     * @param  dn  the relative dir name
     * @return     SrdiMessage.Entries
     */
    public synchronized Vector getEntries(String dn, boolean clearDeltas) {
        Vector res = new Vector();

        try {
            Map map = indexer.getIndexers();
            BTreeFiler listDB = indexer.getListDB();
            Iterator it = map.keySet().iterator();

            while (it != null && it.hasNext()) {
                String indexName = (String) it.next();

                // seperate the index name from attribute
                if (indexName.startsWith(dn)) {
                    String attr = indexName.substring((dn).length());
                    NameIndexer idxr = (NameIndexer) map.get(indexName);

                    idxr.query(null, new Indexer.SearchCallback(listDB, new EntriesCallback(cacheDB, res, attr, Integer.MAX_VALUE)));
                }
            }
        } catch (Exception ex) {
            if (LOG.isEnabledFor(Level.ERROR)) {
                LOG.error("Exception while searching in index", ex);
            }
        }

        if (clearDeltas) {
            clearDeltas(dn);
        }

        return res;
    }

    /**
     * returns all entries that are added since this method was last called
     *
     * @param  dn  the relative dir name
     * @return     SrdiMessage.Entries
     */
    public synchronized Vector getDeltas(String dn) {
        Vector result = new Vector();

        List deltas = (List) deltaMap.get(dn);

        if (deltas != null) {
            result.addAll(deltas);
            deltas.clear();
        }

        return result;
    }

    private synchronized void clearDeltas(String dn) {

        List deltas = (List) deltaMap.get(dn);

        if (deltas == null) {
            return;
        }
        deltas.clear();
    }

    private synchronized void addDelta(String dn, Map indexables, long exp) {

        if (trackDeltas) {
            Iterator eachIndex = indexables.entrySet().iterator();

            if (eachIndex.hasNext()) {
                List deltas = (List) deltaMap.get(dn);

                if (deltas == null) {
                    deltas = new ArrayList();
                    deltaMap.put(dn, deltas);
                }

                while (eachIndex.hasNext()) {
                    Map.Entry anEntry = (Map.Entry) eachIndex.next();
                    String attr = (String) anEntry.getKey();
                    String value = (String) anEntry.getValue();
                    SrdiMessage.Entry entry = new SrdiMessage.Entry(attr, value, exp);

                    deltas.add(entry);
                    if (LOG.isEnabledFor(Level.DEBUG)) {
                        LOG.debug("Added entry  :" + entry + " to deltas");
                    }
                }
            }
        }
    }

    public synchronized void setTrackDeltas(boolean trackDeltas) {

        this.trackDeltas = trackDeltas;

        if (!trackDeltas) {
            deltaMap.clear();
        }
    }

    /**
     * stop the cm
     */
    public synchronized void stop() {
        try {
            cacheDB.close();
            indexer.close();
            stop = true;
            notify();
        } catch (DBException ex) {
            if (LOG.isEnabledFor(Level.ERROR)) {
                LOG.error("Unable to close advertisments.tbl", ex);
            }
        }
    }

    /**
     *  {@inheritDoc}
     */
    public synchronized void run() {
        try {
            while (!stop) {
                try {

                    if (LOG.isEnabledFor(Level.DEBUG)) {
                        LOG.debug("waiting " + gcMinInterval + "ms before garbage collection");
                    }

                    wait(gcMinInterval);
                } catch (InterruptedException woken) {
                    Thread.interrupted();

                    if (LOG.isEnabledFor(Level.DEBUG)) {
                        LOG.debug("Thread interrupted", woken);
                    }
                }

                if (stop) {
                    // if asked to stop, exit
                    break;
                }

                if ((inconvenienceLevel > maxInconvenienceLevel) || (System.currentTimeMillis() > gcTime)) {

                    if (LOG.isEnabledFor(Level.DEBUG)) {
                        LOG.debug("Garbage collection started");
                    }

                    garbageCollect();
                    inconvenienceLevel = 0;
                    gcTime = System.currentTimeMillis() + gcMaxInterval;
                    if (LOG.isEnabledFor(Level.DEBUG)) {
                        LOG.debug("Garbage collection completed");
                    }
                }
            }
        } catch (Throwable all) {
            if (LOG.isEnabledFor(Level.FATAL)) {
                LOG.fatal("Uncaught Throwable in thread :" + Thread.currentThread().getName(), all);
            }
        }
        finally {
            gcThread = null;
        }
    }

    private synchronized void rebuildIndex()
        throws BTreeException, DBException, IOException {

        if (LOG.isEnabledFor(Level.INFO)) {
            LOG.info("Rebuilding indices");
        }

        String pattern = "*";
        IndexQuery any = new IndexQuery(IndexQuery.ANY, pattern);

        cacheDB.query(any, new RebuildIndexCallback(cacheDB, indexer));
    }

    private static final class RebuildIndexCallback implements BTreeCallback {

        private BTreeFiler database = null;
        private Indexer index = null;

        RebuildIndexCallback(BTreeFiler database, Indexer index) {
            this.database = database;
            this.index = index;
        }

        /**
         *  {@inheritDoc}
         */
        public boolean indexInfo(Value val, long pos) {
            try {
                Record record = database.readRecord(pos);

                if (record == null) {
                    return true;
                }

                InputStream is = record.getValue().getInputStream();
                Advertisement adv = AdvertisementFactory.newAdvertisement(MimeMediaType.XMLUTF8, is);
                Map indexables = getIndexfields(adv.getIndexFields(), (StructuredDocument) adv.getDocument(MimeMediaType.XMLUTF8));

                String dn = getDirName(adv);
                Map keyedIdx = addKey(dn, indexables);

                if (LOG.isEnabledFor(Level.DEBUG)) {
                    LOG.debug("Restoring index " + keyedIdx + " at " + pos);
                }
                index.addToIndex(keyedIdx, pos);
            } catch (Exception ex) {
                if (LOG.isEnabledFor(Level.WARN)) {
                    LOG.warn("Exception rebuilding index  at " + pos, ex);
                }
                return true;
            }

            return true;
        }
    }
}