indexcontroller.java

lucene2.2.0版本

/**
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.lucene.gdata.search.index;

import java.io.File;
import java.io.IOException;
import java.util.Collection;
import java.util.Map;
import java.util.Set;
import java.util.Map.Entry;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.lucene.gdata.data.ServerBaseEntry;
import org.apache.lucene.gdata.data.ServerBaseFeed;
import org.apache.lucene.gdata.search.GDataSearcher;
import org.apache.lucene.gdata.search.SearchComponent;
import org.apache.lucene.gdata.search.StandardGdataSearcher;
import org.apache.lucene.gdata.search.config.IndexSchema;
import org.apache.lucene.gdata.server.registry.Component;
import org.apache.lucene.gdata.server.registry.ComponentType;
import org.apache.lucene.gdata.server.registry.EntryEventListener;
import org.apache.lucene.gdata.server.registry.GDataServerRegistry;
import org.apache.lucene.gdata.server.registry.ProvidedService;
import org.apache.lucene.gdata.utils.ReferenceCounter;
import org.apache.lucene.index.IndexFileNameFilter;
import org.apache.lucene.search.Filter;
import org.apache.lucene.search.IndexSearcher;
import org.apache.lucene.store.Directory;
import org.apache.lucene.store.FSDirectory;

/**
 * Default implementation of the {@link SearchComponent} interface. All actions
 * on the index will be controlled from this class. Only this class grants read
 * or write actions access to the index.
 * 
 * @author Simon Willnauer
 * 
 */
