indexer.java

来自「JXTA&#8482 is a set of open, generalize」· Java 代码 · 共 618 行 · 第 1/2 页

            if (indices != null) {
                Iterator<NameIndexer> i = indices.values().iterator();

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

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

        if (indexables == null) {
            return;
        }
        // FIXME add indexer name to NameIndexer, to optimize this loop
        for (String name : indexables.keySet()) {
            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.fine("looking up NameIndexer : " + name);
            }
            NameIndexer indexer = 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 + indexables.get(name));
            Key indexKey = new Key(indexables.get(name));
            long listPos = writeRecord(listDB, dbKey, pos);

            if (Logging.SHOW_FINER && LOG.isLoggable(Level.FINER)) {
                StringBuilder message = new StringBuilder().append("Adding a reference at position :").append(listPos).append(" to ").append(name).append(" index, Key: ").append(
                        indexables.get(name));
                LOG.finer(message.toString());
            }
            indexer.add(indexKey, listPos);
        }
    }

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

        Collection<String> names;

        if (indexables == null) {
            names = indices.keySet();
        } else {
            names = indexables.keySet();
        }

        Long lpos = pos;

        for (String name : names) {
            NameIndexer indexer = 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 = null;

                if (indexables != null) {
                    dbKey = new Key(name + indexables.get(name));
                }
                Key indexKey = null;

                if (indexables != null) {
                    indexKey = new Key(indexables.get(name));
                }
                synchronized (listDB) {
                    Record record = listDB.readRecord(dbKey);
                    Set<Long> 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);
                        indexer.remove(indexKey);
                    }
                }
            }
        }
    }

    /**
     * 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.
     * @throws IOException if an io error occurs
     * @throws BTreeException if an DB error occurs
     */
    public void purge(List<Long> list) throws IOException, BTreeException {
        IndexQuery iq = new IndexQuery(IndexQuery.ANY, "");
        Collection<String> keys = new ArrayList<String>(indices.keySet());

        for (String objKey : keys) {
            NameIndexer index = indices.get(objKey);
            PurgeCallback pc = new PurgeCallback(listDB, index, objKey, 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.
     * @throws IOException    if an io error occurs
     * @throws BTreeException if an BTree error occurs
     */
    public void purge(long pos) throws IOException, BTreeException {
        purge(Collections.<Long>singletonList(pos));
    }

    private static final class PurgeCallback implements BTreeCallback {

        private final NameIndexer indexer;
        private final List<Long> list;
        private final BTreeFiler listDB;
        private final String indexKey;

        PurgeCallback(BTreeFiler listDB, NameIndexer indexer, String indexKey, List<Long> 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 {
                synchronized (listDB) {
                    Record record = listDB.readRecord(pos);
                    Set<Long> offsets = readRecord(record);

                    boolean 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));
                        }
                    }
                }
            } catch (DBException ignore) {
                if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                    LOG.log(Level.WARNING, "An exception occured", ignore);
                }
            }
            return true;
        }
    }

    private static byte[] toByteArray(Set<Long> offsets) {
        try {
            int size = offsets.size();
            ByteArrayOutputStream bos = new ByteArrayOutputStream((size * 8) + 4);
            DataOutputStream dos = new DataOutputStream(bos);

            dos.writeInt(size);
            for (Long lpos : offsets) {
                dos.writeLong(lpos.longValue());
            }
            dos.close();
            return bos.toByteArray();
        } catch (IOException ie) {
            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.log(Level.WARNING, "Exception during array to byte array conversion", ie);
            }
        }
        return null;
    }

    public static Set<Long> readRecord(Record record) {
        Set<Long> result = new TreeSet<Long>();

        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(ois.readLong());
            }
            ois.close();
        } catch (IOException ie) {
            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.log(Level.WARNING, "Exception while reading Entry", ie);
            }
        }
        return result;
    }

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

        synchronized (listDB) {
            Long lpos = pos;
            Record record = listDB.readRecord(key);
            Set<Long> offsets = readRecord(record);

            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE) && offsets != null) {
                LOG.finer("list.contains " + pos + " : " + offsets.contains(lpos));
            }

            if (offsets != null && !offsets.contains(lpos)) {
                if (Logging.SHOW_FINER && LOG.isLoggable(Level.FINER)) {
                    LOG.finer("Adding a reference to record at :" + lpos);
                    LOG.finer("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 (Logging.SHOW_FINER && LOG.isLoggable(Level.FINER)) {
                LOG.finer("Found " + val.toString() + " at " + pos);
            }
            Record record = null;
            Set<Long> offsets = null;
            boolean result = true;

            try {
                synchronized (listDB) {
                    record = listDB.readRecord(pos);
                    offsets = readRecord(record);
                    if (Logging.SHOW_FINER && LOG.isLoggable(Level.FINER)) {
                        LOG.finer("Found " + offsets.size() + " entries");
                    }
                }

                for (Long lpos : offsets) {
                    result &= callback.indexInfo(val, lpos);
                    if (Logging.SHOW_FINER && LOG.isLoggable(Level.FINER)) {
                        LOG.finer("Callback result : " + result);
                    }
                }
            } catch (DBException ex) {
                if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                    LOG.log(Level.WARNING, "Exception while reading indexed", ex);
                }
                return false;
            }
            return result;
        }
    }
}