/*
*
* ====================================================================
*
* Copyright 2004 The Apache Software Foundation
*
* Licensed 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.slide.index.lucene;

import java.io.InputStream;
import java.text.MessageFormat;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashSet;
import java.util.Hashtable;
import java.util.Iterator;
import java.util.Set;
import java.util.StringTokenizer;

import javax.transaction.xa.XAException;
import javax.transaction.xa.XAResource;
import javax.transaction.xa.Xid;

import org.apache.commons.transaction.util.LoggerFacade;
import org.apache.commons.transaction.util.xa.AbstractTransactionalResource;
import org.apache.commons.transaction.util.xa.AbstractXAResource;
import org.apache.commons.transaction.util.xa.TransactionalResource;

import org.apache.slide.common.AbstractServiceBase;
import org.apache.slide.common.NamespaceAccessToken;
import org.apache.slide.common.ServiceAccessException;
import org.apache.slide.common.ServiceConnectionFailedException;
import org.apache.slide.common.ServiceDisconnectionFailedException;
import org.apache.slide.common.ServiceInitializationFailedException;
import org.apache.slide.common.ServiceParameterErrorException;
import org.apache.slide.common.ServiceParameterMissingException;
import org.apache.slide.common.ServiceResetFailedException;
import org.apache.slide.common.Uri;
import org.apache.slide.content.NodeRevisionDescriptor;
import org.apache.slide.content.NodeRevisionNumber;
import org.apache.slide.search.IndexException;
import org.apache.slide.search.basic.IBasicExpressionFactory;
import org.apache.slide.store.IndexStore;
import org.apache.slide.util.logger.Logger;
import org.apache.slide.util.logger.TxLogger;


/**
 * Common base class for lucene bases index stores.
 */