@Component(componentType = ComponentType.SEARCHCONTROLLER)
public class IndexController implements SearchComponent, IndexEventListener,
        EntryEventListener {
    static final Log LOG = LogFactory.getLog(IndexController.class);

    private final AtomicBoolean isInitialized = new AtomicBoolean(false);

    private final AtomicBoolean destroyed = new AtomicBoolean(false);

    protected Map<String, ServiceIndex> indexerMap;

    private final ExecutorService taskExecutor;

    /**
     * Creates a new IndexController -- call
     * {@link IndexController#initialize()} to set up the controller.
     */
    public IndexController() {
        this.taskExecutor = Executors.newCachedThreadPool();
    }

    /**
     * @see org.apache.lucene.gdata.search.SearchComponent#initialize()
     */
    public synchronized void initialize() {
        if (this.isInitialized.get())
            throw new IllegalStateException(
                    "IndexController is already initialized");
        this.destroyed.set(false);
        /*
         * if this fails the server must not startup --> throw runtime exception
         */
        GDataServerRegistry.getRegistry().registerEntryEventListener(this);

        GDataServerRegistry.getRegistry().registerEntryEventListener(this);
        Collection<ProvidedService> services = GDataServerRegistry
                .getRegistry().getServices();
        this.indexerMap = new ConcurrentHashMap<String, ServiceIndex>(services
                .size());
       
        for (ProvidedService service : services) {
            IndexSchema schema = service.getIndexSchema();
            /*
             * initialize will fail if mandatory values are not set. This is
             * just a
             */
            schema.initialize();
            addIndexSchema(schema);
        }
        this.isInitialized.set(true);
        

    }

    /*
     * add a schema to the index controller and create the indexer. create
     * directories and check out existing indexes
     */
    protected void addIndexSchema(final IndexSchema schema) {
        checkDestroyed();
        if (schema.getName() == null)
            throw new IllegalStateException(
                    "schema has no name -- is not associated with any service");
        if (this.indexerMap.containsKey(schema.getName()))
            throw new IllegalStateException("schema for service "
                    + schema.getName() + " is already registered");
        if (LOG.isInfoEnabled())
            LOG.info("add new IndexSchema for service " + schema.getName()
                    + " -- " + schema);
        try {
            ServiceIndex bean = createIndexer(schema);
            ReferenceCounter<IndexSearcher> searcher = getNewServiceSearcher(bean.getDirectory());
            bean.setSearcher(searcher);
            this.indexerMap.put(schema.getName(), bean);
        } catch (IOException e) {
            LOG.error("Can not create indexer for service " + schema.getName(),
                    e);
            throw new GdataIndexerException(
                    "Can not create indexer for service " + schema.getName(), e);
        }

    }

    protected ServiceIndex createIndexer(final IndexSchema schema) throws IOException {
        GDataIndexer indexer;
        File indexLocation = createIndexLocation(schema.getIndexLocation(),
                schema.getName());
        boolean create = createIndexDirectory(indexLocation);
        Directory dir = FSDirectory.getDirectory(indexLocation, create);
        if (LOG.isInfoEnabled())
            LOG.info("Create new Indexer for IndexSchema: " + schema);
        /*
         * timed or committed indexer?! keep the possibility to let users decide
         * to use scheduled commits
         */
        if (schema.isUseTimedIndexer())
            indexer = GDataIndexer.createTimedGdataIndexer(schema, dir, create,
                    schema.getIndexerIdleTime());
        else
            indexer = GDataIndexer.createGdataIndexer(schema, dir, create);
        indexer.registerIndexEventListener(this);
        return new ServiceIndex(schema, indexer, dir);
    }

    /*
     * if this fails the server must not startup!!
     */
    protected File createIndexLocation(final String path,final  String name) {
        if (path == null || name == null)
            throw new GdataIndexerException(
                    "Path or Name of the index location is not set Path: "
                            + path + " name: " + name);
        /*
         * check if parent e.g. the configured path is a directory
         */
        File parent = new File(path);
        if (!parent.isDirectory())
            throw new IllegalArgumentException(
                    "the given path is not a directory -- " + path);
        /*
         * try to create and throw ex if fail
         */
        if (!parent.exists())
            if (!parent.mkdir())
                throw new RuntimeException("Can not create directory -- "
                        + path);
        /*
         * try to create and throw ex if fail
         */
        File file = new File(parent, name);
        if (file.isFile())
            throw new IllegalArgumentException(
                    "A file with the name"
                            + name
                            + " already exists in "
                            + path
                            + " -- a file of the name of the service must not exist in the index location");

        if (!file.exists()) {
            if (!file.mkdir())
                throw new RuntimeException("Can not create directory -- "
                        + file.getAbsolutePath());
        }
        return file;
    }

    protected boolean createIndexDirectory(final File file) {
        /*
         * use a lucene filename filter to figure out if there is an existing
         * index in the defined directory
         */
        String[] luceneFiles = file.list(new IndexFileNameFilter());
        return !(luceneFiles.length > 0);

    }

    /**
     * @see org.apache.lucene.gdata.search.index.IndexEventListener#commitCallBack(java.lang.String)
     */
    public synchronized void commitCallBack(final String service) {
        checkDestroyed();
        if(LOG.isInfoEnabled())
            LOG.info("CommitCallback triggered - register new searcher for service: "+service);
        /*
         * get the old searcher and replace it if possible.
         */
        ServiceIndex index = this.indexerMap.get(service);
        ReferenceCounter<IndexSearcher> searcher = index.getSearcher();

        try {
            index.setSearcher(getNewServiceSearcher(index.getDirectory()));
        } catch (IOException e) {
            LOG.fatal("Can not create new Searcher -- keep the old one ", e);
            return;
        }
        /*
         * if new searcher if registered decrement old one to get it destroyed if unused
         */
        searcher.decrementRef();
    }
    /*
     * create a new ReferenceCounter for the indexSearcher.
     * The reference is already incremented before returned
     */
    private ReferenceCounter<IndexSearcher> getNewServiceSearcher(final Directory dir)
            throws IOException {
        if(LOG.isInfoEnabled())
            LOG.info("Create new ServiceSearcher");
        IndexSearcher searcher = new IndexSearcher(dir);
        ReferenceCounter<IndexSearcher> holder = new ReferenceCounter<IndexSearcher>(
                searcher) {

            @Override
            protected void close() {
                try {
                    LOG
                            .info("Close IndexSearcher -- Zero references remaining");
                    this.resource.close();
                } catch (IOException e) {
                    LOG.warn("Can not close IndexSearcher -- ", e);
                }
            }

        };
        holder.increamentReference();
        return holder;
    }

    /**
     * @see org.apache.lucene.gdata.server.registry.EntryEventListener#fireUpdateEvent(org.apache.lucene.gdata.data.ServerBaseEntry)
     */
    public void fireUpdateEvent(final ServerBaseEntry entry) {