indexcontroller.java

lucene2.2.0版本

        createNewIndexerTask(entry, IndexAction.UPDATE);
    }

    /**
     * @see org.apache.lucene.gdata.server.registry.EntryEventListener#fireInsertEvent(org.apache.lucene.gdata.data.ServerBaseEntry)
     */
    public void fireInsertEvent(final ServerBaseEntry entry) {
        createNewIndexerTask(entry, IndexAction.INSERT);
    }

    /**
     * @see org.apache.lucene.gdata.server.registry.EntryEventListener#fireDeleteEvent(org.apache.lucene.gdata.data.ServerBaseEntry)
     */
    public void fireDeleteEvent(final ServerBaseEntry entry) {
        createNewIndexerTask(entry, IndexAction.DELETE);

    }
    
    /**
     * @see org.apache.lucene.gdata.server.registry.EntryEventListener#fireDeleteAllEntries(org.apache.lucene.gdata.data.ServerBaseFeed)
     */
    public void fireDeleteAllEntries(final ServerBaseFeed feed) {
        createNewDeleteAllEntriesTask(feed);
    }
    
    private void createNewDeleteAllEntriesTask(final ServerBaseFeed feed){
        checkDestroyed();
        checkInitialized();
        if(LOG.isInfoEnabled())
            LOG.info("Deleting all entries for feed dispatch new IndexDocumentBuilder -- "+feed.getId());
        String serviceName = feed.getServiceConfig().getName();
        ServiceIndex bean = this.indexerMap.get(serviceName);
        if (bean == null)
            throw new RuntimeException("no indexer for service " + serviceName
                    + " registered");
        Lock lock = bean.getLock();
        lock.lock();
        try{
            IndexDocumentBuilder<IndexDocument> callable = new IndexFeedDeleteTask(feed.getId());
            sumbitTask(callable,bean.getIndexer());
        }finally{
            lock.unlock();
        }
            
        
    }

    // TODO add test for this method!!
    private void createNewIndexerTask(final ServerBaseEntry entry, final IndexAction action) {
        checkDestroyed();
        checkInitialized();
        String serviceName = entry.getServiceConfig().getName();
        if (LOG.isInfoEnabled())
            LOG.info("New Indexer Task submitted - Action: " + action
                    + " for service: " + serviceName);
        ServiceIndex bean = this.indexerMap.get(serviceName);
        if (bean == null)
            throw new RuntimeException("no indexer for service " + serviceName
                    + " registered");
        /*
         * lock on service to synchronize the event order. This lock has
         * fairness parameter set to true. Grant access to the longest waiting
         * thread. Using fairness is slower but is acceptable in this context
         */
        Lock lock = bean.getLock();
        lock.lock();
        try {
            IndexSchema schema = bean.getSchema();
            boolean commitAfter = bean.incrementActionAndReset(schema.getCommitAfterDocuments());
            IndexDocumentBuilder<IndexDocument> callable = new IndexDocumentBuilderTask<IndexDocument>(
                    entry, bean.getSchema(), action, commitAfter,bean.getOptimize(schema.getOptimizeAfterCommit()));
            sumbitTask(callable,bean.getIndexer());
        } finally {
            /*
             * make sure to unlock
             */
            lock.unlock();
        }

    }

    private void sumbitTask(final Callable<IndexDocument> callable, final GDataIndexer indexer){
        Future<IndexDocument> task = this.taskExecutor.submit(callable);
        try {
            indexer.addIndexableDocumentTask(task);
        } catch (InterruptedException e) {
            throw new GdataIndexerException(
                    "Can not accept any index tasks -- interrupted. ", e);

        }
    }    

    /**
     * @see org.apache.lucene.gdata.search.SearchComponent#getServiceSearcher(org.apache.lucene.gdata.server.registry.ProvidedService)
     */
    public GDataSearcher<String> getServiceSearcher(final ProvidedService service) {
        checkDestroyed();
        checkInitialized();

        /*
         * get and increment. searcher will be decremented if GdataSearcher is
         * closed
         */
        ReferenceCounter<IndexSearcher> searcher;
        synchronized (this) {
            ServiceIndex serviceIndex = this.indexerMap.get(service.getName());
            if(serviceIndex == null)
                throw new RuntimeException("no index for service "+service.getName());
            searcher = serviceIndex.getSearcher();
            searcher.increamentReference();
        }

        return new StandardGdataSearcher(searcher);
    }

    /**
     * @see org.apache.lucene.gdata.search.SearchComponent#destroy()
     */
    public synchronized void destroy() {
        checkDestroyed();
        if(!this.isInitialized.get())
            return;
        this.destroyed.set(true);
        this.isInitialized.set(false);
        LOG.info("Shutting down IndexController -- destroy has been called");
        Set<Entry<String, ServiceIndex>> entrySet = this.indexerMap.entrySet();
        for (Entry<String, ServiceIndex> entry : entrySet) {
            ServiceIndex bean = entry.getValue();
            bean.getSearcher().decrementRef();
            GDataIndexer indexer = bean.getIndexer();
            try {
                indexer.destroy();
            } catch (IOException e) {
                LOG.warn("Can not destroy indexer for service: "
                        + bean.getSchema().getName(), e);
            }
        }
        this.taskExecutor.shutdown();
        this.indexerMap.clear();
    }

    private void checkDestroyed(){
        if (this.destroyed.get())
            throw new IllegalStateException(
                    "IndexController has been destroyed");   
    }
    private void checkInitialized(){
        if(!this.isInitialized.get())
            throw new IllegalStateException(
            "IndexController has not been initialized");
    }   
    
    
    final static class ServiceIndex {
        private AtomicInteger actionCount = new AtomicInteger(0);
        
        private AtomicInteger commitCount = new AtomicInteger(0);
        
        private final Lock lock;

        private final IndexSchema schema;

        private final GDataIndexer indexer;

        private final Directory directory;
        
        private Filter addedDocumentFilter;
        
        private ReferenceCounter<IndexSearcher> searcher;

        // private final Map<String,IndexAction> actionMap;

       

        ServiceIndex(final IndexSchema schema, GDataIndexer indexer,
                Directory directory) {
            this.schema = schema;
            this.indexer = indexer;
            this.lock = new ReentrantLock(true);
            this.directory = directory;
            // this.actionMap = new HashMap<String,IndexAction>(128);
        }

        Lock getLock() {
            return this.lock;
        }

        /**
         * @return Returns the indexer.
         */
        GDataIndexer getIndexer() {
            return this.indexer;
        }

        /**
         * @return Returns the schema.
         */
        IndexSchema getSchema() {
            return this.schema;
        }

        // public void addAction(IndexAction action,ServerBaseEntry entry){
        //            
        // }
        /**
         * Counts how many actions have been executed on this index
         * 
         * @param reset - count mod reset value equals 0 causes a commit
         *            
         * @return <code>true</code> if the count mod reset value equals 0, otherwise
         *         false;
         */
        boolean incrementActionAndReset(int reset) {
            if (this.actionCount.incrementAndGet()%reset == 0) {
                return true;
            }
            return false;
        }

        /**
         * @return Returns the directory.
         */
        public Directory getDirectory() {
            return this.directory;
        }
        /**
         * @return Returns the addedDocumentFilter.
         */
        public Filter getAddedDocumentFilter() {
            return this.addedDocumentFilter;
        }

        /**
         * @param addedDocumentFilter The addedDocumentFilter to set.
         */
        public void setAddedDocumentFilter(Filter addedDocumentFilter) {
            this.addedDocumentFilter = addedDocumentFilter;
        }

        /**
         * @return Returns the searcher.
         */
        public ReferenceCounter<IndexSearcher> getSearcher() {
            return this.searcher;
        }

        /**
         * @param searcher The searcher to set.
         */
        public void setSearcher(ReferenceCounter<IndexSearcher> searcher) {
            this.searcher = searcher;
        }

        /**
         * @return Returns the commitCount.
         */
        public int commitCountIncrement() {
            return this.commitCount.incrementAndGet();
        }
        /**
         * @param reset - the number after how many commits the index should be optimized
         * @return <code>true</code> if and only if the commit count mod reset equals 0, otherwise <code>false</code>.
         */
        public boolean getOptimize(int reset){
            if(this.commitCount.get()%reset == 0){
                return true;
            }
            return false;
        }
    }


}