public abstract class AbstractLuceneIndexer extends AbstractServiceBase implements
        IndexStore
{
    private static final String INCLUDE_PARAM = "includes";
    private static final String INDEX_PATH_PARAM = "indexpath";
    private static final String ASYNC_INDEX_PARAM = "asynchron";
    private static final String PRIORITY_PARAM = "priority";
    private static final String OPT_THESHOLD_PARAM = "optimization-threshold";


    /**
     * XAResource implementation to delegate all XAResource requests to.
     */
    private XAResourceImpl xaResource = null;
    
    private boolean started = false;


    protected Index index = null;
    protected IndexConfiguration indexConfiguration = new IndexConfiguration();

    private Collection includes;

    protected AbstractLuceneIndexer() {
        TxLogger txLogger = new TxLogger(getLogger(), LOG_CHANNEL);
        this.xaResource = new XAResourceImpl(txLogger);
    }
    
    public void initialize(NamespaceAccessToken token)
        throws ServiceInitializationFailedException
    {
        debug("initialize");
        this.indexConfiguration.setNamespaceName(token.getName());
    }
    
    public void setParameters(Hashtable parameters)
        throws ServiceParameterErrorException,
        ServiceParameterMissingException
    {
        String indexpath = (String) parameters.get(INDEX_PATH_PARAM);
        if (indexpath == null || indexpath.length() == 0) {
            throw new ServiceParameterMissingException(this, INDEX_PATH_PARAM);
        }
        this.indexConfiguration.setIndexPath(indexpath);

        this.indexConfiguration.setIndexAsynchron(
                "true".equals(parameters.get(ASYNC_INDEX_PARAM)));
        
        String p = (String)parameters.get(PRIORITY_PARAM);
        if (p != null) {
            try {
                this.indexConfiguration.setPriority(Integer.parseInt(p));
            } catch (NumberFormatException e) {
                error("Invalid " + PRIORITY_PARAM + " parameter: {0} ({1})", p, e);
            }
        }
        p = (String)parameters.get(OPT_THESHOLD_PARAM);
        if (p != null) {
            try {
                this.indexConfiguration.setOptimizeThreshold(Integer.parseInt(p));
            } catch (NumberFormatException e) {
                error("Invalid " + OPT_THESHOLD_PARAM + " parameter: {0} ({1})", p, e);
            }
        }
        
        String includes = (String) parameters.get(INCLUDE_PARAM);
        if (includes != null && includes.length() > 0) {
            StringTokenizer tokenizer = new StringTokenizer(includes, ",");
            this.includes = new ArrayList(tokenizer.countTokens());
            while (tokenizer.hasMoreTokens()) {
                String token = tokenizer.nextToken();
                if (!token.endsWith("/")) token = token + "/";
                this.includes.add(token);
            }
        }
    }
    
    protected boolean isIncluded(String uri)
    {
        if (includes == null) {
            return true;
        }
        Iterator iter = includes.iterator();
        while (iter.hasNext()) {
            if (uri.startsWith((String) iter.next())) {
                return true;
            }
        }
        return false;
    }
    
    /**
     * Deletes service underlying data source, if possible (and meaningful).
     * 
     * @exception ServiceResetFailedException
     *               Reset failed
     */
    public void reset() throws ServiceResetFailedException
    {
        debug("reset");
        // TODO should this empty the index?
    }
    
    /**
     * Method getFactory
     * 
     * @return an IBasicExpressionFactory
     *  
     */
    public IBasicExpressionFactory getBasicExpressionFactory()
    {
        return new LuceneExpressionFactory(this.index);
    }

    // -- Service methods ------------------------------------------------------
    /**
     * Connects to the underlying data source (if any is needed).
     * 
     * @exception ServiceConnectionFailedException
     *               Connection failed
     */
    public void connect() throws ServiceConnectionFailedException
    {
        debug("connect");
        this.index.start();
        started = true;
    }

    /**
     * This function tells whether or not the service is connected.
     * 
     * @return boolean true if we are connected
     * @exception ServiceAccessException
     *               Service access error
     */
    public boolean isConnected() throws ServiceAccessException
    {
        return started;
    }
    
    /**
     * Disconnects from the underlying data source.
     * 
     * @exception ServiceDisconnectionFailedException
     *               Disconnection failed
     */
    public void disconnect() throws ServiceDisconnectionFailedException
    {
        debug("disconnect");
        this.index.stop();
        started = false;
    }
    
    // -------------------------------------------------------------------------
    // XAResource interface, all request are deletgated to this.xaResource

    public void commit(Xid xid, boolean onePhase) throws XAException
    {
        this.xaResource.commit(xid, onePhase);
    }

    public void end(Xid xid, int flags) throws XAException
    {
        this.xaResource.end(xid, flags);
    }

    public void forget(Xid xid) throws XAException
    {
        this.xaResource.forget(xid);
    }

    public int getTransactionTimeout() throws XAException
    {
        return this.xaResource.getTransactionTimeout();
    }

    public boolean isSameRM(XAResource xares) throws XAException
    {
        return this.xaResource != null && this.xaResource.isSameRM(xares); // ??
    }

    public int prepare(Xid xid) throws XAException
    {
        return this.xaResource.prepare(xid);
    }

    public Xid[] recover(int flag) throws XAException
    {
        return this.xaResource.recover(flag);
    }

    public void rollback(Xid xid) throws XAException
    {
        this.xaResource.rollback(xid);
    }

    public boolean setTransactionTimeout(int sec) throws XAException
    {
        return this.xaResource.setTransactionTimeout(sec);
    }

    public void start(Xid xid, int flags) throws XAException
    {
        this.xaResource.start(xid, flags);
    }

    //--------------------------------------------------------------------------
    
    protected TransactionalIndexResource getCurrentTxn() {
        return xaResource.getCurrentTxn();
    }
    
    protected TransactionalResource createTransactionResource(Xid xid)
        throws Exception
    {
        return new TransactionalIndexResource(xid, index);
    }
    
    
    private class XAResourceImpl extends AbstractXAResource
    {
    
        private LoggerFacade loggerFacade;
    
        XAResourceImpl(LoggerFacade loggerFacade)
        {
            this.loggerFacade = loggerFacade;
        }
    
        protected TransactionalResource createTransactionResource(Xid xid)
                throws Exception
        {
            return AbstractLuceneIndexer.this.createTransactionResource(xid);
        }
        
        TransactionalIndexResource getCurrentTxn() {
            return (TransactionalIndexResource)getCurrentlyActiveTransactionalResource();
        }
    
        protected LoggerFacade getLoggerFacade()
        {
            return this.loggerFacade;
        }
    
        protected boolean includeBranchInXid()
        {
            return true;
        }
    
        public boolean isSameRM(XAResource xares) throws XAException
        {
            return xares == this;
        }
    
        public Xid[] recover(int flag) throws XAException
        {
            // TODO Auto-generated method stub
            return null;
        }
    
        public int getTransactionTimeout() throws XAException
        {
            return 0;
        }
    
        public boolean setTransactionTimeout(int seconds) throws XAException
        {
            return false;
        }
    }
    
    /**
     * Simple transaction resource that stores index jobs in memory and 
     * schedules them at the index on commit. 
     */
    protected static class TransactionalIndexResource extends
            AbstractTransactionalResource
    {
        private Index idx;
        private Set indexJobs = new HashSet();
        private Set removeJobs = new HashSet();
    
        TransactionalIndexResource(Xid xid, Index index)
        {
            super(xid);
            this.idx = index;
        }
        
        void addIndexJob(Uri uri, NodeRevisionDescriptor descriptor) {
            this.indexJobs.add(this.idx.createIndexJob(uri, descriptor));
        }
        void addIndexJob(Uri uri, NodeRevisionDescriptor descriptor, 
                InputStream content) {
            this.indexJobs.add(this.idx.createIndexJob(uri, descriptor, content));
        }
        void addRemoveJob(Uri uri, NodeRevisionNumber version) {
            Index.IndexJob job = this.idx.createDeleteJob(uri, version);
            this.indexJobs.remove(job);
            this.removeJobs.remove(job); // remove before add because different jobs for same uri have the same key 
            this.removeJobs.add(job);
        }
        void addUpdateJob(Uri uri, 
                NodeRevisionDescriptor descriptor) {
            Index.IndexJob job = this.idx.createIndexJob(uri, descriptor);
            this.removeJobs.remove(job); // remove before add because different jobs for same uri have the same key
            this.removeJobs.add(job);
            this.indexJobs.remove(job);
            this.indexJobs.add(job);
        }
        void addUpdateJob(Uri uri, 
                NodeRevisionDescriptor descriptor, 
                InputStream content) 
        {
            Index.IndexJob job = this.idx.createIndexJob(uri, descriptor, content);
            this.removeJobs.remove(job); // remove before add because different jobs for same uri have the same key
            this.removeJobs.add(job);
            this.indexJobs.remove(job);
            this.indexJobs.add(job);
        }
    
        public void begin() throws XAException
        {
        }
    
        public void commit() throws XAException
        {
            try {
                this.idx.scheduleIndexTransaction(this.removeJobs, this.indexJobs);
            } catch (IndexException e) {
                throw new XAException(e.toString());
            }
        }
    
        public int prepare() throws XAException
        {
            return 0;
        }
    
        public void resume() throws XAException
        {
        }
    
        public void rollback() throws XAException
        {
        }
    
        public void suspend() throws XAException
        {
        }
    }
    
    void info(String msg)
    {
        if (this.getLogger().isEnabled(Logger.INFO)) {
            this.getLogger().log(msg, LOG_CHANNEL, Logger.INFO);
        }
    }
    
    void info(String msg, Object o)
    {
        if (this.getLogger().isEnabled(Logger.INFO)) {
            Object[] args = { o};
            this.getLogger().log(MessageFormat.format(msg, args), LOG_CHANNEL,
                    Logger.INFO);
        }
    }
    
    void debug(String msg)
    {
        if (this.getLogger().isEnabled(Logger.DEBUG)) {
            this.getLogger().log(msg, LOG_CHANNEL, Logger.DEBUG);
        }
    }
    
    void debug(String msg, Object o1, Object o2)
    {
        if (this.getLogger().isEnabled(Logger.DEBUG)) {
            Object[] args = { o1, o2};
            this.getLogger().log(MessageFormat.format(msg, args), LOG_CHANNEL,
                    Logger.DEBUG);
        }
    }
    
    void error(String msg, Object o1)
    {
        if (this.getLogger().isEnabled(Logger.ERROR)) {
            Object[] args = { o1};
            this.getLogger().log(MessageFormat.format(msg, args), LOG_CHANNEL,
                    Logger.ERROR);
        }
    }
    void error(String msg, Object o1, Object o2)
    {
        if (this.getLogger().isEnabled(Logger.ERROR)) {
            Object[] args = { o1, o2};
            this.getLogger().log(MessageFormat.format(msg, args), LOG_CHANNEL,
                    Logger.ERROR);
        }
    }
}
