multiindex.java

来自「jsr170接口的java实现。是个apache的开源项目。」· Java 代码 · 共 1,855 行 · 第 1/5 页

     * @return the index document.
     * @throws RepositoryException if an error occurs while reading from the
     *                             workspace.
     */
    Document createDocument(NodeState node) throws RepositoryException {
        return handler.createDocument(node, nsMappings);
    }

    /**
     * Returns a lucene Document for the Node with <code>id</code>.
     *
     * @param id the id of the node to index.
     * @return the index document.
     * @throws RepositoryException if an error occurs while reading from the
     *                             workspace or if there is no node with
     *                             <code>id</code>.
     */
    Document createDocument(NodeId id) throws RepositoryException {
        try {
            NodeState state = (NodeState) handler.getContext().getItemStateManager().getItemState(id);
            return createDocument(state);
        } catch (NoSuchItemStateException e) {
            throw new RepositoryException("Node " + id + " does not exist", e);
        } catch (ItemStateException e) {
            throw new RepositoryException("Error retrieving node: " + id, e);
        }
    }

    /**
     * Returns <code>true</code> if the redo log contained entries while
     * this index was instantiated; <code>false</code> otherwise.
     * @return <code>true</code> if the redo log contained entries.
     */
    boolean getRedoLogApplied() {
        return redoLogApplied;
    }

    /**
     * Removes the <code>index</code> from the list of active sub indexes. The
     * Index is not acutally deleted right away, but postponed to the transaction
     * commit.
     * <p/>
     * This method does not close the index, but rather expects that the index
     * has already been closed.
     *
     * @param index the index to delete.
     */
    synchronized void deleteIndex(PersistentIndex index) {
        // remove it from the lists if index is registered
        indexes.remove(index);
        indexNames.removeName(index.getName());
        // during recovery it may happen that an index had already been marked
        // deleted, so we need to check if it is already marked deleted.
        if (!deletable.contains(index.getName())) {
            deletable.addName(index.getName());
        }
    }

    /**
     * Flushes this <code>MultiIndex</code>. Persists all pending changes and
     * resets the redo log.
     *
     * @throws IOException if the flush fails.
     */
    synchronized void flush() throws IOException {
        // commit volatile index
        executeAndLog(new Start(Action.INTERNAL_TRANSACTION));
        commitVolatileIndex();

        // commit persistent indexes
        for (int i = indexes.size() - 1; i >= 0; i--) {
            PersistentIndex index = (PersistentIndex) indexes.get(i);
            // only commit indexes we own
            // index merger also places PersistentIndex instances in indexes,
            // but does not make them public by registering the name in indexNames
            if (indexNames.contains(index.getName())) {
                index.commit();
                // check if index still contains documents
                if (index.getNumDocuments() == 0) {
                    executeAndLog(new DeleteIndex(getTransactionId(), index.getName()));
                }
            }
        }
        executeAndLog(new Commit(getTransactionId()));

        indexNames.write(indexDir);

        // reset redo log
        redoLog.clear();

        lastFlushTime = System.currentTimeMillis();

        // delete obsolete indexes
        attemptDelete();
    }

    //-------------------------< internal >-------------------------------------

    /**
     * Resets the volatile index to a new instance.
     */
    private void resetVolatileIndex() throws IOException {
        volatileIndex = new VolatileIndex(
                handler.getTextAnalyzer(), indexingQueue);
        volatileIndex.setUseCompoundFile(handler.getUseCompoundFile());
        volatileIndex.setMaxFieldLength(handler.getMaxFieldLength());
        volatileIndex.setBufferSize(handler.getBufferSize());
    }

    /**
     * Returns the current transaction id.
     *
     * @return the current transaction id.
     */
    private long getTransactionId() {
        return currentTransactionId;
    }

    /**
     * Executes action <code>a</code> and appends the action to the redo log if
     * successful.
     *
     * @param a the <code>Action</code> to execute.
     * @return the executed action.
     * @throws IOException         if an error occurs while executing the action
     *                             or appending the action to the redo log.
     */
    private Action executeAndLog(Action a)
            throws IOException {
        a.execute(this);
        redoLog.append(a);
        // please note that flushing the redo log is only required on
        // commit, but we also want to keep track of new indexes for sure.
        // otherwise it might happen that unused index folders are orphaned
        // after a crash.
        if (a.getType() == Action.TYPE_COMMIT || a.getType() == Action.TYPE_ADD_INDEX) {
            redoLog.flush();
            // also flush indexing queue
            indexingQueue.commit();
        }
        return a;
    }

    /**
     * Checks if it is needed to commit the volatile index according to {@link
     * SearchIndex#getMinMergeDocs()}.
     *
     * @return <code>true</code> if the volatile index has been committed,
     *         <code>false</code> otherwise.
     * @throws IOException if an error occurs while committing the volatile
     *                     index.
     */
    private boolean checkVolatileCommit() throws IOException {
        if (volatileIndex.getNumDocuments() >= handler.getMinMergeDocs()) {
            commitVolatileIndex();
            return true;
        }
        return false;
    }

    /**
     * Commits the volatile index to a persistent index. The new persistent
     * index is added to the list of indexes but not written to disk. When this
     * method returns a new volatile index has been created.
     *
     * @throws IOException if an error occurs while writing the volatile index
     *                     to disk.
     */
    private void commitVolatileIndex() throws IOException {

        // check if volatile index contains documents at all
        if (volatileIndex.getNumDocuments() > 0) {

            long time = System.currentTimeMillis();
            // create index
            CreateIndex create = new CreateIndex(getTransactionId(), null, true);
            executeAndLog(create);

            // commit volatile index
            executeAndLog(new VolatileCommit(getTransactionId(), create.getIndexName()));

            // add new index
            AddIndex add = new AddIndex(getTransactionId(), create.getIndexName());
            executeAndLog(add);

            // create new volatile index
            resetVolatileIndex();

            time = System.currentTimeMillis() - time;
            log.debug("Committed in-memory index in " + time + "ms.");
        }
    }

    /**
     * Recursively creates an index starting with the NodeState
     * <code>node</code>.
     *
     * @param node     the current NodeState.
     * @param stateMgr the shared item state manager.
     * @throws IOException         if an error occurs while writing to the
     *                             index.
     * @throws ItemStateException  if an node state cannot be found.
     * @throws RepositoryException if any other error occurs
     */
    private void createIndex(NodeState node, ItemStateManager stateMgr)
            throws IOException, ItemStateException, RepositoryException {
        NodeId id = node.getNodeId();
        if (excludedIDs.contains(id)) {
            return;
        }
        executeAndLog(new AddNode(getTransactionId(), id.getUUID()));
        checkVolatileCommit();
        List children = node.getChildNodeEntries();
        for (Iterator it = children.iterator(); it.hasNext();) {
            NodeState.ChildNodeEntry child = (NodeState.ChildNodeEntry) it.next();
            NodeState childState = (NodeState) stateMgr.getItemState(child.getId());
            createIndex(childState, stateMgr);
        }
    }

    /**
     * Attempts to delete all files recorded in {@link #deletable}.
     */
    private void attemptDelete() {
        for (int i = deletable.size() - 1; i >= 0; i--) {
            String indexName = deletable.getName(i);
            File dir = new File(indexDir, indexName);
            if (deleteIndex(dir)) {
                deletable.removeName(i);
            } else {
                log.info("Unable to delete obsolete index: " + indexName);
            }
        }
        try {
            deletable.write(indexDir);
        } catch (IOException e) {
            log.warn("Exception while writing deletable indexes: " + e);
        }
    }

    /**
     * Deletes the index <code>directory</code>.
     *
     * @param directory the index directory to delete.
     * @return <code>true</code> if the delete was successful,
     *         <code>false</code> otherwise.
     */
    private boolean deleteIndex(File directory) {
        // trivial if it does not exist anymore
        if (!directory.exists()) {
            return true;
        }
        // delete files first
        File[] files = directory.listFiles();
        for (int i = 0; i < files.length; i++) {
            if (!files[i].delete()) {
                return false;
            }
        }
        // now delete directory itself
        return directory.delete();
    }

    /**
     * Returns an new index folder which is empty.
     *
     * @return the new index folder.
     * @throws IOException if the folder cannot be created.
     */
    private File newIndexFolder() throws IOException {
        // create new index folder. make sure it does not exist
        File sub;
        do {
            sub = new File(indexDir, indexNames.newName());
        } while (sub.exists());

        if (!sub.mkdir()) {
            throw new IOException("Unable to create directory: " + sub.getAbsolutePath());
        }
        return sub;
    }

    /**
     * Checks the duration between the last commit to this index and the
     * current time and flushes the index (if there are changes at all)
     * if the duration (idle time) is more than {@link SearchIndex#getVolatileIdleTime()}
     * seconds.
     */
    private synchronized void checkFlush() {
        long idleTime = System.currentTimeMillis() - lastFlushTime;
        // do not flush if volatileIdleTime is zero or negative
        if (handler.getVolatileIdleTime() > 0
                && idleTime > handler.getVolatileIdleTime() * 1000) {
            try {
                if (redoLog.hasEntries()) {
                    log.debug("Flushing index after being idle for "
                            + idleTime + " ms.");
                    synchronized (updateMonitor) {
                        updateInProgress = true;
                    }
                    try {
                        flush();
                    } finally {
                        synchronized (updateMonitor) {
                            updateInProgress = false;
                            updateMonitor.notifyAll();
                            if (multiReader != null) {
                                multiReader.close();
                                multiReader = null;
                            }
                        }
                    }
                }
            } catch (IOException e) {
                log.error("Unable to commit volatile index", e);
            }
        }
    }

    /**
     * Checks the indexing queue for finished text extrator jobs and
     * updates the index accordingly if there are any new ones.
     */
    private synchronized void checkIndexingQueue() {
        Document[] docs = indexingQueue.getFinishedDocuments();
        Map finished = new HashMap();
        for (int i = 0; i < docs.length; i++) {
            String uuid = docs[i].get(FieldNames.UUID);
            finished.put(UUID.fromString(uuid), docs[i]);
        }

        // now update index with the remaining ones if there are any
        if (!finished.isEmpty()) {
            log.debug("updating index with {} nodes from indexing queue.",
                    new Long(finished.size()));

            // remove documents from the queue
            for (Iterator it = finished.keySet().iterator(); it.hasNext(); ) {
                try {
                    indexingQueue.removeDocument(it.next().toString());
                } catch (IOException e) {
                    log.error("Failed to remove node from indexing queue", e);
                }
            }

            try {
                update(finished.keySet().iterator(),
                        finished.values().iterator());
            } catch (IOException e) {
                // update failed
                log.warn("Failed to update index with deferred text extraction", e);
            }
        }
    }

    //------------------------< Actions >---------------------------------------

    /**
     * Defines an action on an <code>MultiIndex</code>.
     */
    public abstract static class Action {

        /**
         * Action identifier in redo log for transaction start action.
         */
        static final String START = "STR";

        /**
         * Action type for start action.
         */
        public static final int TYPE_START = 0;