cm.java

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

        return tmp;
    }

    private static final class EntriesCallback implements BTreeCallback {

        private BTreeFiler cacheDB = null;
        private int threshold;
        private List<SrdiMessage.Entry> results;
        private String key;

        EntriesCallback(BTreeFiler cacheDB, List<SrdiMessage.Entry> results, String key, int threshold) {
            this.cacheDB = cacheDB;
            this.results = results;
            this.key = key;
            this.threshold = threshold;
        }

        /**
         * {@inheritDoc}
         */
        public boolean indexInfo(Value val, long pos) {
            if (results.size() >= threshold) {
                return false;
            }
            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.fine("Found " + val.toString() + " at " + pos);
            }

            Record record;
            try {
                record = cacheDB.readRecord(pos);
            } catch (DBException ex) {
                if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                    LOG.log(Level.WARNING, "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 - System.currentTimeMillis());

            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.fine(" key [" + entry.key + "] value [" + entry.value + "] exp [" + entry.expiration + "]");
            }
            results.add(entry);
            return true;
        }
    }


    private final class SearchCallback implements BTreeCallback {

        private BTreeFiler cacheDB = null;
        private Indexer indexer = null;
        private int threshold;
        private List<InputStream> results;
        private List<Long> expirations;
        private boolean purge;

        SearchCallback(BTreeFiler cacheDB, Indexer indexer, List<InputStream> results, List<Long> expirations, int threshold) {
            this(cacheDB, indexer, results, expirations, threshold, false);
        }

        SearchCallback(BTreeFiler cacheDB, Indexer indexer, List<InputStream> results, List<Long> 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 (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.fine("Found " + val.toString() + " at " + pos);
            }

            Record record;
            try {
                record = cacheDB.readRecord(pos);
            } catch (DBException ex) {
                if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                    LOG.log(Level.WARNING, "Exception while reading indexed", ex);
                }
                return false;
            }

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

            if (Logging.SHOW_FINER && LOG.isLoggable(Level.FINEST)) {
                LOG.finest("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 (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                            LOG.log(Level.WARNING, "Exception while reading indexed", ex);
                        }
                    } catch (IOException ie) {
                        if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                            LOG.log(Level.WARNING, "Exception while reading indexed", ie);
                        }
                    }
                } else {
                    ++inconvenienceLevel;
                }
                return true;
            }

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

    protected static IndexQuery getIndexQuery(String value) {

        int operator;

        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 (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
            LOG.fine("Index query operator :" + operator);
        }
        return new IndexQuery(operator, new Value(value));
    }

    /**
     * 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
     * @param expirations List to contain expirations
     * @return Enumeration containing of all the documents names
     */
    public synchronized List<InputStream> search(String dn, String attribute, String value, int threshold, List<Long> expirations) {
        List<InputStream> res = new ArrayList<InputStream>();
        IndexQuery iq = getIndexQuery(value);
        try {
            indexer.search(iq, dn + attribute, new SearchCallback(cacheDB, indexer, res, expirations, threshold));
        } catch (Exception ex) {
            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.log(Level.WARNING, "Exception while searching in index", ex);
            }
        }
        return res;
    }

    /**
     * returns all entries that are cached
     *
     * @param dn          the relative dir name
     * @param clearDeltas if true clears the delta cache
     * @return SrdiMessage.Entries
     */
    public synchronized List<SrdiMessage.Entry> getEntries(String dn, boolean clearDeltas) {
        List<SrdiMessage.Entry> res = new ArrayList<SrdiMessage.Entry>();
        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 (Logging.SHOW_SEVERE && LOG.isLoggable(Level.SEVERE)) {
                LOG.log(Level.SEVERE, "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 List<SrdiMessage.Entry> getDeltas(String dn) {
        List<SrdiMessage.Entry> result = new ArrayList<SrdiMessage.Entry>();
        List<SrdiMessage.Entry> deltas = deltaMap.get(dn);

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

    private synchronized void clearDeltas(String dn) {

        List<SrdiMessage.Entry> deltas = deltaMap.get(dn);

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

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

        if (trackDeltas) {
            Iterator<Map.Entry<String, String>> eachIndex = indexables.entrySet().iterator();

            if (eachIndex.hasNext()) {
                List<SrdiMessage.Entry> deltas = deltaMap.get(dn);

                if (deltas == null) {
                    deltas = new ArrayList<SrdiMessage.Entry>();
                    deltaMap.put(dn, deltas);
                }
                while (eachIndex.hasNext()) {
                    Map.Entry<String, String> anEntry = eachIndex.next();
                    String attr = anEntry.getKey();
                    String value = anEntry.getValue();
                    SrdiMessage.Entry entry = new SrdiMessage.Entry(attr, value, exp);

                    deltas.add(entry);
                    if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                        LOG.fine("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 (Logging.SHOW_SEVERE && LOG.isLoggable(Level.SEVERE)) {
                LOG.log(Level.SEVERE, "Unable to close advertisments.tbl", ex);
            }
        }
    }

    /**
     * {@inheritDoc}
     */
    public synchronized void run() {
        try {
            while (!stop) {
                try {
                    if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                        LOG.fine("waiting " + gcMinInterval + "ms before garbage collection");
                    }
                    wait(gcMinInterval);
                } catch (InterruptedException woken) {
                    Thread.interrupted();

                    if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                        LOG.log(Level.FINE, "Thread interrupted", woken);
                    }
                }

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

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

                    if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                        LOG.fine("Garbage collection started");
                    }
                    garbageCollect();
                    inconvenienceLevel = 0;
                    gcTime = System.currentTimeMillis() + gcMaxInterval;
                    if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                        LOG.fine("Garbage collection completed");
                    }
                }
            }
        } catch (Throwable all) {
            if (Logging.SHOW_SEVERE && LOG.isLoggable(Level.SEVERE)) {
                LOG.log(Level.SEVERE, "Uncaught Throwable in thread :" + Thread.currentThread().getName(), all);
            }
        } finally {
            gcThread = null;
        }
    }

    private synchronized void rebuildIndex() throws DBException, IOException {

        if (Logging.SHOW_INFO && LOG.isLoggable(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();
                XMLDocument asDoc = (XMLDocument) StructuredDocumentFactory.newStructuredDocument(MimeMediaType.XMLUTF8, is);
                Advertisement adv = AdvertisementFactory.newAdvertisement(asDoc);
                Map<String, String> indexables = getIndexfields(adv.getIndexFields(), asDoc);

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

                if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                    LOG.fine("Restoring index " + keyedIdx + " at " + pos);
                }
                index.addToIndex(keyedIdx, pos);
            } catch (Exception ex) {
                if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                    LOG.log(Level.WARNING, "Exception rebuilding index  at " + pos, ex);
                }
                return true;
            }
            return true;
        }
    }
}