indexer.java

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

                if (LOG.isEnabledFor(Level.DEBUG)) {
                    LOG.debug("Searching all indexes");
                }
                while (i.hasNext()) {
                    NameIndexer index = (NameIndexer) i.next();
                    index.query(query, new SearchCallback(listDB, callback));
                }
                return;
            }
        } else {
            NameIndexer indexer = (NameIndexer) indices.get(name);
            if (indexer == null) {
                return;
            }
            if (LOG.isEnabledFor(Level.DEBUG)) {
                LOG.debug("Searching Index : "+ name);
            }
            indexer.query(query, cb);
        }
    }

    public void addToIndex(Map indexables, long pos)
    throws IOException, DBException {

        if (indexables == null) {
            return;
        }

        Iterator ni = indexables.keySet().iterator();
        while (ni.hasNext()) {
            String name = (String) ni.next();
            if (LOG.isEnabledFor(Level.DEBUG)) {
                LOG.debug("looking up NameIndexer : " + name);
            }
            NameIndexer indexer = (NameIndexer) indices.get(name);
            if (indexer == null) {
                indexer = new NameIndexer();
                // location should be the same as in setLocation above
                indexer.setLocation(dir, file + "-" + name);
                indexer.setSync(sync);
                if (!indexer.open()) {
                    indexer.create();
                    indexer.open();
                }
                indices.put(name, indexer);
            }

            //we need to make sure that the db key is unique from the
            //the index key to avoid value collision
            Key dbKey = new Key(name + (String) indexables.get(name));
            Key indexKey = new Key((String) indexables.get(name));
            long listPos = writeRecord(listDB, dbKey, pos);
            if (LOG.isEnabledFor(Level.DEBUG)) {
                LOG.debug("Adding a reference at position :" + listPos +
                          " to "+ name + " index, Key: " + (String) indexables.get(name));
            }
            indexer.add(indexKey, listPos);
        }
    }

    public void removeFromIndex(Map indexables, long pos)
    throws DBException {

        Iterator ni;
        if (indexables == null) {
            ni = indices.keySet().iterator();
        } else {
            ni = indexables.keySet().iterator();
        }

        Long lpos = new Long(pos);
        while (ni.hasNext()) {
            String name = (String) ni.next();
            NameIndexer indexer = (NameIndexer) indices.get(name);
            if (indexer != null) {
                //we need to make sure that the db key is unique from the
                //the index key to avoid value collision
                Key dbKey = new Key(name + (String) indexables.get(name));
                Key indexKey = new Key((String) indexables.get(name));
                synchronized (listDB) {
                    Record record = listDB.readRecord(dbKey);
                    Set offsets = readRecord(record);
                    if (!offsets.isEmpty()) {
                        if (offsets.contains(lpos)) {
                            offsets.remove(lpos);
                            Value recordValue = new Value(toByteArray(offsets));
                            listDB.writeRecord(dbKey, recordValue);
                        }
                        if (offsets.isEmpty()) {
                            // only we can proceed to remove the entry from the index
                            listDB.deleteRecord(dbKey);
                            indexer.remove(indexKey);
                        }
                    } else {
                        // empty record purge it
                        listDB.deleteRecord(dbKey);
                    }
                }
            }
        }
    }

    /**
     * purge all index entries pointing to a certain record.
     *
     * @param list List of Long position(s) at which the record to be purged is
     * located in the main database.
     */
    public void purge(List list)
    throws IOException, BTreeException {

        IndexQuery iq = new IndexQuery(IndexQuery.ANY, "");
        Set keys = indices.keySet();
        Object[] objKeys = keys.toArray();
        for(int i=0; i<objKeys.length; i++ ){
            NameIndexer index = (NameIndexer) indices.get(objKeys[i]);
            PurgeCallback pc = new PurgeCallback(listDB, index, (String)objKeys[i], list);
            index.query(iq, pc);
        }
    }

    /**
     * purge all index entries pointing to a certain record.
     *
     * @param pos the position at which the record to be purged is
     * located in the main database.
     */
    public void purge(long pos)
    throws IOException, BTreeException {

        IndexQuery iq = new IndexQuery(IndexQuery.ANY, "");
        Set keys = indices.keySet();
        Object[] objKeys = keys.toArray();
        for(int i=0; i<objKeys.length; i++ ){
            NameIndexer index = (NameIndexer) indices.get(objKeys[i]);
            PurgeCallback pc = new PurgeCallback(listDB, 
                                                 index,
                                                 (String)objKeys[i], 
                                                 Collections.singletonList(new Long(pos)));
            index.query(iq, pc);
        }
    }

    private static final class PurgeCallback implements BTreeCallback {

        private NameIndexer indexer = null;
        private List list;
        private BTreeFiler listDB = null;
        private String indexKey = null;
        
        PurgeCallback(BTreeFiler listDB, NameIndexer indexer, String indexKey, List list) {
            this.listDB = listDB;
            this.indexer = indexer;
            this.indexKey = indexKey;
            this.list = list;
        }

        /**
         *  {@inheritDoc}
         **/
        public boolean indexInfo(Value val, long pos) {
            // Read record to determine whether there's a refrence to pos
            try {
                boolean changed;
                synchronized (listDB) {
                    Record record = listDB.readRecord(pos);
                    Set offsets = readRecord(record);
                    changed = offsets.removeAll(list);
                    if (changed) {
                        if (!offsets.isEmpty()) {
                            Value recordValue = new Value(toByteArray(offsets));
                            listDB.writeRecord(pos, recordValue);
                        } else {
                            listDB.deleteRecord(new Key(indexKey + val));
                            indexer.remove(new Key(val));
                        }
                    } else {
                        // not a match continue callback
                        return true;
                    }
                }
            } catch (DBException ignore) {
                if (LOG.isEnabledFor(Level.DEBUG)) {
                    LOG.debug("An exception occured", ignore);
                }
            }
            return true;
        }
    }

    private static byte[] toByteArray(Set offsets) {
        try {
            int size = offsets.size();
            ByteArrayOutputStream bos= new ByteArrayOutputStream((size * 8 ) + 4);
            DataOutputStream dos = new DataOutputStream(bos);
            dos.writeInt(size);
            Iterator oi = offsets.iterator();
            while (oi.hasNext()) {
                Long lpos = (Long) oi.next();
                dos.writeLong(lpos.longValue());
            }
            dos.close();
            return bos.toByteArray();
        } catch (IOException ie) {
            if (LOG.isEnabledFor(Level.WARN)) {
                LOG.warn("Exception during array to byte array conversion", ie);
            }
        }
        return null;
    }

    public static Set readRecord(Record record) {
        Set result = new TreeSet();
        if (record == null) {
            return result;
        }
        InputStream is = record.getValue().getInputStream();
        try {
            DataInputStream ois = new DataInputStream(is);
            int size = ois.readInt();
            for (int i=0; i < size; i++) {
                result.add(new Long(ois.readLong()));
            }
            ois.close();
        } catch (IOException ie) {
            if (LOG.isEnabledFor(Level.WARN)) {
                LOG.warn("Exception while reading Entry", ie);
            }
        }
        return result;
    }

    private static long writeRecord(BTreeFiler listDB, Key key, long pos)
    throws DBException, IOException {

        synchronized (listDB) {
            Long lpos = new Long(pos);
            Record record = listDB.readRecord(key);
            Set offsets = readRecord(record);
            if (LOG.isEnabledFor(Level.DEBUG) && offsets != null) {
                LOG.debug("list.contains " + pos + " : " + offsets.contains(lpos));
            }

            if (!offsets.contains(lpos)) {
                if (LOG.isEnabledFor(Level.DEBUG)) {
                    LOG.debug("Adding a reference to record at :" + lpos);
                    LOG.debug("Writing :" + offsets.size() + " references");
                }
                offsets.add(lpos);
            }
            Value recordValue = new Value(toByteArray(offsets));
            return listDB.writeRecord(key, recordValue);
        }
    }

    public static final class SearchCallback implements BTreeCallback {

        private BTreeCallback callback = null;
        private BTreeFiler listDB = null;

        public SearchCallback (BTreeFiler listDB, BTreeCallback callback) {
            this.listDB = listDB;
            this.callback = callback;
        }

        /**
         *  {@inheritDoc}
         **/
        public boolean indexInfo(Value val, long pos) {
            if (LOG.isEnabledFor(Level.DEBUG)) {
                LOG.debug("Found " + val.toString() + " at " + pos);
            }
            Record record = null;
            Set offsets = null;
            boolean result = true;
            try {
                synchronized (listDB) {
                    record = listDB.readRecord(pos);
                    offsets = readRecord(record);
                    if (LOG.isEnabledFor(Level.DEBUG)) {
                        LOG.debug("Found " + offsets.size() + " entries");
                    }
                }

                Iterator oi = offsets.iterator();
                while (oi.hasNext()) {
                    Long lpos = (Long) oi.next();
                    result &= callback.indexInfo(val, lpos.longValue());
                    if (LOG.isEnabledFor(Level.DEBUG)) {
                        LOG.debug("Callback result : "+result);
                    }
                }
            } catch (DBException ex) {
                if (LOG.isEnabledFor(Level.WARN)) {
                    LOG.warn("Exception while reading indexed", ex);
                }
                return false;
            }
            return result;
        }
    }
}